"""Backend-neutral worker dispatch core."""
from __future__ import annotations

import json
import re
import subprocess
import time
from dataclasses import dataclass, field, replace
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Mapping, Sequence

from .dispatch_state import (
    BACKEND_CLI_WRAPPER,
    BACKEND_CMUX_PANE,
    BACKEND_MIXED,
    append_worker_dispatch as _append_worker_dispatch,
    build_dispatch_id as _build_dispatch_id,
    DispatchError,
    WorkerJob,
    dispatch_completion_paths as _completion_paths,
    dispatch_mode as _dispatch_mode,
    dispatch_result_path as _result_path_for_worker,
    dispatch_session_id as _dispatch_session_id,
    LIVENESS_AUDIT_HEARTBEAT,
    LIVENESS_WRAPPER_STATUS,
    load_json_object as _load_json_object,
    link_agent_dispatch_result as _link_agent_dispatch_result,
    missing_completion_paths as _missing_completion_paths,
    mutate_team_state as _mutate_team_state,
    require_string as _require_string,
    resolve_project_path as _resolve_project_path,
    resolve_required_path as _resolve_required_path,
    record_dispatch_facts as _record_dispatch_facts,
    transition_worker_status as _transition_worker_status,
    update_worker_dispatch_status as _update_worker_dispatch_status,
    string_list as _string_list,
    string_value as _string_value,
    TEARDOWN_BEFORE_TERMINAL_REASON,
    utc_now as _utc_now,
    validate_dispatch_prompts as _validate_dispatch_prompts,
    v2_worker_state_key as _v2_worker_state_key,
    worker_execution_identity as _worker_execution_identity,
    worker_jobs_from_file as _worker_jobs_from_file,
    worker_state as _worker_state,
    worktree_path as _worktree_path,
)
from .assignment_resolver import (
    AssignmentEnvironment,
    AssignmentResolutionError,
    RoleInstance,
    resolve_model_assignment,
)
from .application.open_worker import open_worker
from .domain.host import HostSessionContext
from .domain.provider import HostModelBinding, ServedModelAttestation
from .domain.worker_exec import SERVED_MODEL_MISMATCH_EXIT_CODE
from .domain.worker_runtime import (
    SURFACE_CMUX_PANE,
    EnvironmentBlocked,
    ProgressEvent,
    RuntimeHandle,
    SurfaceUnavailable,
    WorkerSpawnRequest,
)
from .ports.worker_runtime import WorkerRuntimePort
from .execution_identity import Attempt, Invocation, RoleExecution, model_spec_digest
from .execution_manifest import (
    ExecutionManifestError,
    finish_attempt_mutation,
    read_execution_manifest,
    record_invocation_attempt,
)
from .execution_mutation_audit import (
    ExecutionMutationAudit,
    MutationAuditResult,
    MutationSnapshot,
    assert_compatible_batch,
)
from .convergence_store import write_json_atomic
from .json_boundary import JsonBoundaryError, external_worker_json_source, load_external_json
from .assignment_environment import load_assignment_context
from .model_pool import ModelPool
from .error_log_write import append_observed
from .lead_events import LeadEvent, append_lead_event
from .initial_prompt_materialization import (
    InitialPromptMaterializationError,
    InitialPromptMaterializationRequest,
    InitialPromptWorkerRequest,
    PromptDeliveryMode,
    materialize_initial_prompts,
)
from .agent_invocation import AgentInvocationError, materialize_retry_invocation
from .path_hints import hydrate_active_run_context
from .schema_excerpt import (
    bundle_excerpt_path,
    describe_changed,
    excerpt_contract_skew,
)
from .seeding import installed_version
from .worker_audit_ledger import (
    check_worker_results_audit,
    parse_worker_result_name,
)
from .worker_prompt_body import REPORT_WRITER_WORKER_ID
from .worker_prompt_headers import (
    WorkerPromptHeaderError,
    resolve_errors_log_path,
)
from .worker_artifact_paths import audit_sidecar_rel
from .wrapper_status import (
    log_path_for_prompt,
    mutation_snapshot_path_for_prompt,
    prompt_derived_paths,
    read_wrapper_status,
    status_path_for_prompt,
)
from .worker_request import verifier_extra_dirs
from .write_policy import (
    WriteEnforcement,
    WritePolicy,
    build_invocation_write_contract,
    planned_paths_from_run_manifest,
    write_enforcement_from_payload,
    write_policy_from_payload,
)


MAX_WORKER_ATTEMPTS = 2
TERMINAL_DISPATCH_STATUSES = {"completed", "timeout", "error", "not-run"}
# What the error log records for a wrapper the dispatcher timed out, matching
# the value `team-contract` prescribes for a polling-cap termination.
_WRAPPER_TIMEOUT_EXIT_CODE = 124
# The excerpt shares one atomic PIPE_BUF append with the rest of the record, so
# it is capped far below the writer's own 2048-byte stderr limit.
_WRAPPER_LOG_TAIL_BYTES = 800


@dataclass(frozen=True)
class RevalidatedBinding:
    role_execution_ref: str
    binding: HostModelBinding
    model_spec_digest: str


def revalidate_role_execution(
    role_execution: RoleExecution,
    pool: ModelPool,
    host: AssignmentEnvironment,
) -> RevalidatedBinding:
    """Recompute one selected model binding without inspecting unrelated rows."""
    if (
        role_execution.model_ref is None
        or role_execution.binding is None
        or role_execution.model_spec_digest is None
    ):
        raise DispatchError("model binding changed: role execution is unbound")
    try:
        assignment = resolve_model_assignment(
            instance=RoleInstance(
                role_execution.role,
                role_execution.role,
                role_execution.ordinal,
            ),
            model_ref=role_execution.model_ref,
            pool=pool,
            host=HostSessionContext(
                host_id=host.host_descriptor.id,
                entry_mode="spawn-process",
                available_functions=frozenset(),
                interaction_surface="role-execution-dispatch",
            ),
            environment=host,
        )
        binding = assignment.binding
        if binding is None:
            raise AssignmentResolutionError("selected model has no worker binding")
        digest = model_spec_digest(pool.resolve(role_execution.model_ref), binding)
    except (ValueError, RuntimeError) as exc:
        raise DispatchError(f"model binding changed: {exc}") from exc
    identity_matches = (
        assignment.provider_id == role_execution.provider
        and assignment.model_id == role_execution.model_id
        and binding == role_execution.binding
        and digest == role_execution.model_spec_digest
    )
    if not identity_matches:
        raise DispatchError("model binding changed for selected role execution")
    return RevalidatedBinding(role_execution.role_execution_ref, binding, digest)


def dispatch_revalidated_role_execution(
    role_execution: RoleExecution,
    pool: ModelPool,
    host: AssignmentEnvironment,
    dispatch,
):
    """Run a dispatch callback only after the pure binding gate succeeds."""
    binding = revalidate_role_execution(role_execution, pool, host)
    return dispatch(binding)


def verify_served_model(
    role_execution: RoleExecution,
    attestation: ServedModelAttestation,
    *,
    pool: ModelPool,
) -> ServedModelAttestation:
    """Reject observed substitution while preserving unobservable attempts."""
    if attestation.level == "unknown":
        return attestation
    if role_execution.model_ref is None or not attestation.normalized_model_ref:
        raise DispatchError("served model differs from selected model")
    # An unregistered ref and a genuine substitution are different failures.
    # Folding both into "differs from selected" hid which one happened, and a
    # catalog gap reads as a provider swapping the model out from under us.
    try:
        selected = pool.resolve(role_execution.model_ref)
    except ValueError as exc:
        raise DispatchError(
            f"selected model is not in the catalog: {role_execution.model_ref}"
        ) from exc
    try:
        observed = pool.resolve(attestation.normalized_model_ref)
    except ValueError as exc:
        raise DispatchError(
            "served model is not in the catalog: "
            f"{attestation.normalized_model_ref} "
            f"(provider reported {attestation.observed_model!r})"
        ) from exc
    expected_level = "channel" if observed.version_kind == "channel" else "exact"
    if attestation.level != expected_level:
        raise DispatchError("served model attestation level is inconsistent")
    if selected.version_kind == "pinned":
        if str(selected.model_ref) != str(observed.model_ref):
            raise DispatchError("served model differs from selected model")
        return attestation
    if (
        selected.provider_id != observed.provider_id
        or not selected.channel_family
        or selected.channel_family != observed.channel_family
    ):
        raise DispatchError("served model channel family differs")
    return attestation


@dataclass(frozen=True)
class WorkerHandle:
    job: WorkerJob
    pane_id: str
    completed_process: subprocess.CompletedProcess[str] | None
    status_sidecar_path: Path | None
    degraded_from: str
    running_process: subprocess.Popen[str] | None = None
    mutation_snapshot: MutationSnapshot | None = None
    mutation_policies: tuple[WritePolicy, ...] = ()
    mutation_retry_allowed: bool = True
    write_policy_digest: str = ""
    write_enforcement: WriteEnforcement | None = None
    runtime_handle: RuntimeHandle | None = None


@dataclass(frozen=True)
class WorkerOutcome:
    returncode: int
    missing_completion_paths: tuple[Path, ...]
    pane_id: str
    status_sidecar_path: Path | None
    timeout: bool
    terminal_stage: str
    degraded_from: str


@dataclass(frozen=True)
class DispatchPlan:
    project_root: Path
    workspace_root: Path
    manifest_path: Path
    team_state_path: Path
    lead_events_path: Path
    manifest: Mapping[str, Any]
    jobs: tuple[WorkerJob, ...]
    default_backend: str
    runtime_chain: tuple[WorkerRuntimePort, ...] = ()

    def to_payload(self, dry_run: bool) -> dict[str, Any]:
        payload = {
            "ok": True,
            "dryRun": dry_run,
            "dispatchMode": _dispatch_mode(self.jobs),
            "workerBackends": {job.worker_id: job.backend for job in self.jobs},
            "runManifestPath": str(self.manifest_path),
            "teamStatePath": str(self.team_state_path),
            "leadEventsPath": str(self.lead_events_path),
            "workspaceRoot": str(self.workspace_root),
            "workers": [job.to_payload() for job in self.jobs],
        }
        if self.jobs:
            payload["workerDispatches"] = [
                _dispatch_record(job, 1, "not-run", "", "") for job in self.jobs
            ]
        return payload


@dataclass(frozen=True)
class _BuildOptions:
    okstra_bin: Path | None
    idle_timeout_seconds: int | None
    default_backend: str
    supported_worker_wrappers: Mapping[str, str]
    unsupported_worker_label: str
    dispatch_kind: str
    cli_wrapper_assignments_only: bool = False
    default_provider_by_worker_id: Mapping[str, str] = field(default_factory=dict)


@dataclass(frozen=True)
class _RosterWorkerFacts:
    worker_id: str
    model: str
    provider: str
    result_path: str
    role: str


def build_dispatch_plan(
    *,
    project_root: Path,
    run_manifest_path: Path,
    workspace_root: Path,
    okstra_bin: Path | None = None,
    requested_workers: Sequence[str] = (),
    idle_timeout_seconds: int | None = None,
    required_lead_runtime: str | None = None,
    default_backend: str = BACKEND_CLI_WRAPPER,
    supported_worker_wrappers: Mapping[str, str],
    unsupported_worker_label: str,
    dispatch_kind: str = "initial",
    jobs_file: Path | None = None,
    cli_wrapper_assignments_only: bool = False,
    default_provider_by_worker_id: Mapping[str, str] | None = None,
) -> DispatchPlan:
    project_root = project_root.resolve()
    manifest_path = _resolve_project_path(project_root, str(run_manifest_path))
    manifest = _load_json_object(manifest_path, "run manifest")
    _validate_manifest(manifest, manifest_path, required_lead_runtime)
    team_state_path = _resolve_required_path(project_root, manifest, "teamStatePath")
    team_state = _load_json_object(team_state_path, "team-state")
    active_context = _load_optional_json(project_root, manifest.get("activeRunContextPath"))
    options = _BuildOptions(
        okstra_bin=okstra_bin,
        idle_timeout_seconds=idle_timeout_seconds,
        default_backend=default_backend,
        supported_worker_wrappers=supported_worker_wrappers,
        unsupported_worker_label=unsupported_worker_label,
        dispatch_kind=dispatch_kind,
        cli_wrapper_assignments_only=cli_wrapper_assignments_only,
        default_provider_by_worker_id=dict(default_provider_by_worker_id or {}),
    )
    if jobs_file:
        jobs = _jobs_from_file(
            project_root, workspace_root, jobs_file, manifest, active_context, options
        )
    else:
        jobs = _jobs_from_roster(
            project_root,
            workspace_root,
            manifest_path,
            manifest,
            team_state,
            active_context,
            requested_workers,
            options,
        )
    # Ahead of the per-prompt checks: a batch the lead must split is a plan-shape
    # problem, and reporting a prompt-header error for a doomed batch buries it.
    _validate_implementation_phase_order(jobs)
    _validate_dispatch_prompts(manifest, active_context, jobs)
    _reject_stale_schema_excerpt(project_root, manifest, jobs)
    return DispatchPlan(
        project_root=project_root,
        workspace_root=workspace_root.resolve(),
        manifest_path=manifest_path,
        team_state_path=team_state_path,
        lead_events_path=_resolve_required_path(project_root, manifest, "leadEventsPath"),
        manifest=manifest,
        jobs=tuple(jobs),
        default_backend=default_backend,
    )


