"""Governed continuation executor for exact existing task branches.

The executor composes the existing continuation lease and durable Job store.
All Git and Gitea side effects live behind injected narrow interfaces so the
state machine can prove Authority, Git-network, API-branch, and commit-object
agreement before it reports completion.
"""

from __future__ import annotations

import base64
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import hashlib
import json
import os
from pathlib import Path, PurePosixPath
import re
import subprocess
from typing import Callable, Mapping, Protocol
from urllib.error import HTTPError, URLError
from urllib.parse import quote, urlsplit
from urllib.request import Request, urlopen

from aios_authority_git_executor import AuthorityGitExecutor
from aios_continuation_lease import WriterLeaseRequestV2
from aios_gitops_job_controller import GitOpsJobController, JobState
from aios_gitops_reconciliation_core import ReconciliationAction, RemoteObservation, decide_reconciliation
from authorized_task_writer_provisioning import validate_writer_lease_request_v2


OID_RE = re.compile(r"^[0-9a-f]{40}$")
SECRET_PATTERNS = (
    re.compile(rb"gh[pousr]_[A-Za-z0-9]{20,}"),
    re.compile(rb"(?i)(?:token|password|secret|api[_-]?key)\s*[:=]\s*[\"'][^\"'\s]{12,}[\"']"),
    re.compile(rb"-----BEGIN " + rb"(?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
)


class ContinuationExecutionError(RuntimeError):
    """Stable fail-closed continuation execution error."""


class DetachedWorkspace(Protocol):
    def is_detached(self) -> bool: ...
    def head_oid(self) -> str: ...
    def commit_files(self, files: Mapping[str, bytes], message: str) -> str: ...


class AuthorityPort(Protocol):
    def branch_exists(self, branch: str) -> bool: ...
    def branch_head(self, branch: str) -> str: ...
    def is_ancestor(self, older: str, newer: str) -> bool: ...
    def create_task_branch_cas(self, branch: str, expected_base: str, new: str) -> str: ...
    def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str: ...


class RemotePort(Protocol):
    def repository_identity(self) -> str: ...
    def push_new_exact(self, branch: str, new: str) -> None: ...
    def push_exact(self, branch: str, expected: str, new: str) -> None: ...
    def read_git_ref(self, branch: str) -> str | None: ...
    def read_api_branch(self, branch: str) -> str | None: ...
    def commit_visible(self, oid: str) -> bool: ...


class CredentialProjection(Protocol):
    def project(self) -> None: ...
    def revoke(self) -> None: ...


def _run(argv: list[str], *, cwd: Path | None = None, environment: Mapping[str, str] | None = None) -> str:
    result = subprocess.run(
        argv,
        cwd=cwd,
        env=None if environment is None else dict(environment),
        text=True,
        capture_output=True,
        check=False,
    )
    if result.returncode != 0:
        raise ContinuationExecutionError("SUBPROCESS_FAILED")
    return result.stdout.strip()


@dataclass(slots=True)
class AuthorityGitAdapter:
    """Branch-name adapter over the existing narrow AuthorityGitExecutor."""

    executor: AuthorityGitExecutor

    @staticmethod
    def _ref(branch: str) -> str:
        return f"refs/heads/{branch}"

    def branch_exists(self, branch: str) -> bool:
        return self.executor.read_ref(self._ref(branch)) is not None

    def branch_head(self, branch: str) -> str:
        value = self.executor.read_ref(self._ref(branch))
        if value is None:
            raise ContinuationExecutionError("EXISTING_BRANCH_MISSING")
        return value

    def is_ancestor(self, older: str, newer: str) -> bool:
        return self.executor.ancestry_check(older, newer)

    def create_task_branch_cas(self, branch: str, expected_base: str, new: str) -> str:
        return self.executor.create_task_branch_CAS(self._ref(branch), expected_base, new)

    def compare_and_swap_branch(self, branch: str, expected: str, new: str) -> str:
        return self.executor.advance_task_branch_CAS(self._ref(branch), expected, new)


@dataclass(slots=True)
class DetachedGitWorkspace:
    path: Path

    def __post_init__(self) -> None:
        if not self.path.is_absolute() or self.path.is_symlink() or not self.path.is_dir():
            raise ContinuationExecutionError("WORKTREE_PATH_INVALID")
        top = Path(_run(["git", "rev-parse", "--show-toplevel"], cwd=self.path)).resolve()
        if top != self.path.resolve():
            raise ContinuationExecutionError("WORKTREE_PATH_INVALID")

    def is_detached(self) -> bool:
        result = subprocess.run(
            ["git", "symbolic-ref", "-q", "HEAD"],
            cwd=self.path,
            capture_output=True,
            check=False,
        )
        return result.returncode == 1

    def head_oid(self) -> str:
        return _run(["git", "rev-parse", "HEAD"], cwd=self.path)

    def _dirty_paths(self) -> set[str]:
        command = subprocess.run(
            ["git", "status", "--porcelain=v1", "--untracked-files=all"],
            cwd=self.path,
            text=True,
            capture_output=True,
            check=False,
        )
        if command.returncode != 0:
            raise ContinuationExecutionError("SUBPROCESS_FAILED")
        output = command.stdout
        result: set[str] = set()
        for line in output.splitlines():
            path = line[3:]
            if " -> " in path:
                path = path.split(" -> ", 1)[1]
            result.add(path)
        return result

    def materialize_files(self, files: Mapping[str, bytes]) -> None:
        dirty = self._dirty_paths()
        expected_paths = set(files)
        if dirty and dirty != expected_paths:
            raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
        for relative, content in files.items():
            target = (self.path / relative).resolve()
            try:
                target.relative_to(self.path.resolve())
            except ValueError as exc:
                raise ContinuationExecutionError("COMMIT_PATH_REJECTED") from exc
            if target.is_symlink():
                raise ContinuationExecutionError("COMMIT_PATH_REJECTED")
            target.parent.mkdir(parents=True, exist_ok=True)
            target.write_bytes(content)

    def commit_files(self, files: Mapping[str, bytes], message: str) -> str:
        expected_paths = set(files)
        if self._dirty_paths() != expected_paths:
            raise ContinuationExecutionError("WORKTREE_DIRTY_SCOPE_MISMATCH")
        for relative, content in files.items():
            target = (self.path / relative).resolve()
            try:
                target.relative_to(self.path.resolve())
            except ValueError as exc:
                raise ContinuationExecutionError("COMMIT_PATH_REJECTED") from exc
            if not target.is_file() or target.is_symlink() or target.read_bytes() != content:
                raise ContinuationExecutionError("AUTHORING_PAYLOAD_MISMATCH")
        _run(["git", "add", "--", *sorted(expected_paths)], cwd=self.path)
        staged = set(_run(["git", "diff", "--cached", "--name-only"], cwd=self.path).splitlines())
        if staged != expected_paths:
            raise ContinuationExecutionError("STAGED_SCOPE_MISMATCH")
        environment = os.environ.copy()
        environment.update(
            {
                "GIT_AUTHOR_NAME": "AIOS Governed Continuation",
                "GIT_AUTHOR_EMAIL": "aios-governed-continuation@localhost",
                "GIT_COMMITTER_NAME": "AIOS Governed Continuation",
                "GIT_COMMITTER_EMAIL": "aios-governed-continuation@localhost",
            }
        )
        _run(["git", "commit", "--no-gpg-sign", "-m", message, "--", *sorted(expected_paths)], cwd=self.path, environment=environment)
        if not self.is_detached():
            raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
        return self.head_oid()


@dataclass(slots=True)
class ServerSideCredentialProjection:
    ssh_key_path: Path
    api_token_path: Path
    active: bool = False

    def project(self) -> None:
        for path in (self.ssh_key_path, self.api_token_path):
            if path.is_symlink() or not path.is_file() or path.stat().st_size < 32:
                raise ContinuationExecutionError("CREDENTIAL_PROJECTION_MISSING")
        self.active = True

    def revoke(self) -> None:
        self.active = False


@dataclass(slots=True)
class GiteaContinuationRemote:
    workspace: Path
    credential: ServerSideCredentialProjection
    remote_name: str = "gitea"
    repository: str = "aios/aios"
    api_base: str = "https://g.cnnetmkt.top/git/api/v1"
    expected_host: str = "g.cnnetmkt.top"

    @staticmethod
    def validate_remote_url(value: str, expected_host: str = "g.cnnetmkt.top") -> None:
        if value.startswith(("/", "file:")) or value.endswith("/data/aios-gitops/repo.git"):
            raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
        if "://" in value:
            parsed = urlsplit(value)
            host, path = parsed.hostname, parsed.path
        else:
            location = value.split("@", 1)[-1]
            host, separator, path = location.partition(":")
            if not separator:
                raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")
            path = "/" + path
        if host != expected_host or path.rstrip("/") not in {"/aios/aios.git", "/git/aios/aios.git"}:
            raise ContinuationExecutionError("GITEA_REMOTE_IDENTITY_INVALID")

    def repository_identity(self) -> str:
        url = _run(["git", "remote", "get-url", self.remote_name], cwd=self.workspace)
        self.validate_remote_url(url, self.expected_host)
        if self.repository != "aios/aios":
            raise ContinuationExecutionError("REMOTE_REPOSITORY_IDENTITY_MISMATCH")
        return self.repository

    def _require_projected(self) -> None:
        if not self.credential.active:
            raise ContinuationExecutionError("CREDENTIAL_PROJECTION_MISSING")

    def read_git_ref(self, branch: str) -> str | None:
        self._require_projected()
        output = _run(["git", "ls-remote", self.remote_name, f"refs/heads/{branch}"], cwd=self.workspace)
        if not output:
            return None
        oid, ref = output.split(maxsplit=1)
        if ref != f"refs/heads/{branch}" or not OID_RE.fullmatch(oid):
            raise ContinuationExecutionError("GITEA_GIT_READBACK_INVALID")
        return oid

    def push_new_exact(self, branch: str, new: str) -> None:
        self._require_projected()
        self.repository_identity()
        if self.read_git_ref(branch) is not None:
            raise ContinuationExecutionError("GITEA_TARGET_BRANCH_ALREADY_EXISTS")
        _run(["git", "push", self.remote_name, f"{new}:refs/heads/{branch}"], cwd=self.workspace)
        if self.read_git_ref(branch) != new:
            raise ContinuationExecutionError("GITEA_CREATE_READBACK_MISMATCH")

    def push_exact(self, branch: str, expected: str, new: str) -> None:
        self._require_projected()
        self.repository_identity()
        if self.read_git_ref(branch) != expected:
            raise ContinuationExecutionError("GITEA_BRANCH_EXPECTATION_MISMATCH")
        _run(["git", "push", self.remote_name, f"{new}:refs/heads/{branch}"], cwd=self.workspace)

    def _api(self, path: str) -> object:
        self._require_projected()
        token = self.credential.api_token_path.read_text(encoding="utf-8").strip()
        request = Request(
            self.api_base.rstrip("/") + path,
            headers={"Accept": "application/json", "Authorization": f"token {token}"},
        )
        try:
            with urlopen(request, timeout=20) as response:
                return json.loads(response.read().decode("utf-8"))
        except HTTPError as exc:
            if exc.code == 404:
                return None
            raise ContinuationExecutionError(f"GITEA_API_HTTP_{exc.code}") from exc
        except (URLError, TimeoutError, OSError, json.JSONDecodeError) as exc:
            raise ContinuationExecutionError("GITEA_API_UNAVAILABLE") from exc

    def read_api_branch(self, branch: str) -> str | None:
        owner, name = self.repository.split("/", 1)
        page = 1
        while page <= 100:
            value = self._api(f"/repos/{quote(owner)}/{quote(name)}/branches?page={page}&limit=50")
            if value is None:
                return None
            if not isinstance(value, list):
                raise ContinuationExecutionError("GITEA_API_BRANCH_COLLECTION_INVALID")
            for item in value:
                if not isinstance(item, dict) or item.get("name") != branch:
                    continue
                oid = item.get("commit", {}).get("id")
                if not isinstance(oid, str) or not OID_RE.fullmatch(oid):
                    raise ContinuationExecutionError("GITEA_API_BRANCH_INVALID")
                return oid
            if len(value) < 50:
                return None
            page += 1
        raise ContinuationExecutionError("GITEA_API_BRANCH_PAGINATION_LIMIT")

    def commit_visible(self, oid: str) -> bool:
        owner, name = self.repository.split("/", 1)
        value = self._api(f"/repos/{quote(owner)}/{quote(name)}/git/commits/{oid}")
        if not isinstance(value, dict):
            return False
        observed = value.get("sha") or value.get("id")
        return observed == oid


@dataclass(frozen=True, slots=True)
class ContinuationPermissions:
    create_target_branch: bool = False
    force_push: bool = False
    tag_write: bool = False
    baseline_write: bool = False
    production_write: bool = False
    main_write: bool = False

    def validate(self) -> None:
        if any(asdict(self).values()):
            raise ContinuationExecutionError("MUTATION_PERMISSION_DENIED")


@dataclass(frozen=True, slots=True)
class ContinuationRuntimeBinding:
    request: WriterLeaseRequestV2
    job_id: str
    workspace_path: Path
    lease_not_before: datetime
    lease_expires_at: datetime
    original_receipt_valid: bool
    original_grant_consumed: bool
    permissions: ContinuationPermissions


@dataclass(frozen=True, slots=True)
class ContinuationCommit:
    idempotency_key: str
    files: Mapping[str, bytes]
    message: str

    def payload(self) -> bytes:
        if not self.idempotency_key or not self.message.strip() or not self.files:
            raise ContinuationExecutionError("COMMIT_REQUEST_INVALID")
        normalized: dict[str, str] = {}
        for path, content in self.files.items():
            pure = PurePosixPath(path)
            if (
                not isinstance(path, str)
                or pure.is_absolute()
                or ".." in pure.parts
                or path.startswith((".git/", "baseline/", "production/"))
                or not isinstance(content, bytes)
            ):
                raise ContinuationExecutionError("COMMIT_PATH_REJECTED")
            if any(pattern.search(content) for pattern in SECRET_PATTERNS):
                raise ContinuationExecutionError("SECRET_LIKE_PAYLOAD_REJECTED")
            normalized[path] = base64.b64encode(content).decode("ascii")
        return json.dumps(
            {"files_base64": normalized, "message": self.message},
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")


class ContinuationExecutor:
    def __init__(
        self,
        controller: GitOpsJobController,
        authority: AuthorityPort,
        workspace: DetachedWorkspace,
        remote: RemotePort,
        credential: CredentialProjection,
        *,
        now: Callable[[], datetime] = lambda: datetime.now(timezone.utc),
    ) -> None:
        self.controller = controller
        self.authority = authority
        self.workspace = workspace
        self.remote = remote
        self.credential = credential
        self.now = now
        self._observed_generations: dict[str, int] = {}

    def observe_generation(self, branch: str, generation: int) -> None:
        self._observed_generations[branch] = max(generation, self._observed_generations.get(branch, 0))

    def _max_generation(self, branch: str, *, exclude_job_id: str | None = None) -> int:
        observed = self._observed_generations.get(branch, 0)
        with self.controller.store._connect() as connection:
            if exclude_job_id is None:
                row = connection.execute(
                    "SELECT MAX(generation) AS generation FROM branch_leases WHERE branch=?",
                    (branch,),
                ).fetchone()
            else:
                row = connection.execute(
                    "SELECT MAX(generation) AS generation FROM branch_leases "
                    "WHERE branch=? AND job_id!=?",
                    (branch, exclude_job_id),
                ).fetchone()
        durable = int(row["generation"] or 0)
        return max(observed, durable)

    @staticmethod
    def _request_mapping(request: WriterLeaseRequestV2) -> dict[str, object]:
        return {
            "schema": WriterLeaseRequestV2.schema,
            "task_id": request.task_id,
            "job_id": request.job_id,
            "repository": request.repository,
            "branch_mode": request.branch_mode,
            "base_oid": request.base_oid,
            "expected_branch_oid": request.expected_branch_oid,
            "target_branch": request.target_branch,
            "generation": request.generation,
            "lease_seconds": request.lease_seconds,
        }

    def _validate_preflight(self, binding: ContinuationRuntimeBinding, *, job_id: str | None = None) -> None:
        request = binding.request
        request_mapping = self._request_mapping(request)
        normalized = validate_writer_lease_request_v2(request_mapping, self.authority)
        expected_normalized = {
            **request_mapping,
            "create_target_branch": request.branch_mode == "CREATE_NEW",
            "worktree_mode": (
                "detached_at_base_oid"
                if request.branch_mode == "CREATE_NEW"
                else "detached_at_expected_head"
            ),
            "cas_ref_update_required": True,
        }
        if normalized != expected_normalized:
            raise ContinuationExecutionError("CONTINUATION_NORMALIZATION_INVALID")
        if not binding.original_receipt_valid or not binding.original_grant_consumed:
            raise ContinuationExecutionError("ORIGINAL_GRANT_LINEAGE_INVALID")
        binding.permissions.validate()
        now = self.now()
        if now < binding.lease_not_before:
            raise ContinuationExecutionError("LEASE_NOT_ACTIVE")
        if now >= binding.lease_expires_at:
            raise ContinuationExecutionError("LEASE_EXPIRED")
        if request.generation <= self._max_generation(request.target_branch, exclude_job_id=job_id):
            raise ContinuationExecutionError("GENERATION_NOT_MONOTONIC")
        if not self.workspace.is_detached():
            raise ContinuationExecutionError("WORKTREE_NOT_DETACHED")
        if self.workspace.head_oid() != request.expected_branch_oid:
            raise ContinuationExecutionError("WORKTREE_HEAD_MISMATCH")
        if self.remote.repository_identity() != request.repository:
            raise ContinuationExecutionError("REMOTE_REPOSITORY_IDENTITY_MISMATCH")

    def _job_request(self, binding: ContinuationRuntimeBinding, commit: ContinuationCommit, payload: bytes) -> dict[str, object]:
        request = binding.request
        return {
            "idempotency_key": commit.idempotency_key,
            "campaign_id": request.task_id,
            "task_id": request.task_id,
            "repository": request.repository,
            "operation": "CONTINUATION_COMMIT_FILES",
            "branch": request.target_branch,
            "expected_branch_oid": request.expected_branch_oid,
            "base_oid": request.base_oid,
            "branch_mode": request.branch_mode,
            "payload_digest": hashlib.sha256(payload).hexdigest(),
            "lineage_anchor_oid": request.base_oid,
            "generation": request.generation,
        }

    def execute(
        self,
        binding: ContinuationRuntimeBinding,
        commit: ContinuationCommit,
        *,
        before_complete: Callable[[str], None] | None = None,
    ) -> dict[str, object]:
        commit.payload()
        if not binding.job_id or binding.request.job_id != binding.job_id:
            raise ContinuationExecutionError("RUNTIME_LINEAGE_MISSING_JOB_ID")
        current = self.controller.store.read(binding.job_id)
        current_state = JobState(current["state"])
        if current_state == JobState.COMPLETED:
            return self.controller.store._receipt(current, duplicate=True).as_dict()
        if current_state == JobState.REMOTE_VERIFIED:
            receipt = json.loads(str(current["receipt_json"] or "{}"))
            commit_oid = str(receipt.get("authority_oid") or "")
            if not OID_RE.fullmatch(commit_oid):
                raise ContinuationExecutionError("GOVERNED_COMMIT_OID_INVALID")
            if before_complete is not None:
                before_complete(commit_oid)
            return self.controller.advance(
                binding.job_id,
                JobState.COMPLETED,
                side_effect_state="REMOTE_VERIFIED",
                receipt=receipt,
            ).as_dict()
        self._validate_preflight(binding, job_id=binding.job_id)
        request = binding.request
        projected = False
        try:
            self.credential.project()
            projected = True
            row = self.controller.store.read(binding.job_id)
            state = JobState(row["state"])
            if state == JobState.REQUESTED:
                self.controller.advance(
                    binding.job_id,
                    JobState.AUTHORIZED,
                    side_effect_state="AUTHORIZED",
                    receipt={"task_id": request.task_id, "original_grant_consumed": True},
                    expected_revision=int(row["revision"]),
                )
                row = self.controller.store.read(binding.job_id)
                state = JobState(row["state"])
            if state == JobState.AUTHORIZED:
                self.controller.lease_job(
                    binding.job_id,
                    lease_id=f"lease-{binding.job_id}",
                    lineage_anchor_oid=request.base_oid,
                    current_head_oid=request.expected_branch_oid,
                    generation=request.generation,
                    lease_expires_at=binding.lease_expires_at.astimezone(timezone.utc).isoformat(),
                    expected_revision=int(row["revision"]),
                )
                row = self.controller.store.read(binding.job_id)
                state = JobState(row["state"])
            if state == JobState.LEASED:
                self.controller.advance(binding.job_id, JobState.WORKTREE_READY,
                                        side_effect_state="WORKTREE_READY", expected_revision=int(row["revision"]))
                row = self.controller.store.read(binding.job_id)
                state = JobState(row["state"])
            if state == JobState.WORKTREE_READY:
                self.controller.advance(binding.job_id, JobState.EXECUTING,
                                        side_effect_state="IN_PROGRESS", expected_revision=int(row["revision"]))
            elif state != JobState.EXECUTING:
                raise ContinuationExecutionError("JOB_EXECUTION_STATE_INVALID")
            new_oid = self.workspace.commit_files(commit.files, commit.message)
            if not OID_RE.fullmatch(new_oid) or not self.authority.is_ancestor(request.expected_branch_oid, new_oid):
                raise ContinuationExecutionError("NON_FAST_FORWARD_COMMIT")
            if request.branch_mode == "CREATE_NEW":
                authority_oid = self.authority.create_task_branch_cas(request.target_branch, request.base_oid, new_oid)
            else:
                authority_oid = self.authority.compare_and_swap_branch(request.target_branch, request.expected_branch_oid, new_oid)
            if authority_oid != new_oid:
                raise ContinuationExecutionError("AUTHORITY_CAS_READBACK_MISMATCH")
            self.controller.advance(
                binding.job_id,
                JobState.LOCAL_VERIFIED,
                side_effect_state="LOCAL_VERIFIED",
                receipt={"authority_oid": authority_oid, "force_push": False},
            )
            try:
                self.controller.advance(binding.job_id, JobState.REMOTE_SYNCING, side_effect_state="IN_PROGRESS")
                if request.branch_mode == "CREATE_NEW":
                    self.remote.push_new_exact(request.target_branch, new_oid)
                else:
                    self.remote.push_exact(request.target_branch, request.expected_branch_oid, new_oid)
            except Exception:
                return self._remote_pending(binding.job_id, authority_oid, new_oid)
            return self._finish_remote_verification(
                binding.job_id,
                authority_oid,
                new_oid,
                before_complete=before_complete,
            )
        finally:
            if projected:
                self.credential.revoke()

    def _observe_remote(self, branch: str, oid: str) -> tuple[str | None, str | None, bool]:
        return self.remote.read_git_ref(branch), self.remote.read_api_branch(branch), self.remote.commit_visible(oid)

    def _remote_pending(self, job_id: str, authority_oid: str, new_oid: str) -> dict[str, object]:
        receipt = {
            "authority_oid": authority_oid,
            "desired_remote_oid": new_oid,
            "remote_verified": False,
            "force_push": False,
        }
        result = self.controller.advance(
            job_id,
            JobState.RECONCILIATION_REQUIRED,
            side_effect_state="REMOTE_PUSH_PENDING",
            receipt=receipt,
        )
        return result.as_dict()

    def _finish_remote_verification(
        self,
        job_id: str,
        authority_oid: str,
        new_oid: str,
        *,
        before_complete: Callable[[str], None] | None = None,
    ) -> dict[str, object]:
        row = self.controller.store.read(job_id)
        branch = row["branch"]
        request = json.loads(row["request_json"])["binding"]
        git_oid, api_oid, visible = self._observe_remote(branch, new_oid)
        decision = decide_reconciliation(RemoteObservation("aios/aios", branch, new_oid, authority_oid, git_oid, api_oid, visible, datetime.now(timezone.utc).isoformat(), request.get("expected_branch_oid")))
        if decision.action is not ReconciliationAction.MARK_REMOTE_VERIFIED:
            return self._remote_pending(job_id, authority_oid, new_oid)
        receipt = {
            "authority_oid": authority_oid,
            "git_remote_oid": git_oid,
            "api_branch_oid": api_oid,
            "commit_visible": True,
            "remote_verified": True,
            "reconciliation_decision": decision.action.value,
            "force_push": False,
        }
        self.controller.advance(
            job_id,
            JobState.REMOTE_VERIFIED,
            side_effect_state="REMOTE_VERIFIED",
            receipt=receipt,
        )
        if before_complete is not None:
            before_complete(new_oid)
        return self.controller.advance(
            job_id,
            JobState.COMPLETED,
            side_effect_state="REMOTE_VERIFIED",
            receipt=receipt,
        ).as_dict()

    def reconcile(self, job_id: str) -> dict[str, object]:
        row = self.controller.store.read(job_id)
        state = JobState(row["state"])
        if state == JobState.COMPLETED:
            return self.controller.store._receipt(row, duplicate=True).as_dict()
        if state != JobState.RECONCILIATION_REQUIRED:
            raise ContinuationExecutionError("JOB_NOT_RECONCILABLE")
        request = json.loads(row["request_json"])["binding"]
        receipt = json.loads(row["receipt_json"])
        new_oid = receipt["desired_remote_oid"]
        authority_oid = self.authority.branch_head(row["branch"])
        try:
            generation = int(request.get("generation") or 0)
        except (TypeError, ValueError) as exc:
            raise ContinuationExecutionError("DURABLE_BINDING_GENERATION_INVALID") from exc
        if generation > 0:
            with self.controller.store._connect() as connection:
                earlier = connection.execute(
                    "SELECT 1 FROM jobs WHERE branch=? AND state=? "
                    "AND CAST(json_extract(request_json, '$.binding.generation') AS INTEGER) < ? LIMIT 1",
                    (row["branch"], JobState.RECONCILIATION_REQUIRED.value, generation),
                ).fetchone()
            if earlier is not None:
                raise ContinuationExecutionError("EARLIER_RECONCILIATION_REQUIRED")
        self.credential.project()
        try:
            git_oid, api_oid, visible = self._observe_remote(row["branch"], new_oid)
            if authority_oid == git_oid == api_oid == new_oid and visible:
                self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST")
                return self._finish_remote_verification(job_id, new_oid, new_oid)
            expected_oid = request["expected_branch_oid"]
            if (
                self.authority.is_ancestor(new_oid, authority_oid)
                and git_oid is not None
                and git_oid == api_oid
                and self.authority.is_ancestor(git_oid, new_oid)
            ):
                self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST_CHAIN")
                self.remote.push_exact(row["branch"], git_oid, new_oid)
                return self._finish_remote_verification(job_id, new_oid, new_oid)
            if authority_oid == new_oid and git_oid == api_oid == expected_oid:
                self.controller.advance(job_id, JobState.RECONCILING, side_effect_state="OBSERVATION_FIRST_RETRY")
                self.remote.push_exact(row["branch"], expected_oid, new_oid)
                return self._finish_remote_verification(job_id, new_oid, new_oid)
            return self.controller.store._receipt(row, duplicate=True).as_dict()
        finally:
            self.credential.revoke()