def dispatch_plan(plan: DispatchPlan, *, wait: bool = True) -> int:
    plan = _ensure_runtime_chain(plan)
    _validate_report_writer_isolation(plan.jobs)
    _validate_implementation_phase_order(plan.jobs)
    _revalidate_plan_role_executions(plan)
    if wait:
        pane_backends = sorted(
            {
                job.backend
                for job in plan.jobs
                if job.backend == BACKEND_CMUX_PANE
            }
        )
        if pane_backends:
            raise DispatchError(
                f"wait=True dispatch does not support {'/'.join(pane_backends)} "
                "workers: the per-job blocking loop would serialize panes "
                "instead of running them concurrently; dispatch panes with "
                "wait=False"
            )
        _record_dispatch_facts(plan.team_state_path, _dispatch_mode(plan.jobs))
        for job in plan.jobs:
            result = _dispatch_job_with_retry(plan, job)
            if result != 0:
                return result
        return 0
    round_artifact_paths = _round_artifact_paths(plan)
    handles = [
        _spawn_job(
            plan, job, _next_attempt(plan, job),
            batch_artifact_paths=round_artifact_paths,
        )
        for job in plan.jobs
    ]
    _record_dispatch_facts(plan.team_state_path, _mode_from_handles(handles))
    return 0


def _orchestrator_artifact_roots(plan: DispatchPlan) -> tuple[Path, ...]:
    """okstra 자신이 이 run 동안 쓰는 산출물 경로.

    `_round_artifact_paths` 는 이번 배치의 job 만 알기 때문에 원리상 완결될 수
    없다. 감사 창에 들어오는 것은 형제 job 의 산출물만이 아니라 같은 run 의 다른
    라운드가 남기는 것들이다 — 재시도 프롬프트와 그 `.meta.json` / `.publish.lock`,
    이전 라운드 워커의 `.log` / `.status.json` / `.mutation-audit.json`, critic 의
    결과, design-prep 요청. 실측 한 건에서 그렇게 빠져나간 경로가 16개였고 그
    배치의 디스패치 5건이 전부 `readonly source changed` 로 떨어졌다.

    그래서 파일을 더 세는 대신 트리를 싣는다. 이 디렉터리 아래는 정의상 okstra 의
    산출물이고 프로젝트 소스가 아니다 — 소스가 바뀌었는지를 묻는 질문의 답이 될 수
    없다.
    """
    # 실행 매니페스트는 `<run_dir>/manifests/` 아래 놓인다.
    roots = [plan.manifest_path.resolve().parents[1]]
    # task 매니페스트와 discovery 카탈로그는 run 디렉터리 위에 있고, prep 과
    # 완료 기록이 실행 중에 갱신한다. 경로는 매니페스트가 이미 들고 있으므로
    # 디렉터리 깊이를 세지 않는다 — implementation 은 stage 층이 하나 더 있어
    # 고정 인덱스로는 task root 에 닿지 못한다.
    for key in ("taskManifestPath", "taskCatalogPath"):
        value = plan.manifest.get(key)
        if isinstance(value, str) and value:
            candidate = Path(value)
            roots.append(
                candidate if candidate.is_absolute()
                else (plan.project_root / candidate)
            )
    return tuple(roots)


# 이 모듈이 정산하는 모든 디스패치의 집행 방식. 상수인 이유는
# `dispatch_state._AGENT_ENFORCEMENT_MODES` 의 다른 값(`host-native-spec-link-gate`)
# 이 호스트 원시 호출 전용이고 이 경로로 오지 않기 때문이다.
CLI_DISPATCH_ENFORCEMENT_MODE = "core-pre-dispatch"


def _round_artifact_paths(plan: DispatchPlan) -> tuple[Path, ...]:
    """Every artifact this round's workers are entitled to write.

    These jobs run concurrently into one shared artifact root, so each worker's
    own result, audit sidecar, status sidecar and live log land inside every
    sibling's audit window. Judged against one worker's policy alone, the
    siblings' writes read as unauthorized artifact-root changes and failed a
    whole round of otherwise clean verifiers.

    They travel as orchestrator paths rather than as a widened policy union
    because the snapshot carries orchestrator paths and the audit excuses them
    without consulting a policy — while `_validate_snapshot_authority` compares
    the recorded policy digests exactly, so a snapshot taken under a union can
    no longer be closed by the per-job policy the awaiting process rebuilds.
    Same effect on the verdict; no coupling between two processes' plans.
    """
    return tuple(
        path
        for job in plan.jobs
        for path in _worker_artifact_paths(plan, job)
    )


def dispatch_cli_wrapper_plan(plan: DispatchPlan) -> int:
    """Start one dependency-free CLI batch before collecting any worker."""
    if any(job.backend != BACKEND_CLI_WRAPPER for job in plan.jobs):
        raise DispatchError("concurrent CLI dispatch requires cli-wrapper jobs only")
    plan = _ensure_runtime_chain(plan)
    _validate_report_writer_isolation(plan.jobs)
    _validate_implementation_phase_order(plan.jobs)
    _revalidate_plan_role_executions(plan)
    _record_dispatch_facts(plan.team_state_path, _dispatch_mode(plan.jobs))
    final_codes = _dispatch_cli_wrapper_batch(plan, plan.jobs)
    return next((final_codes[job.worker_id] for job in plan.jobs
                 if final_codes.get(job.worker_id, 0) != 0), 0)


def _revalidate_job_role_execution(plan: DispatchPlan, job: WorkerJob) -> None:
    manifest = read_execution_manifest(plan.manifest_path)
    executions = {
        row.role_execution_ref: row for row in manifest.role_executions
    }
    context = load_assignment_context(
        host_runtime=_require_string(plan.manifest, "leadRuntime"),
        terminal_backend=_require_string(plan.manifest, "terminalBackend"),
        execution_provider_ids=(job.provider,),
    )
    role_execution = executions.get(job.role_execution_ref)
    if (
        role_execution is None
        or role_execution.participant_ref != job.participant_ref
        or role_execution.execution_label != job.execution_label
    ):
        raise DispatchError("worker job role execution does not match run manifest")
    revalidate_role_execution(
        role_execution,
        context.pool,
        context.environment,
    )


def _revalidate_plan_role_executions(plan: DispatchPlan) -> None:
    for job in plan.jobs:
        if job.has_execution_identity:
            _revalidate_job_role_execution(plan, job)


def _validate_report_writer_isolation(jobs: Sequence[WorkerJob]) -> None:
    if (
        any(job.worker_id == REPORT_WRITER_WORKER_ID for job in jobs)
        and len(jobs) > 1
    ):
        raise DispatchError(
            "report-writer must run in a separate Phase 6 dispatch after "
            "analysis and convergence"
        )


def _validate_implementation_phase_order(jobs: Sequence[WorkerJob]) -> None:
    """Keep the executor and the verifiers in two dispatches, never one batch.

    A verifier reads the diff the executor wrote, so `implementation.md` binds
    its contract to the window "between Executor stage completion and the first
    verifier dispatch". Nothing enforced that ordering: one
    `okstra team dispatch --workers claude,codex` started both at once and the
    verifier observed base HEAD instead of the stage diff, which is what its
    result then reported on.

    The refusal names the worker IDs on both sides because the second dispatch
    has to drop the executor's ID: `worker_prompt_policy` re-materializes that
    ID as the executor every time, so a `--workers` list that keeps it is
    refused again and a message saying only "split the batch" would loop.

    `build_dispatch_plan` is the only call every production dispatch takes, and
    the one `--dry-run` stops at, so it is what makes the preview refuse. The
    two dispatch entry points are public and take a caller-assembled plan, so
    they check it again there rather than trust their input.
    """
    executors = [
        job.worker_id for job in jobs if job.audience == "implementation-executor"
    ]
    verifiers = [
        job.worker_id for job in jobs if job.audience == "implementation-verifier"
    ]
    if executors and verifiers:
        executor_ids = ", ".join(executors)
        raise DispatchError(
            "an implementation batch may not hold the executor and a verifier: "
            f"{_named_role(executors, 'executor')}, "
            f"{_named_role(verifiers, 'verifier')}. "
            "Dispatch the executor alone, settle it with `okstra team await`, "
            f"then dispatch the verifiers with `--workers` omitting {executor_ids} "
            "— that worker ID materializes as the executor again, so a second "
            "batch holding it is refused too. The split is what lets a verifier "
            "observe the stage diff instead of base HEAD"
        )


def _named_role(worker_ids: Sequence[str], role: str) -> str:
    """`verifier codex` for one, `verifiers codex, antigravity` for several.

    A roster that adds antigravity puts two verifiers in the same refusal, and
    a singular label there reads as one worker named "codex, antigravity".
    """
    return f"{role}{'s' if len(worker_ids) > 1 else ''} {', '.join(worker_ids)}"


def _dispatch_cli_wrapper_batch(
    plan: DispatchPlan,
    jobs: Sequence[WorkerJob],
) -> dict[str, int]:
    """Run one concurrent batch through all retries before the next dependency."""
    pending = [(job, 1) for job in jobs]
    final_codes: dict[str, int] = {}
    baseline_snapshot: MutationSnapshot | None = None
    while pending:
        active: list[tuple[WorkerHandle, int]] = []
        draft_contracts = tuple(
            _canonical_write_contract(plan, job) if job.has_execution_identity else None
            for job, _attempt in pending
        )
        draft_policies = tuple(
            contract[0] for contract in draft_contracts if contract is not None
        )
        if draft_policies:
            assert_compatible_batch(draft_policies)
        prepared = [
            (_prepare_job_attempt(plan, job, attempt), attempt)
            for job, attempt in pending
        ]
        contracts = tuple(
            _persisted_write_contract(plan, job) for job, _ in prepared
        )
        policies = tuple(
            contract[0] for contract in contracts if contract is not None
        )
        needs_batch = any(
            contract is not None and contract[1].mutation_audit == "batch"
            for contract in contracts
        )
        if needs_batch and baseline_snapshot is None:
            baseline_snapshot = _mutation_snapshot(
                plan, policies, tuple(job for job, _ in prepared)
            )
        snapshot = baseline_snapshot if needs_batch else None
        starting_job: WorkerJob | None = None
        starting_attempt = 1
        try:
            for (job, attempt), contract in zip(prepared, contracts, strict=True):
                starting_job = job
                starting_attempt = attempt
                handle = _spawn_cli_job_nonblocking(plan, job, attempt)
                handle = replace(
                    handle,
                    mutation_snapshot=snapshot,
                    mutation_policies=policies,
                    write_policy_digest=contract[0].digest if contract else "",
                    write_enforcement=contract[1] if contract else None,
                )
                active.append((handle, attempt))
                _record_dispatch(
                    plan.team_state_path, handle, attempt, "running", ""
                )
                _append_event(
                    plan,
                    "worker-dispatched",
                    _attempt_details(job, attempt, handle),
                )
        except (OSError, DispatchError) as exc:
            if starting_job is not None:
                _abort_cli_batch_start(
                    plan,
                    active,
                    starting_job,
                    starting_attempt,
                    prepared,
                    snapshot,
                    policies,
                    exc,
                )
            raise
        from .adapters.runtime.assembly import port_for

        outcomes: list[tuple[WorkerHandle, int, WorkerOutcome]] = []
        for handle, attempt in active:
            runtime = handle.runtime_handle
            if runtime is None or not runtime.waitable:
                raise DispatchError("concurrent CLI worker process is missing")
            port = port_for(plan.runtime_chain, runtime.surface)
            returncode = port.wait(runtime)
            completed = subprocess.CompletedProcess(handle.job.command, returncode)
            settled = replace(handle, completed_process=completed)
            outcomes.append((settled, attempt, _outcome_from_completed(settled)))

        next_round: list[tuple[WorkerJob, int]] = []
        for handle, attempt, outcome in outcomes:
            job = handle.job
            retry_allowed = _finish_attempt(
                plan,
                job,
                attempt,
                outcome,
                snapshot=handle.mutation_snapshot,
                policies=handle.mutation_policies,
            )
            if _job_terminal_status(plan, job, attempt) == "completed":
                final_codes[job.worker_id] = 0
                continue
            if _should_retry(outcome, attempt) and retry_allowed:
                _append_event(
                    plan,
                    "worker-retry-scheduled",
                    _retry_details(job, attempt, outcome),
                )
                next_round.append((_job_for_next_attempt(job), attempt + 1))
                continue
            final_codes[job.worker_id] = outcome.returncode or 1
        pending = next_round
    return final_codes


def _abort_cli_batch_start(
    plan: DispatchPlan,
    active: Sequence[tuple[WorkerHandle, int]],
    failed_job: WorkerJob,
    failed_attempt: int,
    pending: Sequence[tuple[WorkerJob, int]],
    snapshot: MutationSnapshot | None,
    policies: Sequence[WritePolicy],
    error: Exception,
) -> None:
    from .adapters.runtime.assembly import port_for

    reason = f"concurrent CLI batch launch failed: {error}"
    for handle, _attempt in active:
        runtime = handle.runtime_handle
        if runtime is None:
            continue
        port_for(plan.runtime_chain, runtime.surface).terminate(runtime)
    for handle, attempt in active:
        runtime = handle.runtime_handle
        if runtime is not None:
            port_for(plan.runtime_chain, runtime.surface).wait(runtime)
        _transition_job_status(plan, handle.job, "error", reason)
        _update_dispatch_status(
            plan.team_state_path, handle.job, attempt, "error", reason
        )
    _transition_job_status(plan, failed_job, "error", reason)
    if not _update_dispatch_status(
        plan.team_state_path,
        failed_job,
        failed_attempt,
        "error",
        reason,
    ):
        failed_contract = _persisted_write_contract(plan, failed_job)
        failed_handle = WorkerHandle(
            failed_job,
            "",
            None,
            status_path_for_prompt(failed_job.prompt_path),
            "",
            mutation_snapshot=snapshot,
            mutation_policies=tuple(policies),
            write_policy_digest=(
                failed_contract[0].digest if failed_contract else ""
            ),
            write_enforcement=(failed_contract[1] if failed_contract else None),
        )
        _record_dispatch(
            plan.team_state_path,
            failed_handle,
            failed_attempt,
            "error",
            reason,
        )
    closed = {
        (handle.job.worker_id, attempt) for handle, attempt in active
    }
    closed.add((failed_job.worker_id, failed_attempt))
    for job, attempt in pending:
        if (job.worker_id, attempt) in closed:
            continue
        _transition_job_status(plan, job, "error", reason)
        if not _update_dispatch_status(
            plan.team_state_path, job, attempt, "error", reason
        ):
            contract = _persisted_write_contract(plan, job)
            handle = WorkerHandle(
                job,
                "",
                None,
                status_path_for_prompt(job.prompt_path),
                "",
                mutation_snapshot=snapshot,
                mutation_policies=tuple(policies),
                write_policy_digest=contract[0].digest if contract else "",
                write_enforcement=contract[1] if contract else None,
            )
            _record_dispatch(
                plan.team_state_path,
                handle,
                attempt,
                "error",
                reason,
            )
    for job, attempt in pending:
        if not job.has_execution_identity:
            continue
        outcome = WorkerOutcome(
            returncode=1,
            missing_completion_paths=_missing_completion_paths(job),
            pane_id="",
            status_sidecar_path=status_path_for_prompt(job.prompt_path),
            timeout=False,
            terminal_stage="launch-failed",
            degraded_from="",
        )
        mutation = _audit_attempt(job, outcome, snapshot, policies)
        _finish_manifest_attempt(plan, job, attempt, mutation, None)


def await_dispatches(
    plan: DispatchPlan,
    *,
    poll_interval_seconds: int = 5,
    timeout_seconds: int | None = None,
    heartbeat_seconds: int = 30,
) -> int:
    _correct_teardown_marked_dispatches(plan)
    deadline = time.monotonic() + timeout_seconds if timeout_seconds is not None else None
    last_heartbeat = 0.0
    while True:
        running = _running_dispatches(plan.team_state_path)
        if not running:
            return 0
        _advance_running_dispatches(plan, running)
        running = _running_dispatches(plan.team_state_path)
        if not running:
            return 0
        if deadline is not None and time.monotonic() >= deadline:
            return 1
        now = time.monotonic()
        if heartbeat_seconds > 0 and now - last_heartbeat >= heartbeat_seconds:
            print(f"WAITING {len(running)} worker dispatch(es)")
            last_heartbeat = now
        time.sleep(max(poll_interval_seconds, 0))


def _jobs_from_roster(
    project_root: Path,
    workspace_root: Path,
    manifest_path: Path,
    manifest: Mapping[str, Any],
    team_state: Mapping[str, Any],
    active_context: Mapping[str, Any],
    requested_workers: Sequence[str],
    options: _BuildOptions,
) -> list[WorkerJob]:
    selected = _select_workers(
        manifest,
        requested_workers,
        options,
        team_state=team_state,
    )
    if selected == ["translator"]:
        return [_translator_job_from_reservation(
            project_root,
            workspace_root,
            manifest,
            team_state,
            active_context,
            options,
        )]
    _mark_roster_skips(project_root, manifest, team_state, selected, requested_workers, options)
    worker_facts, prompt_paths = _materialize_roster_prompts(
        project_root,
        workspace_root,
        manifest_path,
        manifest,
        team_state,
        selected,
        options,
    )
    return [
        _job_from_roster_worker(
            project_root,
            workspace_root,
            manifest,
            active_context,
            fact,
            prompt_path,
            options,
        )
        for fact, prompt_path in (
            (fact, prompt_paths[fact.worker_id]) for fact in worker_facts
        )
    ]


def _select_workers(
    manifest: Mapping[str, Any],
    requested_workers: Sequence[str],
    options: _BuildOptions,
    *,
    team_state: Mapping[str, Any] | None = None,
) -> list[str]:
    recommended = _string_list(manifest.get("recommendedWorkers"))
    if not recommended:
        raise DispatchError("run manifest has no recommendedWorkers entries")
    if tuple(requested_workers) == ("translator",):
        provider = _canonical_translator_provider(manifest)
        if provider is None:
            raise DispatchError(
                "translator dispatch requires a canonical translator role execution"
            )
        if provider not in options.supported_worker_wrappers:
            raise DispatchError(
                f"unsupported {options.unsupported_worker_label} translator provider: "
                f"{provider}"
            )
        return ["translator"]
    if options.cli_wrapper_assignments_only:
        return _select_cli_wrapper_assignments(
            recommended,
            requested_workers,
            team_state or {},
            options,
        )
    supported = set(options.supported_worker_wrappers)
    selected = (
        list(requested_workers)
        if requested_workers
        else [
            worker
            for worker in recommended
            if worker in supported and worker != REPORT_WRITER_WORKER_ID
        ]
    )
    unknown = [worker for worker in selected if worker not in recommended]
    if unknown:
        raise DispatchError("requested worker(s) are not in this run roster: " + ", ".join(unknown))
    unsupported = [worker for worker in selected if worker not in supported]
    if unsupported:
        allowed = ", ".join(sorted(supported))
        raise DispatchError(
            f"unsupported {options.unsupported_worker_label} worker(s): "
            + ", ".join(unsupported)
            + f" (dispatcher supports: {allowed})"
        )
    return selected


def _canonical_translator_provider(manifest: Mapping[str, Any]) -> str | None:
    execution = _canonical_translator_execution(manifest)
    return _string_value(execution.get("provider")) if execution else None


def _canonical_translator_execution(
    manifest: Mapping[str, Any],
) -> Mapping[str, Any] | None:
    assignments = manifest.get("invocationAssignments")
    assignment = (
        assignments.get("translator") if isinstance(assignments, Mapping) else None
    )
    if not isinstance(assignment, Mapping):
        return None
    provider = _string_value(assignment.get("provider"))
    model_execution = _string_value(assignment.get("modelExecutionValue"))
    runner = _string_value(assignment.get("runner"))
    if not provider or not model_execution or runner != BACKEND_CLI_WRAPPER:
        return None
    role_executions = manifest.get("roleExecutions")
    matches = [
        row
        for row in role_executions
        if isinstance(row, Mapping)
        and row.get("role") == "translator"
        and row.get("provider") == provider
        and isinstance(row.get("binding"), Mapping)
        and row["binding"].get("runner") == runner
        and row["binding"].get("resolvedExecutionValue") == model_execution
    ] if isinstance(role_executions, list) else []
    return matches[0] if len(matches) == 1 else None


_TRANSLATOR_RESERVATION_KEYS = {
    "schemaVersion", "executionIdentityVersion", "invocationId",
    "assignmentRef", "audience", "dispatchKind", "promptPath",
    "metadataPath", "invocationRef", "participantRef",
    "roleExecutionRef", "dutyId", "attempt",
}


def _translator_job_from_reservation(
    project_root: Path,
    workspace_root: Path,
    manifest: Mapping[str, Any],
    team_state: Mapping[str, Any],
    active_context: Mapping[str, Any],
    options: _BuildOptions,
) -> WorkerJob:
    execution = _canonical_translator_execution(manifest)
    if execution is None:
        raise DispatchError(
            "translator dispatch requires a canonical translator role execution"
        )
    contract = manifest.get("agentContract")
    if not isinstance(contract, Mapping):
        raise DispatchError("translator dispatch requires an agent contract")
    reservation_root = _resolve_required_path(
        project_root, contract, "invocationReservationRootPath"
    )
    dispatched_ids = {
        str(row.get("dispatchId") or "")
        for collection in (
            team_state.get("workerDispatches") or [],
            team_state.get("agentDispatches") or [],
        )
        for row in collection
        if isinstance(row, Mapping)
    }
    candidates: list[Mapping[str, Any]] = []
    for path in sorted(reservation_root.glob("*.json")):
        reservation = _load_json_object(path, "agent invocation reservation")
        if (
            set(reservation) == _TRANSLATOR_RESERVATION_KEYS
            and reservation.get("schemaVersion") == "2.0"
            and reservation.get("executionIdentityVersion") == 2
            and reservation.get("assignmentRef") == "translator"
            and reservation.get("audience") == "translator"
            and reservation.get("dispatchKind") == "translator"
            and reservation.get("dutyId") == "translator"
            and reservation.get("participantRef") == execution.get("participantRef")
            and reservation.get("roleExecutionRef") == execution.get("roleExecutionRef")
            and _build_dispatch_id(
                str(reservation.get("invocationId") or ""),
                int(reservation.get("attempt") or 0),
            ) not in dispatched_ids
        ):
            candidates.append(reservation)
    if len(candidates) != 1:
        raise DispatchError(
            "translator dispatch requires exactly one canonical invocation reservation"
        )
    reservation = candidates[0]
    prompt_path = _resolve_project_path(
        project_root, _require_string(reservation, "promptPath")
    )
    metadata_path = _resolve_project_path(
        project_root, _require_string(reservation, "metadataPath")
    )
    expected_metadata = prompt_path.with_name(prompt_path.name + ".meta.json")
    if metadata_path != expected_metadata:
        raise DispatchError("translator invocation metadata path is not prompt-adjacent")
    metadata = _load_json_object(metadata_path, "agent invocation metadata")
    invocation = _invocation_job_fields(metadata, prompt_path, manifest)
    expected = {
        "invocationId": invocation["invocation_id"],
        "invocationRef": invocation["invocation_ref"],
        "participantRef": invocation["participant_ref"],
        "roleExecutionRef": invocation["role_execution_ref"],
        "dutyId": invocation["duty_id"],
        "attempt": invocation["attempt"],
    }
    if any(reservation.get(key) != value for key, value in expected.items()):
        raise DispatchError("translator reservation and invocation metadata differ")
    if (
        invocation["audience"] != "translator"
        or invocation["assignment_ref"] != "translator"
        or metadata.get("dispatchKind") != "translator"
    ):
        raise DispatchError("translator invocation identity is invalid")
    result_path, worker_result_path = _translator_prompt_outputs(
        project_root, prompt_path
    )
    if result_path == worker_result_path:
        raise DispatchError(
            "translator result and standard worker result paths must differ"
        )
    provider = _require_string(execution, "provider")
    binding = execution.get("binding")
    if not isinstance(binding, Mapping):
        raise DispatchError("translator role execution has no binding")
    return WorkerJob(
        worker_id="translator",
        provider=provider,
        backend=options.default_backend,
        project_root=project_root,
        model_execution_value=_require_string(binding, "resolvedExecutionValue"),
        wrapper_path=_resolve_wrapper(provider, workspace_root, options),
        prompt_path=prompt_path,
        result_path=result_path,
        worker_result_path=worker_result_path,
        completion_paths=(result_path, worker_result_path),
        worktree_path=_worktree_path(manifest, active_context),
        role="translator",
        idle_timeout_seconds=options.idle_timeout_seconds,
        dispatch_kind="translator",
        **invocation,
    )


def _translator_prompt_outputs(
    project_root: Path, prompt_path: Path,
) -> tuple[Path, Path]:
    try:
        lines = prompt_path.read_text(encoding="utf-8").splitlines()
    except OSError as exc:
        raise DispatchError(f"cannot read translator prompt: {exc}") from exc
    values: list[str] = []
    for header in ("Result Path", "Worker Result Path"):
        prefix = f"**{header}:** "
        matches = [line[len(prefix):].strip() for line in lines if line.startswith(prefix)]
        if len(matches) != 1 or not matches[0]:
            raise DispatchError(
                f"translator prompt requires exactly one non-empty {prefix.strip()} header"
            )
        values.append(matches[0])
    return (
        _resolve_project_path(project_root, values[0]),
        _resolve_project_path(project_root, values[1]),
    )


def _select_cli_wrapper_assignments(
    recommended: Sequence[str],
    requested_workers: Sequence[str],
    team_state: Mapping[str, Any],
    options: _BuildOptions,
) -> list[str]:
    dispatchable = {
        worker_id
        for worker_id in recommended
        if _is_cli_wrapper_assignment(team_state, worker_id, options)
    }
    if not requested_workers:
        selected = [
            worker
            for worker in recommended
            if worker in dispatchable and worker != REPORT_WRITER_WORKER_ID
        ]
        if selected:
            return selected
        raise DispatchError("run roster has no cli-wrapper assignments for codex-dispatch")
    selected = list(requested_workers)
    unknown = [worker for worker in selected if worker not in recommended]
    if unknown:
        raise DispatchError(
            "requested worker(s) are not in this run roster: " + ", ".join(unknown)
        )
    native = [
        worker
        for worker in selected
        if _runner_for_worker(team_state, worker) == "native-session"
    ]
    if native:
        raise DispatchError(
            "requested worker(s) use runner=native-session and must be dispatched "
            "by the current Codex host: " + ", ".join(native)
        )
    unsupported = [worker for worker in selected if worker not in dispatchable]
    if unsupported:
        supported = ", ".join(sorted(dispatchable))
        raise DispatchError(
            "unsupported Codex lead worker(s): "
            + ", ".join(unsupported)
            + f" (CLI dispatcher supports: {supported})"
        )
    return selected


def _is_cli_wrapper_assignment(
    team_state: Mapping[str, Any],
    worker_id: str,
    options: _BuildOptions,
) -> bool:
    state = _worker_state(team_state, worker_id)
    provider = _string_value(state.get("provider")) or _provider_for_worker(
        worker_id,
        options,
    )
    return (
        _string_value(state.get("runner")) or BACKEND_CLI_WRAPPER
    ) == BACKEND_CLI_WRAPPER and provider in options.supported_worker_wrappers


def _runner_for_worker(team_state: Mapping[str, Any], worker_id: str) -> str:
    state = _worker_state(team_state, worker_id)
    return _string_value(state.get("runner")) or BACKEND_CLI_WRAPPER


def _job_from_roster_worker(
    project_root: Path,
    workspace_root: Path,
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any],
    fact: _RosterWorkerFacts,
    prompt_path: Path,
    options: _BuildOptions,
) -> WorkerJob:
    result_path = _resolve_project_path(project_root, fact.result_path)
    metadata = _agent_metadata(prompt_path)
    invocation = _invocation_job_fields(metadata, prompt_path, manifest)
    return WorkerJob(
        worker_id=fact.worker_id,
        provider=fact.provider,
        backend=options.default_backend,
        project_root=project_root,
        model_execution_value=fact.model,
        wrapper_path=_resolve_wrapper(fact.provider, workspace_root, options),
        prompt_path=prompt_path,
        result_path=_result_path_for_worker(
            fact.worker_id, result_path, manifest, project_root
        ),
        worker_result_path=result_path,
        completion_paths=_completion_paths(
            fact.worker_id, result_path, manifest, project_root
        ),
        worktree_path=_worktree_path(manifest, active_context),
        role=fact.role,
        idle_timeout_seconds=options.idle_timeout_seconds,
        dispatch_kind=options.dispatch_kind,
        session_id=_dispatch_session_id(fact.provider),
        **invocation,
    )


def _agent_metadata(prompt_path: Path) -> Mapping[str, Any] | None:
    metadata_path = prompt_path.with_name(prompt_path.name + ".meta.json")
    if not metadata_path.is_file():
        return None
    return _load_json_object(metadata_path, "agent invocation metadata")


def _invocation_job_fields(
    metadata: Mapping[str, Any] | None,
    prompt_path: Path,
    manifest: Mapping[str, Any],
) -> dict[str, Any]:
    if metadata is None:
        return {}
    digests = metadata.get("digests")
    assignment = metadata.get("modelAssignment")
    if not isinstance(digests, Mapping) or not isinstance(assignment, Mapping):
        raise DispatchError("agent invocation metadata is incomplete")
    host_model_value = assignment.get("hostModelValue")
    if host_model_value is not None and not isinstance(host_model_value, str):
        raise DispatchError("agent invocation hostModelValue is invalid")
    fields = {
        "invocation_id": _require_string(metadata, "invocationId"),
        "audience": _require_string(metadata, "audience"),
        "assignment_ref": _require_string(metadata, "assignmentRef"),
        "prompt_metadata_path": prompt_path.with_name(
            prompt_path.name + ".meta.json"
        ),
        "catalog_digest": _require_string(digests, "catalogDigest"),
        "assignment_digest": _require_string(digests, "assignmentDigest"),
        "duty_digest": _require_string(digests, "dutyDigest"),
        "instruction_digest": _require_string(digests, "instructionDigest"),
        "prompt_digest": _require_string(digests, "promptDigest"),
        "host_model_value": host_model_value,
        "enforcement_mode": CLI_DISPATCH_ENFORCEMENT_MODE,
    }
    schema = metadata.get("schemaVersion")
    identity_version = metadata.get("executionIdentityVersion")
    if schema == 1 and identity_version is None:
        return fields
    if schema != "2.0" or identity_version != 2:
        raise DispatchError("agent invocation metadata mixes v1 and v2 identity")
    participant_ref = _require_string(metadata, "participantRef")
    role_execution_ref = _require_string(metadata, "roleExecutionRef")
    role_executions = manifest.get("roleExecutions")
    execution = next((
        row for row in role_executions
        if isinstance(row, Mapping)
        and row.get("roleExecutionRef") == role_execution_ref
    ), None) if isinstance(role_executions, list) else None
    if execution is None or execution.get("participantRef") != participant_ref:
        raise DispatchError("agent invocation role execution does not match run manifest")
    execution_label = _require_string(metadata, "executionLabel")
    if execution.get("executionLabel") != execution_label:
        raise DispatchError("agent invocation execution label does not match run manifest")
    attempt = metadata.get("attempt")
    if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
        raise DispatchError("agent invocation v2 attempt must be positive")
    fields.update({
        "participant_ref": participant_ref,
        "role_execution_ref": role_execution_ref,
        "execution_label": execution_label,
        "duty_id": _require_string(metadata, "dutyId"),
        "invocation_ref": _require_string(metadata, "invocationRef"),
        "attempt": attempt,
    })
    return fields


def _materialize_roster_prompts(
    project_root: Path,
    workspace_root: Path,
    manifest_path: Path,
    manifest: Mapping[str, Any],
    team_state: Mapping[str, Any],
    selected: Sequence[str],
    options: _BuildOptions,
) -> tuple[tuple[_RosterWorkerFacts, ...], dict[str, Path]]:
    worker_facts = tuple(
        _roster_worker_facts(team_state, worker_id, options)
        for worker_id in selected
    )
    try:
        prompt_paths = materialize_initial_prompts(
            InitialPromptMaterializationRequest(
                project_root=project_root,
                run_manifest_path=manifest_path,
                runtime_root=workspace_root,
                delivery_mode=PromptDeliveryMode.EAGER_INCLUDE,
                workers=tuple(
                    InitialPromptWorkerRequest(fact.worker_id, fact.model)
                    for fact in worker_facts
                ),
            )
        )
    except InitialPromptMaterializationError as exc:
        raise DispatchError(f"{exc.reason}: {exc}") from exc
    return worker_facts, prompt_paths


def _roster_worker_facts(
    team_state: Mapping[str, Any],
    worker_id: str,
    options: _BuildOptions | None = None,
) -> _RosterWorkerFacts:
    state = _worker_state(team_state, worker_id)
    return _RosterWorkerFacts(
        worker_id=worker_id,
        model=_require_string(state, "modelExecutionValue"),
        provider=_string_value(state.get("provider"))
        or _provider_for_worker(worker_id, options),
        result_path=_require_string(state, "resultPath"),
        role=_string_value(state.get("role")) or "worker",
    )


def _jobs_from_file(
    project_root: Path,
    workspace_root: Path,
    jobs_file: Path | None,
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any],
    options: _BuildOptions,
) -> list[WorkerJob]:
    if jobs_file is None:
        return []
    return _worker_jobs_from_file(
        project_root,
        jobs_file,
        manifest=manifest,
        active_context=active_context,
        backend=options.default_backend,
        idle_timeout_seconds=options.idle_timeout_seconds,
        default_dispatch_kind=options.dispatch_kind,
        resolve_wrapper=lambda provider: _resolve_wrapper(
            provider, workspace_root, options
        ),
        default_provider=lambda worker_id: _provider_for_worker(worker_id, options),
    )


def _spawn_job(
    plan: DispatchPlan,
    job: WorkerJob,
    attempt: int,
    *,
    batch_artifact_paths: Sequence[Path] = (),
) -> WorkerHandle:
    job = _prepare_job_attempt(plan, job, attempt)
    # Everything between recording the attempt and starting the worker runs
    # before any worker process exists. A failure here used to leave the attempt
    # `started` forever: the manifest then refused attempt 1 again ("next
    # attempt must be 2") and refused attempt 2 as well, because the prompt
    # metadata still said attempt 1. The invocation had no way forward and the
    # lead had to mint a new invocation id and prompt path to escape.
    try:
        contract = _persisted_write_contract(plan, job)
        policies = (contract[0],) if contract else ()
        snapshot = None
        if contract is not None and contract[1].mutation_audit == "batch":
            snapshot_path = _mutation_snapshot_path(job)
            if snapshot_path.is_file():
                snapshot = MutationSnapshot.from_payload(
                    _load_json_object(snapshot_path, "mutation audit snapshot")
                )
            else:
                snapshot = _mutation_snapshot(
                    plan,
                    policies,
                    (job,),
                    round_artifact_paths=batch_artifact_paths,
                )
    except Exception:
        _abandon_unstarted_attempt(plan, job, attempt)
        raise
    handle = replace(
        _start_job(plan, job),
        mutation_snapshot=snapshot,
        mutation_policies=policies,
        write_policy_digest=contract[0].digest if contract else "",
        write_enforcement=contract[1] if contract else None,
    )
    _record_dispatch(plan.team_state_path, handle, attempt, "running", "")
    _append_event(plan, "worker-dispatched", _attempt_details(job, attempt, handle))
    if handle.completed_process is not None:
        outcome = _outcome_from_completed(handle)
        retry_allowed = _finish_attempt(
            plan,
            job,
            attempt,
            outcome,
            snapshot=snapshot,
            policies=policies,
        )
        handle = replace(handle, mutation_retry_allowed=retry_allowed)
    return handle


def _spawn_cli_job_nonblocking(
    plan: DispatchPlan, job: WorkerJob, attempt: int,
) -> WorkerHandle:
    plan = _ensure_runtime_chain(plan)
    try:
        runtime = open_worker(
            _chain_starting_at(plan.runtime_chain, job.backend),
            lambda surface: _spawn_request(plan, job, surface),
        )
    except EnvironmentBlocked as exc:
        raise DispatchError(str(exc)) from exc
    except (SurfaceUnavailable, OSError) as exc:
        raise DispatchError(str(exc)) from exc
    return WorkerHandle(
        job,
        "",
        None,
        status_path_for_prompt(job.prompt_path),
        "",
        runtime_handle=runtime,
    )


def _next_attempt(plan: DispatchPlan, job: WorkerJob) -> int:
    """이 invocation 이 다음에 청구할 attempt 번호.

    attempt 를 1 로 고정하면 같은 워커를 두 번째로 디스패치할 수 없다. 원장은
    단조 증가를 요구하므로(`execution_manifest._validate_next_attempt`) 두 번째
    호출이 항상 `next attempt must be 2` 로 거부되고, 재시도 예산이 계약에는
    있는데 그 경로에는 쓸 수단이 없는 상태가 된다. 예산은 프로세스가 아니라
    invocation 에 붙어 있고, 그 잔액이 적힌 곳은 원장뿐이다.
    """
    if not job.has_execution_identity:
        return job.attempt
    manifest = read_execution_manifest(plan.manifest_path)
    prior = [
        row.attempt for row in manifest.attempts
        if row.invocation_ref == job.invocation_ref
    ]
    if not prior:
        return job.attempt
    spent = max(prior)
    if spent >= MAX_WORKER_ATTEMPTS:
        raise DispatchError(
            f"worker retry budget is spent: {job.invocation_ref} used "
            f"{spent} of {MAX_WORKER_ATTEMPTS} attempts. Materialize a new "
            "invocation to dispatch this worker again."
        )
    return spent + 1


def _prepare_job_attempt(
    plan: DispatchPlan, job: WorkerJob, attempt: int
) -> WorkerJob:
    if job.has_execution_identity:
        _revalidate_job_role_execution(plan, job)
        job = _job_for_attempt(job, attempt)
        _record_execution_attempt(plan, job)
    _transition_job_status(
        plan,
        job,
        "in-progress",
        "",
        model_execution_value=job.model_execution_value,
    )
    return job


def _persisted_write_contract(
    plan: DispatchPlan, job: WorkerJob
) -> tuple[WritePolicy, WriteEnforcement] | None:
    if not job.has_execution_identity:
        return None
    manifest = read_execution_manifest(plan.manifest_path)
    invocation = next(
        (
            row for row in manifest.invocations
            if row.invocation_ref == job.invocation_ref
        ),
        None,
    )
    if invocation is None:
        raise DispatchError("worker invocation write policy was not persisted")
    try:
        enforcement = write_enforcement_from_payload(
            invocation.write_enforcement
        )
        policy = replace(
            write_policy_from_payload(invocation.write_policy),
            maximum_boundary_precision=enforcement.boundary_precision,
        )
    except (TypeError, ValueError) as exc:
        raise DispatchError(f"persisted worker write policy is invalid: {exc}") from exc
    if policy.digest != invocation.write_policy_digest:
        raise DispatchError("persisted worker write policy digest changed")
    return policy, enforcement


def _mutation_snapshot(
    plan: DispatchPlan,
    policies: Sequence[WritePolicy],
    jobs: Sequence[WorkerJob],
    round_artifact_paths: Sequence[Path] = (),
) -> MutationSnapshot:
    sidecars = tuple(_mutation_snapshot_path(job) for job in jobs)
    snapshot = ExecutionMutationAudit().snapshot(
        policies,
        orchestrator_paths=(
            *round_artifact_paths,
            *_run_errors_log_path(plan),
            *_orchestrator_artifact_roots(plan),
            plan.manifest_path,
            plan.team_state_path,
            Path(f"{plan.team_state_path}.lock"),
            plan.lead_events_path,
            plan.lead_events_path.with_name(
                f".{plan.lead_events_path.name}.lock"
            ),
            *sidecars,
        ),
    )
    for path in sidecars:
        write_json_atomic(path, snapshot.to_payload())
    return snapshot


def _mutation_snapshot_path(job: WorkerJob) -> Path:
    return mutation_snapshot_path_for_prompt(job.prompt_path)


def _record_execution_attempt(plan: DispatchPlan, job: WorkerJob) -> None:
    task_key = _require_string(plan.manifest, "taskKey")
    write_policy, write_enforcement = _canonical_write_contract(plan, job)
    record_invocation_attempt(
        plan.manifest_path,
        Invocation(
            invocation_ref=job.invocation_ref,
            participant_ref=job.participant_ref,
            role_execution_ref=job.role_execution_ref,
            source_invocation_ref=None,
            recovery_ref=None,
            duty_id=job.duty_id,
            dispatch_kind=job.dispatch_kind,
            round=_dispatch_round(job.dispatch_kind),
            input_digest=job.prompt_digest,
            write_policy=write_policy.to_payload(),
            write_policy_digest=write_policy.digest,
            write_enforcement=write_enforcement.to_payload(),
        ),
        Attempt(
            invocation_ref=job.invocation_ref,
            attempt=job.attempt,
            started_at=_utc_now(),
            finished_at=None,
            status="started",
            result_path=None,
            error_path=None,
            change_summary={},
            evidence_seal_ref=None,
        ),
        task_key=task_key,
    )


def _canonical_write_contract(
    plan: DispatchPlan, job: WorkerJob
) -> tuple[WritePolicy, WriteEnforcement]:
    manifest = read_execution_manifest(plan.manifest_path)
    execution = next(
        (
            row for row in manifest.role_executions
            if row.role_execution_ref == job.role_execution_ref
        ),
        None,
    )
    if execution is None or execution.binding is None:
        raise DispatchError("worker write policy has no role execution binding")
    capability = execution.binding.worker_write_capability
    if capability is None:
        raise DispatchError("worker write policy has no runner write capability")
    artifacts = _worker_artifact_paths(plan, job)
    worktree = Path(job.worktree_path) if job.worktree_path else None
    planned = (
        planned_paths_from_run_manifest(plan.project_root, plan.manifest)
        if execution.role == "implementer"
        else ((), True)
    )
    try:
        return build_invocation_write_contract(
            role=execution.role,
            project_root=plan.project_root,
            worktree=worktree,
            artifact_paths=artifacts,
            maximum_precision=capability.max_boundary_precision,
            planned_paths=planned[0],
            planned_paths_declared=planned[1],
            auxiliary_roots=verifier_extra_dirs(execution.role),
            validated_auxiliary_roots=verifier_extra_dirs(execution.role),
        )
    except (OSError, ValueError) as exc:
        raise DispatchError(f"worker write policy is invalid: {exc}") from exc


def _worker_artifact_paths(plan: DispatchPlan, job: WorkerJob) -> tuple[Path, ...]:
    active_context = _load_optional_json(
        plan.project_root, plan.manifest.get("activeRunContextPath")
    )
    paths = {
        job.result_path,
        job.worker_result_path,
        *job.completion_paths,
        Path(audit_sidecar_rel(str(job.worker_result_path))),
        # The prompt's three derived files are written together and belong in
        # one list. The audit snapshot used to be listed only for the job whose
        # snapshot it was, so a sibling's snapshot — written by okstra as that
        # sibling started — landed inside this worker's window as an
        # unauthorized artifact-root change.
        *prompt_derived_paths(job.prompt_path),
    }
    error_logs = active_context.get("errorLogs")
    if isinstance(error_logs, Mapping):
        sidecars = error_logs.get("sidecarsByWorkerId")
        if isinstance(sidecars, Mapping):
            value = _string_value(sidecars.get(job.worker_id))
            if value:
                paths.add(_resolve_project_path(plan.project_root, value))
    return tuple(sorted(paths, key=str))


def _run_errors_log_path(plan: DispatchPlan) -> tuple[Path, ...]:
    """The run-level errors log, which the lead appends to while workers run.

    The lead contract requires it to record an observed worker failure as soon
    as it happens, so a worker still running at that moment sees the write. It
    is a lead-owned run artifact like the manifest and the lead events log, and
    is listed for the same reason.
    """
    active_context = _load_optional_json(
        plan.project_root, plan.manifest.get("activeRunContextPath")
    )
    error_logs = active_context.get("errorLogs")
    if not isinstance(error_logs, Mapping):
        return ()
    value = _string_value(error_logs.get("runErrorsLogPath"))
    if not value:
        return ()
    return (_resolve_project_path(plan.project_root, value),)




def _job_for_attempt(job: WorkerJob, attempt: int) -> WorkerJob:
    if attempt == job.attempt:
        return job
    try:
        retry = materialize_retry_invocation(
            job.prompt_metadata_path,
            project_root=job.project_root,
            attempt=attempt,
        )
    except AgentInvocationError as exc:
        raise DispatchError(f"cannot materialize retry invocation: {exc}") from exc
    if retry.invocation_ref != job.invocation_ref:
        raise DispatchError("retry invocation reference changed")
    return replace(
        job,
        prompt_path=retry.prompt_path,
        prompt_metadata_path=retry.metadata_path,
        invocation_id=retry.invocation_id,
        attempt=retry.attempt,
    )


def _dispatch_round(dispatch_kind: str) -> int:
    prefix = "reverify-r"
    if not dispatch_kind.startswith(prefix):
        return 1
    suffix = dispatch_kind[len(prefix):]
    if not suffix.isdigit() or int(suffix) < 1:
        raise DispatchError(f"invalid reverify dispatch kind: {dispatch_kind}")
    return int(suffix)


def _dispatch_job_with_retry(plan: DispatchPlan, job: WorkerJob) -> int:
    for attempt in range(_next_attempt(plan, job), MAX_WORKER_ATTEMPTS + 1):
        handle = _spawn_job(plan, job, attempt)
        if handle.completed_process is None:
            return await_dispatches(plan, timeout_seconds=None)
        outcome = _outcome_from_completed(handle)
        if _job_terminal_status(plan, job, attempt) == "completed":
            return 0
        if _should_retry(outcome, attempt) and handle.mutation_retry_allowed:
            _append_event(plan, "worker-retry-scheduled", _retry_details(job, attempt, outcome))
            job = _job_for_next_attempt(job)
            continue
        return outcome.returncode or 1
    return 1


def _ensure_runtime_chain(plan: DispatchPlan) -> DispatchPlan:
    if plan.runtime_chain:
        return plan
    from .adapters.runtime.assembly import runtime_chain as build_chain
    return replace(plan, runtime_chain=build_chain(plan.default_backend))


def _spawn_request(
    plan: DispatchPlan, job: WorkerJob, surface: str
) -> WorkerSpawnRequest:
    targeted = replace(job, backend=surface)
    title = (
        targeted.execution_label
        if targeted.has_execution_identity
        else f"{targeted.worker_id}-worker"
    )
    return WorkerSpawnRequest(
        command=tuple(targeted.command),
        cwd=plan.project_root,
        title=title,
        owned_surface_ids=_opened_cmux_surfaces(plan.team_state_path),
    )


def _chain_starting_at(
    chain: tuple[WorkerRuntimePort, ...], surface: str
) -> tuple[WorkerRuntimePort, ...]:
    # A retry job is already cli-wrapper; do not walk back up to cmux-pane.
    for index, port in enumerate(chain):
        if port.surface == surface:
            return chain[index:]
    raise DispatchError(f"unsupported worker backend: {surface}")


def _start_job(plan: DispatchPlan, job: WorkerJob) -> WorkerHandle:
    plan = _ensure_runtime_chain(plan)
    try:
        runtime = open_worker(
            _chain_starting_at(plan.runtime_chain, job.backend),
            lambda surface: _spawn_request(plan, job, surface),
        )
    except EnvironmentBlocked as exc:
        raise DispatchError(str(exc)) from exc
    except SurfaceUnavailable as exc:
        raise DispatchError(str(exc)) from exc
    degraded_from = job.backend if runtime.surface != job.backend else ""
    settled = replace(job, backend=runtime.surface) if degraded_from else job
    completed = None
    if runtime.waitable:
        from .adapters.runtime.assembly import port_for
        returncode = port_for(plan.runtime_chain, runtime.surface).wait(runtime)
        completed = subprocess.CompletedProcess(list(settled.command), returncode)
    pane_id = runtime.identifier if runtime.surface == SURFACE_CMUX_PANE else ""
    return WorkerHandle(
        settled,
        pane_id,
        completed,
        status_path_for_prompt(settled.prompt_path),
        degraded_from,
        runtime_handle=runtime,
    )


def _opened_cmux_surfaces(team_state_path: Path) -> tuple[str, ...]:
    """The surfaces this run has already opened in the user's cmux workspace.

    Placement needs them to tell okstra's own panes from the rest of the
    workspace, which belongs to the user and may hold another agent session or
    a shell okstra must not split or stack into. A degraded dispatch opened no
    surface and records an empty id, which would name no pane at all.
    """
    payload = _load_json_object(team_state_path, "team-state")
    dispatches = payload.get("workerDispatches")
    if not isinstance(dispatches, list):
        return ()
    opened = []
    for record in dispatches:
        if not isinstance(record, dict):
            continue
        surface_id = str(record.get("paneId", ""))
        if surface_id:
            opened.append(surface_id)
    return tuple(opened)


def _outcome_from_completed(handle: WorkerHandle) -> WorkerOutcome:
    completed = handle.completed_process
    if completed is None:
        raise DispatchError("completed process missing")
    return WorkerOutcome(
        returncode=completed.returncode,
        missing_completion_paths=_missing_completion_paths(handle.job),
        pane_id=handle.pane_id,
        status_sidecar_path=handle.status_sidecar_path,
        timeout=False,
        terminal_stage="exited",
        degraded_from=handle.degraded_from,
    )


def _correct_teardown_marked_dispatches(plan: DispatchPlan) -> None:
    """Let the wrapper's own exit settle a record teardown wrote off.

    `okstra team teardown` marks every non-terminal dispatch `error` — right for a
    worker it killed, wrong for one that had already exited while its record was
    still `running`, which happens when the lead awaited by polling artifacts
    instead of `team await` (the component that transitions the record). The run
    then reports `error` for workers whose wrapper exited 0 with every output in
    place, and nothing could undo it: teardown's `error` drops the record out of
    the running set this function's caller scans, so a later await never saw it.

    Only a success is corrected, and only from process-level evidence — the
    wrapper's terminal status plus its exit code, judged by the same
    `_outcome_from_status` path every other dispatch goes through. A non-zero exit
    or a missing output stays an error, and no retry is spawned: teardown has
    already taken the panes, so re-running here would start work nobody is
    watching.
    """
    for record in _teardown_marked_dispatches(plan.team_state_path):
        status_path = _optional_path(record.get("statusSidecarPath"))
        status = read_wrapper_status(status_path) if status_path else None
        if status is None or not status.is_terminal:
            continue
        outcome = _outcome_from_status(record, status)
        if outcome.returncode != 0 or outcome.missing_completion_paths or outcome.timeout:
            continue
        _finish_record(plan, record, outcome)


def _advance_running_dispatches(plan: DispatchPlan, records: Sequence[Mapping[str, Any]]) -> None:
    for record in records:
        status_path = _optional_path(record.get("statusSidecarPath"))
        status = read_wrapper_status(status_path) if status_path else None
        if status is None or not status.is_terminal:
            continue
        outcome = _outcome_from_status(record, status)
        retry_allowed = _finish_record(plan, record, outcome)
        if (
            _should_retry(outcome, int(record.get("attempt", 1)))
            and retry_allowed
        ):
            _retry_from_record(plan, record, outcome)


def _retry_from_record(
    plan: DispatchPlan, record: Mapping[str, Any], outcome: WorkerOutcome
) -> None:
    attempt = int(record.get("attempt", 1))
    job = _job_from_record(plan.project_root, record)
    reason = "required worker artifact was not produced"
    _update_dispatch_status(plan.team_state_path, job, attempt, "error", reason)
    # `team-contract` counts the first attempt's failure as a recorded
    # `cli-failure`; a retry that succeeds settles `completed` and would
    # otherwise leave no trace that anything had to be re-run.
    details = _event_execution_identity(job, attempt)
    details["errorLogAppend"] = _record_wrapper_failure(
        plan, job, attempt, outcome, reason
    )
    _append_event(plan, "worker-retry-scheduled", details)
    # The restored job settles attempt 1 above and must keep its session; only
    # the attempt being opened here gets a new one.
    _spawn_job(plan, _job_for_next_attempt(job), attempt + 1)


# 사유 한 줄에 실을 경로 개수. 전부 실으면 team-state 행이 부풀고, 하나도 안
# 실으면 읽는 쪽이 다른 파일을 뒤져야 한다.
_MUTATION_REASON_PATH_LIMIT = 5


def _mutation_failure_reason(mutation) -> str:
    """왜 실패했는지와 무엇이 바뀌었는지를 한 줄에 담는다.

    종전에는 디스패치 행에 `readonly source changed` 만 남고 바뀐 경로는
    lead-events 에만 있었다. 읽는 쪽이 그 파일의 존재를 알아야 원인에 닿을 수
    있었고, 실제로 두 번의 진단이 같은 자리에서 빗나갔다 — 한 번은 감사를 손으로
    재구현해서, 한 번은 사용자 편집을 의심해서. 감사는 답을 이미 계산해 두었으니
    사유 옆에 놓는다.
    """
    reason = "; ".join(mutation.violations) or mutation.status
    paths = list(mutation.changed_paths)
    if not paths:
        return reason
    shown = ", ".join(paths[:_MUTATION_REASON_PATH_LIMIT])
    remaining = len(paths) - _MUTATION_REASON_PATH_LIMIT
    if remaining > 0:
        shown += f", +{remaining} more"
    return f"{reason} [changed: {shown}]"


def _finish_attempt(
    plan: DispatchPlan,
    job: WorkerJob,
    attempt: int,
    outcome: WorkerOutcome,
    *,
    snapshot: MutationSnapshot | None = None,
    policies: Sequence[WritePolicy] = (),
) -> bool:
    mutation = _audit_attempt(job, outcome, snapshot, policies)
    if mutation is not None and mutation.status in {
        "contract-failed-unattributed",
        "mutation-present-unresolved",
    }:
        reason = _mutation_failure_reason(mutation)
        _transition_job_status(plan, job, "error", reason)
        _update_dispatch_status(
            plan.team_state_path, job, attempt, "error", reason
        )
        details = _failure_details(job, attempt, outcome, reason)
        details["mutationAudit"] = mutation.change_summary()
        _append_event(plan, "worker-failed", details)
        _finish_manifest_attempt(plan, job, attempt, mutation, None)
        return False
    settlement = _settle(plan, job, attempt, outcome)
    if settlement.completed:
        result_link = _link_result(plan, job, attempt)
        _transition_job_status(plan, job, "completed", "")
        _update_dispatch_status(
            plan.team_state_path,
            job,
            attempt,
            "completed",
            "; ".join(part for part in (settlement.note, result_link["reason"]) if part),
        )
        details = _result_details(job, attempt, outcome)
        details["resultLink"] = result_link
        if settlement.error_log_append is not None:
            details["errorLogAppend"] = settlement.error_log_append
        if mutation is not None:
            details["mutationAudit"] = mutation.change_summary()
        _append_event(plan, "worker-result-collected", details)
        _finish_manifest_attempt(
            plan, job, attempt, mutation, job.worker_result_path
        )
        return False
    reason = settlement.reason
    status = "timeout" if outcome.timeout else "error"
    _transition_job_status(plan, job, status, reason)
    _update_dispatch_status(plan.team_state_path, job, attempt, status, reason)
    details = _failure_details(job, attempt, outcome, reason)
    if settlement.error_log_append is not None:
        details["errorLogAppend"] = settlement.error_log_append
    if mutation is not None:
        details["mutationAudit"] = mutation.change_summary()
    _append_event(plan, "worker-failed", details)
    _finish_manifest_attempt(plan, job, attempt, mutation, None)
    return mutation.retry_allowed if mutation is not None else True


def _audit_attempt(
    job: WorkerJob,
    outcome: WorkerOutcome,
    snapshot: MutationSnapshot | None,
    policies: Sequence[WritePolicy],
) -> MutationAuditResult | None:
    if not job.has_execution_identity or not policies:
        return None
    succeeded = outcome.returncode == 0 and not outcome.timeout
    result_present = not outcome.missing_completion_paths
    if snapshot is None:
        status = "ok" if succeeded and result_present else "failed-no-mutation"
        return MutationAuditResult(
            status=status,
            changed_paths=(),
            source_changed=False,
            git_changed=False,
            attribution="call",
            retry_allowed=status == "failed-no-mutation",
            violations=(),
            before_digest="",
            after_digest="",
            git_projection={},
        )
    return ExecutionMutationAudit().compare(
        snapshot,
        policies,
        out_of_plan_edits=_out_of_plan_edit_paths(
            job.result_path, project_root=job.project_root
        ),
        attempt_succeeded=succeeded,
        result_present=result_present,
    )


def _declared_out_of_plan_paths(result_path: Path) -> tuple[str, ...]:
    """워커가 자기 결과의 `Out-of-plan edits` 블록에 선언한 경로.

    감사는 워커가 끝나는 시점에 돈다. 그때 디스크에 있는 것은 워커 결과
    마크다운뿐이고, `implementation.outOfPlanEdits` 는 리드가 나중에 쓰는 최종
    리포트의 필드다. JSON 형태만 읽었기 때문에 executor 의 선언이 한 번도 보이지
    않았고, 계약대로 선언한 편집까지 미허가 변경으로 집계됐다. 블록의 형태는 이
    판독기를 위해 `_implementation-executor.md` 가 고정한다 — `- ` 줄마다 첫 백틱
    토큰이 경로다.
    """
    if not result_path.is_file() or result_path.suffix != ".md":
        return ()
    try:
        text = result_path.read_text(encoding="utf-8")
    except (OSError, UnicodeError):
        return ()
    paths: list[str] = []
    inside = False
    for line in text.splitlines():
        stripped = line.strip()
        if stripped.startswith("#"):
            inside = stripped.lstrip("#").strip().lower() == "out-of-plan edits"
            continue
        if not inside or not stripped.startswith("- "):
            continue
        quoted = re.findall(r"`([^`\n]+)`", stripped)
        if quoted and quoted[0].strip():
            paths.append(quoted[0].strip())
    return tuple(paths)


def _out_of_plan_edit_paths(
    result_path: Path, *, project_root: Path | None = None
) -> tuple[str, ...]:
    if result_path.suffix != ".json":
        return _declared_out_of_plan_paths(result_path)
    if not result_path.is_file():
        return ()
    if project_root is None:
        raise ValueError("JSON worker result requires project_root")
    try:
        # 외부 입력: 워커 모델이 게시한 결과 봉투를 디스패처가 수락 전에 판독한다.
        payload = load_external_json(
            external_worker_json_source(
                result_path,
                trusted_root=project_root,
                lane_root=result_path.parent,
            ),
            artifact="worker result",
        )
    except JsonBoundaryError:
        return ()
    implementation = payload.get("implementation") if isinstance(payload, Mapping) else None
    rows = implementation.get("outOfPlanEdits") if isinstance(implementation, Mapping) else None
    if not isinstance(rows, list):
        return ()
    return tuple(
        str(row["file"])
        for row in rows
        if isinstance(row, Mapping)
        and isinstance(row.get("file"), str)
        and row["file"]
    )


def _abandon_unstarted_attempt(
    plan: DispatchPlan, job: WorkerJob, attempt: int
) -> None:
    """Close an attempt whose worker never started, so a retry can follow it.

    `failed-no-mutation` is the truthful status: the dispatch died before the
    worker process existed, so nothing wrote anything. It is also the only
    terminal status the manifest lets another attempt follow.
    """
    if not job.has_execution_identity:
        return
    try:
        finish_attempt_mutation(
            plan.manifest_path,
            invocation_ref=job.invocation_ref,
            attempt=attempt,
            finished_at=_utc_now(),
            status="failed-no-mutation",
            result_path=None,
            error_path=None,
            change_summary={},
            task_key=_require_string(plan.manifest, "taskKey"),
        )
    except (ExecutionManifestError, DispatchError, OSError):
        # The original dispatch failure is what the caller needs to see; a
        # manifest that cannot be closed here is reported by the next read.
        return


def _finish_manifest_attempt(
    plan: DispatchPlan,
    job: WorkerJob,
    attempt: int,
    mutation: MutationAuditResult | None,
    result_path: Path | None,
) -> None:
    if mutation is None:
        return
    manifest = read_execution_manifest(plan.manifest_path)
    existing = next(
        (
            row for row in manifest.attempts
            if row.invocation_ref == job.invocation_ref
            and row.attempt == attempt
        ),
        None,
    )
    expected_result = str(result_path) if result_path is not None else None
    if existing is not None and existing.finished_at is not None:
        if (
            existing.status == mutation.status
            and existing.result_path == expected_result
            and dict(existing.change_summary) == mutation.change_summary()
        ):
            return
        raise DispatchError("attempt terminal mutation result drift")
    finish_attempt_mutation(
        plan.manifest_path,
        invocation_ref=job.invocation_ref,
        attempt=attempt,
        finished_at=_utc_now(),
        status=mutation.status,
        result_path=expected_result,
        error_path=None,
        change_summary=mutation.change_summary(),
        task_key=_require_string(plan.manifest, "taskKey"),
    )
    _seal_finished_attempt(plan, job, attempt, mutation)


def _seal_finished_attempt(plan, job, attempt, mutation) -> None:
    from .attempt_evidence import TerminalObservation, finalize_attempt

    events = job.prompt_path.with_name(job.prompt_path.name + ".host-events.jsonl")
    if not events.is_file() or not job.has_execution_identity:
        return
    artifacts = tuple(
        path for path in (job.result_path, job.worker_result_path) if path
    )
    try:
        finalize_attempt(
            plan.manifest_path,
            _build_dispatch_id(job.invocation_ref, attempt),
            TerminalObservation(
                status=mutation.status,
                source_diff_digest=mutation.change_summary().get("beforeDigest", ""),
                git_projection_digest=mutation.change_summary().get("afterDigest", ""),
                changed_paths=tuple(mutation.changed_paths),
                artifact_paths=artifacts,
                host_event_stream_path=events,
                project_root=plan.project_root,
                worktree=Path(job.worktree_path) if job.worktree_path else None,
            ),
            task_key=_require_string(plan.manifest, "taskKey"),
        )
    except (OSError, ValueError):
        return


def _link_result(
    plan: DispatchPlan, job: WorkerJob, attempt: int
) -> dict[str, Any]:
    """Bind this result to its verified dispatch, reporting rather than raising.

    Linking is bookkeeping around a dispatch that has already settled, and it
    used to run before any status was written. A refused link — the live case is
    a corrective re-dispatch claiming a path the first attempt still owns — threw
    out of `_finish_attempt`, so nothing transitioned, the row stayed `running`,
    and the exception reached the caller as exit 2. The next `await` re-read the
    same terminal sidecar, re-settled the same way, and threw at the same line:
    a worker with complete artifacts wedged the run permanently, and no amount of
    waiting could clear it.

    So the settle is written either way and the refusal travels back as data. It
    is not swallowed: the reason lands in the dispatch row and in the lead event,
    and the post-hoc validators still require every accepted result to carry a
    link, so an unlinked result fails where an audit failure belongs rather than
    by stopping the run mid-phase. `agent-prompt reject-result` is the remedy the
    refusal names.
    """
    link: dict[str, Any] = {"ok": True, "reason": ""}
    if not job.invocation_id:
        return link
    try:
        _link_agent_dispatch_result(
            project_root=plan.project_root,
            run_manifest_path=plan.manifest_path,
            dispatch_id=_build_dispatch_id(job.invocation_id, attempt),
            result_path=job.worker_result_path,
        )
    except (DispatchError, OSError) as exc:
        link["ok"] = False
        link["reason"] = f"result link refused: {exc}"
    return link


def _finish_record(
    plan: DispatchPlan,
    record: Mapping[str, Any],
    outcome: WorkerOutcome,
) -> bool:
    job = _job_from_record(plan.project_root, record)
    snapshot = _snapshot_from_record(record)
    contract = _persisted_write_contract(plan, job)
    policies = (contract[0],) if contract is not None else ()
    return _finish_attempt(
        plan,
        job,
        int(record.get("attempt", 1)),
        outcome,
        snapshot=snapshot,
        policies=policies,
    )


def _snapshot_from_record(record: Mapping[str, Any]) -> MutationSnapshot | None:
    value = _string_value(record.get("mutationAuditSnapshotPath"))
    if not value:
        return None
    payload = _load_json_object(Path(value), "mutation audit snapshot")
    try:
        return MutationSnapshot.from_payload(payload)
    except (KeyError, TypeError, ValueError) as exc:
        raise DispatchError("mutation audit snapshot is invalid") from exc


@dataclass(frozen=True)
class _Settlement:
    """How one attempt's terminal status was decided."""

    completed: bool
    reason: str
    note: str
    error_log_append: dict[str, Any] | None


def _settle(
    plan: DispatchPlan, job: WorkerJob, attempt: int, outcome: WorkerOutcome
) -> _Settlement:
    """Judge an attempt by its artifacts, not by the wrapper's exit code alone.

    A wrapper can die after its worker has already written everything — an
    observed case is a connection dropped at session teardown, long after the
    result file and its audit sidecar were on disk. Settling that as `error`
    discards a complete analysis, and not figuratively: `convergence_engine`
    admits only dispatches that settled `completed`, so the worker's findings
    never reach re-verification. The lead's own re-dispatch triggers agree —
    they name a missing, unparseable, or audit-failing result, never an exit
    code — but the only signal `team await` gave was the status.

    So a non-zero exit with every completion path present is re-judged by the
    audit-sidecar contract, the same rules `okstra worker-audit-check` runs. It
    passes and the dispatch settles `completed`; it fails and the dispatch stays
    `error` exactly as before. Either way the wrapper's failure is written to the
    run error log, so a `completed` here is never a swallowed failure.
    """
    if outcome.returncode == 0 and not outcome.missing_completion_paths and not outcome.timeout:
        return _Settlement(True, "", "", None)
    reason = _failure_reason(outcome)
    completed = False
    note = ""
    if (
        outcome.returncode != SERVED_MODEL_MISMATCH_EXIT_CODE
        and not outcome.timeout
        and not outcome.missing_completion_paths
    ):
        audit_failures = _audit_sidecar_failures(job)
        if audit_failures:
            reason = (
                f"{reason}; worker artifacts failed the audit-sidecar "
                f"contract: {audit_failures[0]}"
            )
        else:
            completed = True
            note = (
                f"{reason}, but every completion artifact was written and "
                f"passed the audit-sidecar contract"
            )
    append = _record_wrapper_failure(plan, job, attempt, outcome, note or reason)
    return _Settlement(completed, "" if completed else reason, note, append)


def _audit_sidecar_failures(job: WorkerJob) -> tuple[str, ...]:
    """This worker's audit-sidecar contract failures, if the check can run.

    The check's arguments come from the result filename rather than the manifest
    so the scan cannot widen past the file this job produced: `worker-results/`
    accumulates every run's artifacts, and the `worker=` filter matches the
    `-worker`-suffixed role, not the bare provider id. A non-canonical name
    leaves nothing to enforce, and an unverifiable artifact must not be promoted
    to `completed`, so that reports one failure rather than an empty tuple.
    """
    parsed = parse_worker_result_name(job.worker_result_path.name)
    if parsed is None:
        return (
            f"worker result `{job.worker_result_path.name}` is not a canonical "
            f"`<role>-worker-<task-type>-<seq>.md` name, so the audit-sidecar "
            f"contract could not be checked",
        )
    return tuple(
        check_worker_results_audit(
            job.worker_result_path.parent.parent,
            parsed.task_type,
            parsed.seq,
            worker=parsed.worker_role,
        )
    )


def _record_wrapper_failure(
    plan: DispatchPlan,
    job: WorkerJob,
    attempt: int,
    outcome: WorkerOutcome,
    message: str,
) -> dict[str, Any]:
    """Write the wrapper's own failure to the run-level error log.

    `okstra-lead-contract` tells Lead the deterministic dispatcher records this
    and that Lead does not need to re-record it. Nothing did: no code path
    anywhere called the error-log writer, so every wrapper failure vanished, and
    with it the `instruction-set/prior-run-errors.md` digest the next run reads
    and the `/okstra-inspect errors` report. The dispatcher is the only component
    that holds the exit code, so it is the one that writes.

    Never raises. Logging is bookkeeping around a dispatch that has already
    settled; letting a rejected or unwritable record throw here would turn a
    recorded outcome into an unrecorded crash. What went wrong travels back in
    the lead event instead.
    """
    result: dict[str, Any] = {"ok": False, "reason": "", "path": ""}
    try:
        out_path = resolve_errors_log_path(
            plan.project_root,
            plan.manifest,
            _load_optional_json(
                plan.project_root, plan.manifest.get("activeRunContextPath")
            ),
        )
        result["path"] = str(out_path)
        append_observed(
            out_path=out_path,
            task_key=_string_value(plan.manifest.get("taskKey")),
            phase=_workflow_phase(plan.manifest),
            agent=_error_log_agent(job.worker_id),
            agent_role=(
                "report-writer"
                if job.worker_id == REPORT_WRITER_WORKER_ID
                else "worker"
            ),
            model=job.model_execution_value,
            error_type="cli-failure",
            command=" ".join(job.command),
            command_kind="wrapper",
            exit_code=_WRAPPER_TIMEOUT_EXIT_CODE if outcome.timeout else outcome.returncode,
            duration_ms=_wrapper_duration_ms(outcome),
            message=f"attempt {attempt}: {message}",
            stderr_excerpt=_wrapper_log_tail(job),
            context=None,
            identity=(
                _job_stored_identity(job, attempt)
                if job.has_execution_identity
                else None
            ),
        )
    except (OSError, ValueError, TypeError, WorkerPromptHeaderError) as exc:
        result["reason"] = f"{type(exc).__name__}: {exc}"
        return result
    result["ok"] = True
    return result


def _job_stored_identity(job: WorkerJob, attempt: int) -> dict[str, Any]:
    """Copy the job's stored execution refs. Do not infer them from labels."""
    return {
        "participantRef": job.participant_ref,
        "roleExecutionRef": job.role_execution_ref,
        "invocationRef": job.invocation_ref,
        "attempt": job.attempt or attempt,
        "executionLabel": job.execution_label,
    }


def _error_log_agent(worker_id: str) -> str:
    """The error log's `--agent` enum value for a worker id.

    The log's own allow-list is the authority on what it accepts; a worker whose
    name is outside it is reported as such by `append_observed` rather than
    silently rewritten into some other agent's records.
    """
    if worker_id == REPORT_WRITER_WORKER_ID:
        return REPORT_WRITER_WORKER_ID
    return f"{worker_id}-worker"


def _workflow_phase(manifest: Mapping[str, Any]) -> str:
    workflow = manifest.get("workflow")
    if isinstance(workflow, Mapping):
        return _string_value(workflow.get("currentPhase"))
    return ""


def _wrapper_duration_ms(outcome: WorkerOutcome) -> int | None:
    if outcome.status_sidecar_path is None:
        return None
    status = read_wrapper_status(outcome.status_sidecar_path)
    if status is None:
        return None
    value = status.raw.get("duration_ms")
    return value if isinstance(value, int) and not isinstance(value, bool) else None


def _wrapper_log_tail(job: WorkerJob) -> str | None:
    """The tail of the wrapper transcript, which usually names the real failure.

    The observed case put `API Error: Connection lost mid-response.` in the last
    two lines and nothing anywhere else; without it the record says only that
    some process exited 1. Capped well under the writer's own excerpt limit
    because a whole record must stay inside one atomic `PIPE_BUF` append.
    """
    log_path = log_path_for_prompt(job.prompt_path)
    try:
        with log_path.open("rb") as handle:
            handle.seek(0, 2)
            handle.seek(max(0, handle.tell() - _WRAPPER_LOG_TAIL_BYTES))
            tail = handle.read()
    except OSError:
        return None
    return tail.decode("utf-8", errors="replace").strip() or None


def _projectionless_translator(plan: DispatchPlan, job: WorkerJob) -> bool:
    if not (
        job.has_execution_identity
        and job.role == "translator"
        and job.worker_id == "translator"
        and job.audience == "translator"
        and job.assignment_ref == "translator"
        and job.duty_id == "translator"
        and job.dispatch_kind == "translator"
    ):
        return False
    execution = _canonical_translator_execution(plan.manifest)
    return bool(
        execution
        and execution.get("roleExecutionRef") == job.role_execution_ref
        and execution.get("participantRef") == job.participant_ref
        and execution.get("executionLabel") == job.execution_label
        and execution.get("provider") == job.provider
    )


def _transition_job_status(
    plan: DispatchPlan,
    job: WorkerJob,
    status: str,
    reason: str,
    *,
    model_execution_value: str = "",
) -> None:
    if not _projectionless_translator(plan, job):
        _transition_worker_status(
            plan.team_state_path,
            job.worker_id,
            status,
            reason,
            model_execution_value=model_execution_value,
        )
        return
    team_state = _load_json_object(plan.team_state_path, "team-state")
    workers = team_state.get("workers")
    if not isinstance(workers, list):
        raise DispatchError("team-state workers must be an array")
    if not any(
        isinstance(row, Mapping) and row.get("workerId") == job.worker_id
        for row in workers
    ):
        return
    _transition_worker_status(
        plan.team_state_path,
        job.worker_id,
        status,
        reason,
        model_execution_value=model_execution_value,
    )


def _job_terminal_status(
    plan: DispatchPlan, job: WorkerJob, attempt: int,
) -> str:
    team_state = _load_json_object(plan.team_state_path, "team-state")
    workers = team_state.get("workers")
    if not isinstance(workers, list):
        raise DispatchError("team-state workers must be an array")
    matches = [
        row for row in workers
        if isinstance(row, Mapping) and row.get("workerId") == job.worker_id
    ]
    if len(matches) == 1:
        return _string_value(matches[0].get("status"))
    if matches or not _projectionless_translator(plan, job):
        raise DispatchError(f"team-state has no unique workerId={job.worker_id}")
    dispatches = team_state.get("workerDispatches")
    records = [
        row for row in dispatches
        if isinstance(row, Mapping)
        and row.get("promptPath") == str(job.prompt_path)
        and row.get("attempt") == attempt
    ] if isinstance(dispatches, list) else []
    if len(records) != 1:
        raise DispatchError("translator dispatch has no unique status record")
    return _string_value(records[0].get("status"))


def _record_dispatch(
    team_state_path: Path, handle: WorkerHandle, attempt: int, status: str, reason: str
) -> None:
    _append_worker_dispatch(
        team_state_path,
        _dispatch_record(
            handle.job,
            attempt,
            status,
            handle.pane_id,
            handle.degraded_from,
            reason,
            mutation_snapshot_path=(
                _mutation_snapshot_path(handle.job)
                if handle.mutation_snapshot is not None
                else None
            ),
            write_policy_digest=handle.write_policy_digest,
            write_enforcement=handle.write_enforcement,
        ),
    )


def _dispatch_record(
    job: WorkerJob,
    attempt: int,
    status: str,
    pane_id: str,
    degraded_from: str,
    reason: str = "",
    *,
    mutation_snapshot_path: Path | None = None,
    write_policy_digest: str = "",
    write_enforcement: WriteEnforcement | None = None,
) -> dict[str, Any]:
    has_execution_identity = job.has_execution_identity
    record = {
        "role": job.role,
        "kind": job.dispatch_kind,
        "attempt": attempt,
        "backendType": BACKEND_CLI_WRAPPER if degraded_from else job.backend,
        "provider": job.provider,
        "status": status,
        "paneId": pane_id,
        # The only handle on a pane worker's transcript: it runs as its own
        # `claude -p` process and leaves no `agentName` in its session jsonl, so
        # token collection resolves its usage by this value or not at all.
        "sessionId": job.session_id,
        "promptPath": str(job.prompt_path),
        "resultPath": str(job.result_path),
        "workerResultPath": str(job.worker_result_path),
        "auditSidecarPath": audit_sidecar_rel(str(job.worker_result_path)),
        "livenessMode": _liveness_mode(
            BACKEND_CLI_WRAPPER if degraded_from else job.backend
        ),
        "statusSidecarPath": str(status_path_for_prompt(job.prompt_path)),
        "degradedFrom": degraded_from,
        "reason": reason,
        "modelExecutionValue": job.model_execution_value,
        "wrapperPath": str(job.wrapper_path),
        "completionPaths": [str(path) for path in job.completion_paths],
        "worktreePath": job.worktree_path,
    }
    if job.invocation_id:
        record.update({
            "dispatchId": _build_dispatch_id(job.invocation_id, attempt),
            "invocationId": job.invocation_id,
            "audience": job.audience,
            "assignmentRef": job.assignment_ref,
            "promptMetadataPath": str(job.prompt_metadata_path),
            **job.digests,
            "hostModelValue": job.host_model_value,
            "enforcementMode": job.enforcement_mode,
        })
    if has_execution_identity:
        record.update({
            "schemaVersion": "2.0",
            "executionIdentityVersion": 2,
            "participantRef": job.participant_ref,
            "roleExecutionRef": job.role_execution_ref,
            "executionLabel": job.execution_label,
            "dutyId": job.duty_id,
            "invocationRef": job.invocation_ref,
            "writePolicyDigest": write_policy_digest,
            "writeEnforcement": (
                write_enforcement.to_payload()
                if write_enforcement is not None
                else None
            ),
            "mutationAuditSnapshotPath": (
                str(mutation_snapshot_path) if mutation_snapshot_path else ""
            ),
        })
    else:
        record["workerId"] = job.worker_id
    return record


def _liveness_mode(backend: str) -> str:
    if backend in (BACKEND_CLI_WRAPPER, BACKEND_CMUX_PANE):
        return LIVENESS_WRAPPER_STATUS
    return LIVENESS_AUDIT_HEARTBEAT


def _update_dispatch_status(
    team_state_path: Path, job: WorkerJob, attempt: int, status: str, reason: str
) -> bool:
    return _update_worker_dispatch_status(
        team_state_path,
        prompt_path=job.prompt_path,
        attempt=attempt,
        status=status,
        reason=reason,
    )


def _teardown_marked_dispatches(team_state_path: Path) -> list[Mapping[str, Any]]:
    """Records `okstra team teardown` wrote off, matched by its own reason string
    so a genuine dispatch error is never mistaken for one."""
    payload = _load_json_object(team_state_path, "team-state")
    dispatches = payload.get("workerDispatches")
    if not isinstance(dispatches, list):
        return []
    return [
        record
        for record in dispatches
        if isinstance(record, dict)
        and record.get("status") == "error"
        and _string_value(record.get("reason")) == TEARDOWN_BEFORE_TERMINAL_REASON
    ]


def _running_dispatches(team_state_path: Path) -> list[Mapping[str, Any]]:
    payload = _load_json_object(team_state_path, "team-state")
    dispatches = payload.get("workerDispatches")
    if not isinstance(dispatches, list):
        return []
    return [record for record in dispatches if isinstance(record, dict) and record.get("status") == "running"]


def _mark_roster_skips(
    project_root: Path,
    manifest: Mapping[str, Any],
    team_state: Mapping[str, Any],
    selected: Sequence[str],
    requested: Sequence[str],
    options: _BuildOptions,
) -> None:
    reasons = _skip_reasons(
        manifest,
        selected,
        requested,
        options,
        team_state=team_state,
    )
    if not reasons:
        return
    team_state_path = _resolve_required_path(project_root, manifest, "teamStatePath")

    def mark_pristine_skips(current_team_state: dict[str, Any]) -> bool:
        changed = False
        for worker_id, reason in reasons.items():
            worker = _worker_state(current_team_state, worker_id)
            if (
                _string_value(worker.get("status")) != "not-run"
                or _string_value(worker.get("reason"))
            ):
                continue
            worker["reason"] = reason
            worker.pop("startedAt", None)
            worker.pop("endedAt", None)
            changed = True
        return changed

    _mutate_team_state(team_state_path, mark_pristine_skips)


def _skip_reasons(
    manifest: Mapping[str, Any],
    selected: Sequence[str],
    requested: Sequence[str],
    options: _BuildOptions,
    *,
    team_state: Mapping[str, Any] | None = None,
) -> dict[str, str]:
    recommended = _string_list(manifest.get("recommendedWorkers"))
    if options.cli_wrapper_assignments_only:
        recommended = [
            worker_id
            for worker_id in recommended
            if _runner_for_worker(team_state or {}, worker_id) != "native-session"
        ]
    if requested:
        return {
            worker_id: (
                "skipped by worker dispatch: worker was not requested in this invocation"
            )
            for worker_id in recommended
            if worker_id not in set(selected)
        }
    supported = set(options.supported_worker_wrappers)
    return {
        worker_id: (
            "skipped by worker dispatch default: worker is not supported by this dispatcher"
        )
        for worker_id in recommended
        if worker_id not in set(selected) and worker_id not in supported
    }


_SIDEBAR_LEVELS = {
    "worker-dispatched": "progress",
    "worker-result-collected": "success",
    "worker-retry-scheduled": "warning",
    "worker-failed": "error",
}


def _relay_to_sidebar(plan: DispatchPlan, event_type: str, details: Mapping[str, Any]) -> None:
    """Mirror a dispatch event onto the lead surface.

    A long run is mostly silence, and the lead surface is the one still
    visible after the user scrolls away or switches workspaces. A failed worker
    additionally raises a notification, because that is the event whose cost
    grows the longer it goes unnoticed. The lead port is `runtime_chain[0]`,
    so a worker that degraded to the wrapper still lands on a cmux run's
    sidebar. A cli-wrapper lead is a no-op.
    """
    plan = _ensure_runtime_chain(plan)
    lead = plan.runtime_chain[0]
    worker_id = str(
        details.get("executionLabel") or details.get("workerId") or "worker"
    )
    notify_title = None
    notify_body = None
    if event_type == "worker-failed":
        notify_title = f"okstra — {_require_string(plan.manifest, 'taskType')}"
        notify_body = f"{worker_id} failed: {details.get('reason', 'no reason recorded')}"
    lead.notify(
        ProgressEvent(
            message=f"{worker_id}: {event_type.removeprefix('worker-')}",
            level=_SIDEBAR_LEVELS.get(event_type, "info"),
            notify_title=notify_title,
            notify_body=notify_body,
        )
    )


def _append_event(plan: DispatchPlan, event_type: str, details: Mapping[str, Any]) -> None:
    _relay_to_sidebar(plan, event_type, details)
    append_lead_event(
        plan.lead_events_path,
        LeadEvent(
            event_type=event_type,
            lead_runtime=_require_string(plan.manifest, "leadRuntime"),
            task_key=_require_string(plan.manifest, "taskKey"),
            task_type=_require_string(plan.manifest, "taskType"),
            run_seq=_run_seq(plan.manifest),
            timestamp=_utc_now(),
            details=details,
        ),
    )


def _attempt_details(job: WorkerJob, attempt: int, handle: WorkerHandle) -> dict[str, Any]:
    return {
        **_event_details(job, handle),
        **_event_execution_identity(job, attempt),
        "maxAttempts": MAX_WORKER_ATTEMPTS,
    }


def _event_details(job: WorkerJob, handle: WorkerHandle) -> dict[str, Any]:
    details = {
        "dispatchMode": BACKEND_CLI_WRAPPER if handle.degraded_from else job.backend,
        "promptPath": str(job.prompt_path),
        "resultPath": str(job.result_path),
        "workerResultPath": str(job.worker_result_path),
        "wrapperPath": str(job.wrapper_path),
        "role": job.role,
        "paneId": handle.pane_id,
        "degradedFrom": handle.degraded_from,
    }
    return details


def _event_execution_identity(job: WorkerJob, attempt: int) -> dict[str, Any]:
    if job.has_execution_identity:
        return {
            "participantRef": job.participant_ref,
            "roleExecutionRef": job.role_execution_ref,
            "executionLabel": job.execution_label,
            "dutyId": job.duty_id,
            "invocationRef": job.invocation_ref,
            "attempt": attempt,
        }
    return {"workerId": job.worker_id, "attempt": attempt}


def _result_details(job: WorkerJob, attempt: int, outcome: WorkerOutcome) -> dict[str, Any]:
    return {
        **_event_execution_identity(job, attempt),
        "dispatchMode": BACKEND_CLI_WRAPPER if outcome.degraded_from else job.backend,
        "missingCompletionPaths": [],
        # A collected result can still come from a wrapper that exited non-zero
        # (`_settle`). The status says the artifacts are good; this says what the
        # process did, so the trace never loses one fact to the other.
        "wrapperExitCode": outcome.returncode,
    }


def _failure_details(
    job: WorkerJob,
    attempt: int,
    outcome: WorkerOutcome,
    reason: str,
    *,
    exit_code: int | None = None,
) -> dict[str, Any]:
    return {
        **_event_execution_identity(job, attempt),
        "maxAttempts": MAX_WORKER_ATTEMPTS,
        "dispatchMode": BACKEND_CLI_WRAPPER if outcome.degraded_from else job.backend,
        "exitCode": outcome.returncode if exit_code is None else exit_code,
        "missingCompletionPaths": [str(path) for path in outcome.missing_completion_paths],
        "reason": reason,
        "promptPath": str(job.prompt_path),
        "resultPath": str(job.result_path),
        "workerResultPath": str(job.worker_result_path),
        "wrapperPath": str(job.wrapper_path),
        "role": job.role,
    }


def _retry_details(job: WorkerJob, attempt: int, outcome: WorkerOutcome) -> dict[str, Any]:
    return {
        **_event_execution_identity(job, attempt),
        "missingCompletionPaths": [str(path) for path in outcome.missing_completion_paths],
        "reason": "required worker artifact was not produced",
    }


def _outcome_from_status(record: Mapping[str, Any], status) -> WorkerOutcome:
    return WorkerOutcome(
        returncode=status.exit_code if status.exit_code is not None else 1,
        missing_completion_paths=tuple(path for path in _record_completion_paths(record) if not path.is_file()),
        pane_id=_string_value(record.get("paneId")),
        status_sidecar_path=status.path,
        timeout=status.timeout,
        terminal_stage=status.stage,
        degraded_from=_string_value(record.get("degradedFrom")),
    )


def _job_from_record(project_root: Path, record: Mapping[str, Any]) -> WorkerJob:
    identity = _worker_execution_identity(record)
    return WorkerJob(
        worker_id=(
            _require_string(record, "workerId")
            if identity is None
            else _v2_worker_state_key(record)
        ),
        provider=_require_string(record, "provider"),
        backend=BACKEND_CLI_WRAPPER,
        project_root=project_root,
        model_execution_value=_require_string(record, "modelExecutionValue"),
        wrapper_path=Path(_require_string(record, "wrapperPath")),
        prompt_path=Path(_require_string(record, "promptPath")),
        result_path=Path(_require_string(record, "resultPath")),
        worker_result_path=Path(_require_string(record, "workerResultPath")),
        completion_paths=tuple(_record_completion_paths(record)),
        worktree_path=_string_value(record.get("worktreePath")),
        role=_require_string(record, "role"),
        # 복원된 job 은 다시 실행되지 않으므로 예산을 정하지 않는다.
        idle_timeout_seconds=None,
        dispatch_kind=_require_string(record, "kind"),
        # Restored, never re-issued: both callers rebuild a job from its record
        # to settle a dispatch that already ran, and a fresh id would settle it
        # against a session that never existed.
        session_id=_string_value(record.get("sessionId")),
        invocation_id=_string_value(record.get("invocationId")),
        audience=_string_value(record.get("audience")),
        assignment_ref=_string_value(record.get("assignmentRef")),
        prompt_metadata_path=Path(
            _string_value(record.get("promptMetadataPath"))
        ),
        catalog_digest=_string_value(record.get("catalogDigest")),
        assignment_digest=_string_value(record.get("assignmentDigest")),
        duty_digest=_string_value(record.get("dutyDigest")),
        instruction_digest=_string_value(record.get("instructionDigest")),
        prompt_digest=_string_value(record.get("promptDigest")),
        host_model_value=(
            record.get("hostModelValue")
            if isinstance(record.get("hostModelValue"), str)
            else None
        ),
        # jobs 파일은 `agent-prompt materialize` 가 쓴 `.meta.json` 을 옮겨 적는
        # 것이 정상 경로인데, materialize 는 이 값을 기록하지 않는다 — 기록할
        # 것이 없기 때문이다. worker-dispatch 를 타는 디스패치의 집행 방식은
        # 하나뿐이고, 다른 하나(`host-native-spec-link-gate`)는 리드가 호스트
        # 원시 호출을 직접 기록할 때만 쓴다. 그래서 손으로 채우게 하지 않는다.
        enforcement_mode=(
            _string_value(record.get("enforcementMode"))
            or CLI_DISPATCH_ENFORCEMENT_MODE
        ),
        **(identity or {}),
    )


def _record_completion_paths(record: Mapping[str, Any]) -> list[Path]:
    return [Path(path) for path in _string_list(record.get("completionPaths"))]


def _should_retry(outcome: WorkerOutcome, attempt: int) -> bool:
    return outcome.returncode == 0 and bool(outcome.missing_completion_paths) and attempt < MAX_WORKER_ATTEMPTS


def _job_for_next_attempt(job: WorkerJob) -> WorkerJob:
    """The same work as a new attempt, which means a new session.

    The first attempt's process is dead; a retry starts another one. Carrying
    the id over would merge two processes into one attribution unit, and it
    would hand `claude --session-id` a value that is already taken — behaviour
    the CLI does not document. Every re-dispatch goes through here so no path
    can quietly keep the old session.

    A provider with no session flag gets '' back, so this is a no-op for it.
    """
    return replace(job, session_id=_dispatch_session_id(job.provider))


def _failure_reason(outcome: WorkerOutcome) -> str:
    if outcome.timeout:
        return "worker wrapper timed out"
    if outcome.returncode != 0:
        return f"wrapper exited with code {outcome.returncode}"
    if outcome.missing_completion_paths:
        missing = ", ".join(str(path) for path in outcome.missing_completion_paths)
        return f"required worker artifact was not produced: {missing}"
    return "worker dispatch failed"


def _mode_from_handles(handles: Sequence[WorkerHandle]) -> str:
    modes = {BACKEND_CLI_WRAPPER if h.degraded_from else h.job.backend for h in handles}
    if len(modes) == 1:
        return next(iter(modes))
    return BACKEND_MIXED


def _validate_manifest(manifest: Mapping[str, Any], path: Path, required: str | None) -> None:
    if required is not None and manifest.get("leadRuntime") != required:
        raise DispatchError(f"run manifest is not a {required} lead run: {path}")
    if not isinstance(manifest.get("leadEventsPath"), str):
        raise DispatchError(f"run manifest has no leadEventsPath: {path}")


def _resolve_wrapper(provider: str, workspace_root: Path, options: _BuildOptions) -> Path:
    script = options.supported_worker_wrappers.get(provider)
    if not script:
        raise DispatchError(f"unsupported worker provider: {provider}")
    candidates = []
    if options.okstra_bin is not None:
        candidates.append(options.okstra_bin / script)
    candidates.extend([workspace_root / "bin" / script, workspace_root / "scripts" / script])
    for candidate in candidates:
        if candidate.is_file():
            return candidate.resolve()
    raise DispatchError(f"{script} not found (searched: {', '.join(str(c) for c in candidates)})")


def _provider_for_worker(
    worker_id: str,
    options: _BuildOptions | None = None,
) -> str:
    if options is not None:
        configured = options.default_provider_by_worker_id.get(worker_id)
        if configured:
            return configured
    if worker_id == REPORT_WRITER_WORKER_ID:
        return "claude"
    return worker_id




def _reject_stale_schema_excerpt(
    project_root: Path,
    manifest: Mapping[str, Any],
    jobs: Sequence[WorkerJob],
) -> None:
    """Refuse to send the report writer at a schema excerpt from another runtime.

    The bundle's `instruction-set/final-report-schema.json` is cut at prep time
    and never moves again, while validation always runs against the installed
    schema. A run long enough to straddle a runtime upgrade therefore has the
    author writing to one contract and the validator reading another — and the
    only thing that noticed was the renderer, in Phase 6, after the worker had
    authored the whole report. The same inputs are already on hand the moment the
    dispatch is built, and the remedy is the same either way, so it belongs here.

    Only the report writer is stopped: it is the only worker that authors against
    the excerpt. Re-running bundle prep re-cuts it from the installed schema.

    The stamp alone is not the test. Most releases change nothing this task-type
    authors against, and re-prep is not a cheap way to rewrite one stamp line: it
    opens a new run, so every artifact bound to the current one — convergence
    state, critic merges, the plan-item queue and its verdicts — is left behind.
    The guard therefore fires on the contract text and names what moved.
    """
    writer = next(
        (job for job in jobs if job.worker_id == REPORT_WRITER_WORKER_ID), None
    )
    if writer is None:
        return
    expected = _string_value(manifest.get("expectedReportRecordPath"))
    task_type = _string_value(manifest.get("taskType"))
    if not expected or not task_type:
        return
    excerpt_path = bundle_excerpt_path(_resolve_project_path(project_root, expected))
    if excerpt_path is None:
        return
    installed = installed_version()
    skew = excerpt_contract_skew(excerpt_path, task_type, installed)
    if skew is None:
        return
    raise DispatchError(
        f"the bundle's schema excerpt ({excerpt_path}) was cut from okstra "
        f"{skew.cut_from} and this runtime is {installed}, which states "
        f"{task_type}'s contract differently: {describe_changed(skew.changed)}. "
        f"The report writer authors against that excerpt and validation runs "
        f"against the installed schema, so dispatching now spends a full authoring "
        f"pass on the wrong contract. Re-prepare the task bundle to re-cut the "
        f"excerpt, then dispatch again."
    )


def _load_optional_json(project_root: Path, value: Any) -> dict[str, Any]:
    if not isinstance(value, str) or not value.strip():
        return {}
    path = _resolve_project_path(project_root, value)
    if not path.is_file():
        return {}
    return hydrate_active_run_context(_load_json_object(path, "active-run-context"))


def _optional_path(value: object) -> Path | None:
    if not isinstance(value, str) or not value.strip():
        return None
    return Path(value)


def _run_seq(manifest: Mapping[str, Any]) -> str:
    seqs = manifest.get("runSequencesByCategory")
    if isinstance(seqs, Mapping):
        return _string_value(seqs.get("manifests")) or "001"
    return "001"