"""Worker dispatch plumbing shared by both lead dispatchers.

`dispatch_core` (claude / external lead) and `codex_dispatch` (codex lead) differ
in capability — only the former has cmux surfaces, blocking waits, and retry from
a persisted record. What they do *not* differ in is the job value object and the
run's state files: the same run-manifest keys, the same team-state document, the
same wrapper argv.

Those used to be two copies that had already drifted apart: one dispatcher
wrapped a truncated team-state in `DispatchError` while the other let
`json.JSONDecodeError` escape, one raised on an unknown `workerId` while the
other silently did nothing, and one gated the worktree argument on task type
while the other handed it to every phase. This module is the single reference
point so a fix cannot land on one lead runtime only.

Enforced by `tests/contract/test_dispatcher_shared_helpers.py`.
"""
from __future__ import annotations

import contextlib
import fcntl
import json
import os
import tempfile
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Iterable, Mapping, Sequence

from . import cmux
from .agent.invocation import (
    AgentInvocationError,
    AgentModelAssignment,
    InvocationMetadataIdentity,
    agent_model_assignment_from_payload,
    invocation_metadata_identity,
    invocation_input_digest,
    v2_role_assignment_authority_errors,
    verify_agent_invocation,
)
from .domain.role import RoleCatalogError, role_for_duty
from .execution_identity import (  # noqa: F401 — build_dispatch_id 재노출
    Attempt,
    Invocation,
    build_dispatch_id,
)
from .execution_manifest import (
    finish_attempt_mutation,
    read_execution_manifest,
    record_invocation_attempt,
)
from .execution_mutation_audit import ExecutionMutationAudit, MutationSnapshot
from .final_report_paths import final_report_data_path
from .report_inputs import report_narrative_path, uses_report_contract_v3
from .report_narrative import narrative_structure_defect
from .verdict_blocks import finding_vote_defect
from .worker_prompt_body import REPORT_WRITER_WORKER_ID
from .worker_prompt_contract import (
    PromptRecord,
    validate_initial_prompt_records,
    validate_prompt_model_header,
    validate_reverify_prompt,
)
from .worker_prompt_policy import (
    is_verification_dispatch_kind,
    verification_dispatch_round,
)
from .worker_runner import LIVE, QUIET
from .worker_request import verifier_extra_dirs
from .worker_artifact_paths import audit_sidecar_rel
from .wrapper_status import (
    log_path_for_prompt,
    prompt_derived_paths,
    status_path_for_prompt,
)
from .write_policy import (
    build_invocation_write_contract,
    planned_paths_from_run_manifest,
    task_root_from_run_manifest,
    write_policy_from_payload,
)
from .convergence_store import write_json_atomic
from .json_boundary import (
    JsonBoundaryError,
    external_worker_json_source,
    load_external_json,
    load_owned_object,
    write_owned_object_atomic,
)

BACKEND_CLI_WRAPPER = "cli-wrapper"
BACKEND_CMUX_PANE = "cmux-pane"
BACKEND_MIXED = "mixed"


def detect_terminal_backend() -> str:
    """Which backend this run's workers get. Called once, by prepare.

    cmux is the only backend that gives a worker a surface of its own. Without
    it a worker runs as a cli-wrapper subprocess, which owns no pane and is read
    through its status sidecar instead. The answer is written to the run manifest
    and read back from there — consumers must not re-detect, or two phases of one
    run can disagree.
    """
    if cmux.cmux_available():
        return BACKEND_CMUX_PANE
    return BACKEND_CLI_WRAPPER

# `livenessMode` picks which artifact answers "is this worker still alive": the
# in-process worker's audit sidecar heartbeat, or the CLI wrapper's status
# sidecar. Both dispatchers write it and `worker_liveness` reads it, so the
# vocabulary lives here rather than in whichever module happened to need it.
LIVENESS_AUDIT_HEARTBEAT = "audit-heartbeat"
LIVENESS_WRAPPER_STATUS = "wrapper-status"

# `okstra team teardown` stamps this on a dispatch it wrote off, and a later
# `okstra team await` matches it to tell those apart from a genuine dispatch
# error before letting the wrapper's own exit settle them. Two sides of one fact,
# so the string lives here rather than in whichever module wrote it first.
TEARDOWN_BEFORE_TERMINAL_REASON = "teardown before terminal status"

WORKER_STATUSES = frozenset(
    {"in-progress", "completed", "timeout", "error", "not-run"}
)
# Which of those mean the dispatch is still expected to produce something. Two
# readers key off this split — `team reclaim` closes a finished dispatch's pane
# and must never touch a live one, and the compact-reminder hook calls a run
# in-flight when any dispatch is still here. Both restated the terminal four
# locally before, so a sixth status would have read as finished in one place and
# as live in the other.
NON_TERMINAL_WORKER_STATUSES = frozenset({"in-progress"})
TERMINAL_WORKER_STATUSES = WORKER_STATUSES - NON_TERMINAL_WORKER_STATUSES
REASON_REQUIRED_STATUSES = frozenset({"timeout", "error", "not-run"})


def generate_claude_session_id() -> str:
    """A UUIDv4 for `claude --session-id`.

    The lead mints one for its own session (`run`) and dispatch mints one per
    worker (`dispatch_session_id`); they are the same kind of value, so one
    generator serves both. It sits here rather than in `session` because
    `session` imports this module, so a worker-side call the other way would
    close an import cycle.
    """
    return str(uuid.uuid4())


class DispatchError(Exception):
    """Raised when a worker dispatch request cannot be executed."""


class IncompletePromptWriteArtifacts(DispatchError):
    """Raised when a worker prompt omits the standard artifact headers."""


@dataclass(frozen=True)
class WorkerJob:
    worker_id: str
    provider: str
    backend: str
    project_root: Path
    model_execution_value: str
    wrapper_path: Path
    prompt_path: Path
    result_path: Path
    worker_result_path: Path
    completion_paths: tuple[Path, ...]
    worktree_path: str
    role: str
    # None 은 "이 디스패치가 예산을 정하지 않는다" 는 뜻이고, 그때 예산은
    # 역할이 정한다(`domain/worker_role.role_spec`). 여기에 숫자 기본값을 두면
    # 역할별 예산이 영영 도달 불가능해진다 — 엔트리포인트는 이 자리가 비었을
    # 때만 역할을 보기 때문이다.
    idle_timeout_seconds: int | None
    dispatch_kind: str
    invocation_id: str = ""
    audience: str = ""
    assignment_ref: str = ""
    prompt_metadata_path: Path = Path()
    catalog_digest: str = ""
    assignment_digest: str = ""
    duty_digest: str = ""
    instruction_digest: str = ""
    prompt_digest: str = ""
    host_model_value: str | None = None
    enforcement_mode: str = ""
    # The session id dispatch mints and carries through to the provider CLI, so
    # token collection can find the worker's jsonl by this value rather than by
    # guessing at `agentName`. A worker started in a cmux pane is a separate CLI
    # process and leaves no `agentName` behind, so without this its usage cannot
    # be attributed at all.
    session_id: str = ""
    participant_ref: str = ""
    role_execution_ref: str = ""
    execution_label: str = ""
    duty_id: str = ""
    invocation_ref: str = ""
    result_aliases: tuple[Path, ...] = ()
    attempt: int = 0

    @property
    def command(self) -> list[str]:
        # The flag trails the positional contract the entrypoint reads by index.
        argv = [
            str(self.wrapper_path),
            str(self.project_root),
            self.model_execution_value,
            str(self.prompt_path),
            self.worktree_path,
            self.wrapper_role,
            self._idle_timeout_argument,
            "--presentation",
            self._presentation(),
        ]
        if self.has_execution_identity:
            argv.extend([
                "--invocation-metadata",
                str(self.prompt_metadata_path),
            ])
        if self.session_id:
            argv += ["--session-id", self.session_id]
        return argv

    @property
    def _idle_timeout_argument(self) -> str:
        """엔트리포인트가 읽는 유휴 예산 자리.

        비워 두면 엔트리포인트가 역할의 예산을 쓴다. 이 자리를 항상 채우던
        동안에는 `worker_request.idle_timeout` 의 역할 분기가 한 번도 실행되지
        않았고, executor/verifier 의 1500s 도 함께 죽어 있었다.
        """
        if self.idle_timeout_seconds is None:
            return ""
        return str(self.idle_timeout_seconds)

    @property
    def wrapper_role(self) -> str:
        """The canonical role the entrypoint's role positional is read as.

        `self.role` is the roster label team-state carries (`Antigravity
        worker`) so the report's execution-status row can quote it. The
        entrypoint reads that same position through `normalize_role`, which
        knows only the eleven canonical ids, and compares it against
        `role_for_duty(dutyId)` from the invocation metadata. Handing it the
        label failed that check before the status sidecar was written, so a
        pane worker exited 64 leaving no `.log` and no `.status.json` while
        team-state still read `in-progress`. It also silently denied the
        verifier the toolchain dirs `verifier_extra_dirs` grants by role.

        A legacy v1 job carries no duty id; the entrypoint skips the metadata
        role check for it, so its existing value is preserved.
        """
        if not self.duty_id:
            return self.role
        try:
            return role_for_duty(self.duty_id)
        except RoleCatalogError:
            return self.role

    def _presentation(self) -> str:
        # A pane is a screen a person watches; a cli-wrapper dispatch's stdout is
        # a subagent's context window. Progress belongs in the first and not the
        # second.
        return QUIET if self.backend == BACKEND_CLI_WRAPPER else LIVE

    def to_payload(self) -> dict[str, Any]:
        payload = {
            "provider": self.provider,
            "backend": self.backend,
            "modelExecutionValue": self.model_execution_value,
            "wrapperPath": str(self.wrapper_path),
            "promptPath": str(self.prompt_path),
            "resultPath": str(self.result_path),
            "workerResultPath": str(self.worker_result_path),
            "completionPaths": [str(path) for path in self.completion_paths],
            "worktreePath": self.worktree_path,
            "role": self.role,
            "dispatchKind": self.dispatch_kind,
            "command": self.command,
        }
        if self.has_execution_identity:
            payload.update({
                "schemaVersion": "2.0",
                "executionIdentityVersion": 2,
                "participantRef": self.participant_ref,
                "roleExecutionRef": self.role_execution_ref,
                "executionLabel": self.execution_label,
                "dutyId": self.duty_id,
                "invocationRef": self.invocation_ref,
                "attempt": self.attempt,
            })
        else:
            payload.update({"schemaVersion": 1, "workerId": self.worker_id})
        if self.invocation_id:
            payload.update({
                "invocationId": self.invocation_id,
                "audience": self.audience,
                "assignmentRef": self.assignment_ref,
                "promptMetadataPath": str(self.prompt_metadata_path),
                "digests": self.digests,
                "hostModelValue": self.host_model_value,
                "enforcementMode": self.enforcement_mode,
            })
        return payload

    @property
    def has_execution_identity(self) -> bool:
        values = (
            self.participant_ref,
            self.role_execution_ref,
            self.execution_label,
            self.duty_id,
            self.invocation_ref,
        )
        present = tuple(bool(value) for value in values)
        if any(present) and not all(present):
            raise DispatchError("worker job has partial v2 execution identity")
        if all(present) and self.attempt < 1:
            raise DispatchError("worker job v2 attempt must be positive")
        return all(present)

    @property
    def digests(self) -> dict[str, str]:
        return {
            "catalogDigest": self.catalog_digest,
            "assignmentDigest": self.assignment_digest,
            "dutyDigest": self.duty_digest,
            "instructionDigest": self.instruction_digest,
            "promptDigest": self.prompt_digest,
        }


# --- run-manifest reads -------------------------------------------------------

def string_value(value: Any) -> str:
    return value.strip() if isinstance(value, str) else ""


def require_string(payload: Mapping[str, Any], key: str) -> str:
    value = payload.get(key)
    if not isinstance(value, str) or not value.strip():
        raise DispatchError(f"missing required string field: {key}")
    return value.strip()


def string_list(value: Any) -> list[str]:
    if not isinstance(value, list):
        return []
    return [str(item).strip() for item in value if str(item).strip()]


def resolve_project_path(project_root: Path, value: str) -> Path:
    path = Path(value)
    return path if path.is_absolute() else project_root / path


def resolve_required_path(
    project_root: Path, manifest: Mapping[str, Any], key: str
) -> Path:
    return resolve_project_path(project_root, require_string(manifest, key))


def worktree_path(
    manifest: Mapping[str, Any], active_context: Mapping[str, Any]
) -> str:
    worktree = active_context.get("executorWorktree")
    if not isinstance(worktree, Mapping):
        return ""
    return str(worktree.get("path") or "")


# --- team-state document ------------------------------------------------------

def load_json_object(path: Path, label: str) -> dict[str, Any]:
    if not path.is_file():
        raise DispatchError(f"{label} not found: {path}")
    try:
        payload = load_owned_object(path, artifact=label)
    except JsonBoundaryError as exc:
        raise DispatchError(f"{label} is invalid JSON: {path}: {exc}") from exc
    if not isinstance(payload, dict):
        raise DispatchError(f"{label} must be a JSON object: {path}")
    return payload


def write_json(path: Path, payload: Mapping[str, Any]) -> None:
    # Write via temp file + os.replace so a crash mid-write cannot leave a
    # truncated team-state.json that every later load_json_object rejects;
    # os.replace is atomic on POSIX, matching run_context._atomic_write_json.
    write_owned_object_atomic(path, payload, artifact="team state")


@contextlib.contextmanager
def _team_state_lock(path: Path):
    lock_path = path.with_name(path.name + ".lock")
    lock_path.parent.mkdir(parents=True, exist_ok=True)
    handle = lock_path.open("a+")
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
        yield
    finally:
        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        handle.close()


def mutate_team_state(
    team_state_path: Path,
    mutation: Callable[[dict[str, Any]], bool],
) -> bool:
    """Apply one arbitrary team-state mutation under the shared file lock."""
    with _team_state_lock(team_state_path):
        payload = load_json_object(team_state_path, "team-state")
        changed = mutation(payload)
        if changed:
            write_json(team_state_path, payload)
        return changed


def append_worker_dispatch(
    team_state_path: Path,
    record: Mapping[str, Any],
) -> None:
    """Record one worker dispatch without losing another state writer's update.

    A row carrying a dispatch id replaces the row that already holds it rather
    than stacking beside it: an observed lead re-sent `okstra team dispatch` for
    a prompt and attempt it had already sent, team-state kept both copies, and
    `validate-run` failed that run with `agent dispatch ID is duplicated`.

    The replaced row takes its `paneId` with it, and recorded ids are the only
    reclaim candidates `team._reclaimable_panes` has, so that pane is left on
    screen for the session. That is the price of the trade: before this, the
    same re-send cost the whole run its report.

    A job with no invocation carries no dispatch id, so it has no key to
    collapse on and folding those rows would merge two unrelated dispatches into
    one. They still stack, and `update_worker_dispatch_status` settles the
    newest of them.
    """
    with _team_state_lock(team_state_path):
        payload = load_json_object(team_state_path, "team-state")
        dispatches = payload.setdefault("workerDispatches", [])
        if not isinstance(dispatches, list):
            raise DispatchError(
                f"team-state workerDispatches must be an array: {team_state_path}"
            )
        new_record = dict(record)
        new_id = str(new_record.get("dispatchId") or "")
        replaced = False
        if new_id:
            for index, existing in enumerate(dispatches):
                if (
                    isinstance(existing, Mapping)
                    and existing.get("dispatchId") == new_id
                ):
                    new_record["startedAt"] = existing.get("startedAt") or _utc_timestamp(None)
                    dispatches[index] = new_record
                    replaced = True
                    break
        if not replaced:
            new_record.setdefault("startedAt", _utc_timestamp(None))
            dispatches.append(new_record)
        write_json(team_state_path, payload)


def update_worker_dispatch_status(
    team_state_path: Path,
    *,
    prompt_path: Path,
    attempt: int,
    status: str,
    reason: str,
) -> bool:
    """Settle the newest matching dispatch row under the shared state lock."""
    with _team_state_lock(team_state_path):
        payload = load_json_object(team_state_path, "team-state")
        dispatches = payload.get("workerDispatches", [])
        if not isinstance(dispatches, list):
            raise DispatchError(
                f"team-state workerDispatches must be an array: {team_state_path}"
            )
        matches = [
            record
            for record in dispatches
            if dispatch_record_matches(record, prompt_path, attempt)
        ]
        if not matches:
            return False
        # A re-send can reuse prompt + attempt. The newest row is the live one.
        matches[-1]["status"] = status
        matches[-1]["reason"] = reason
        write_json(team_state_path, payload)
        return True


_AGENT_DIGEST_KEYS = (
    "catalogDigest",
    "assignmentDigest",
    "dutyDigest",
    "instructionDigest",
    "promptDigest",
)
_AGENT_ENFORCEMENT_MODES = frozenset({
    "core-pre-dispatch",
    "host-native-spec-link-gate",
})


def _require_prompt_model_matches_assignment(
    metadata_path: Path, assignment: Any
) -> None:
    """디스패치 직전, 프롬프트가 이 배정의 모델을 가리키는지 확인한다.

    메타데이터는 프롬프트 옆에 `<prompt>.meta.json` 으로 놓인다. 읽을 수 없으면
    조용히 통과한다 — 이 검사는 어긋남을 앞당겨 잡는 것이지, 프롬프트를 못 읽는
    사정을 새 차단 사유로 만드는 것이 아니다. 그 사정은 이미 다른 검사가 본다.
    """
    prompt_path = metadata_path.with_name(
        metadata_path.name.removesuffix(".meta.json")
    )
    if prompt_path == metadata_path or not prompt_path.is_file():
        return
    expected = str(getattr(assignment, "model_execution_value", "") or "").strip()
    if not expected:
        return
    try:
        text = prompt_path.read_text(encoding="utf-8")
    except OSError:
        return
    errors = validate_prompt_model_header(text, expected)
    if errors:
        raise DispatchError(
            f"agent dispatch prompt does not name this assignment's model: "
            + "; ".join(errors)
        )


def record_verified_agent_dispatch(
    *,
    project_root: Path,
    run_manifest_path: Path,
    metadata_path: Path,
    enforcement_mode: str,
    allow_native_core: bool = False,
) -> dict[str, Any]:
    """Append one immutable dispatch-to-invocation association.

    Host-native calls use this record as a specification link; it deliberately
    does not claim that Okstra observed the bytes delivered by the host. A
    code-owned lead process may set ``allow_native_core`` because lead model
    assignments use the host's native model vocabulary even when Okstra owns
    the outer provider CLI process.
    """
    project_root = project_root.resolve()
    run_manifest_path = resolve_project_path(
        project_root, str(run_manifest_path)
    ).resolve(strict=True)
    metadata_path = resolve_project_path(
        project_root, str(metadata_path)
    ).resolve(strict=True)
    if enforcement_mode not in _AGENT_ENFORCEMENT_MODES:
        raise DispatchError(f"invalid agent enforcement mode: {enforcement_mode}")
    manifest = load_json_object(run_manifest_path, "run manifest")
    metadata = load_json_object(metadata_path, "agent invocation metadata")
    assignment_ref = require_string(metadata, "assignmentRef")
    assignments = manifest.get("invocationAssignments")
    if not isinstance(assignments, Mapping):
        raise DispatchError("run manifest has no invocationAssignments object")
    try:
        assignment = agent_model_assignment_from_payload(assignments.get(assignment_ref))
    except AgentInvocationError as exc:
        raise DispatchError(str(exc)) from exc
    try:
        execution_identity = invocation_metadata_identity(metadata)
    except AgentInvocationError as exc:
        raise DispatchError(str(exc)) from exc
    worker_id = (
        require_string(metadata, "workerId")
        if execution_identity is None
        else None
    )
    audience = require_string(metadata, "audience")
    invocation_id = require_string(metadata, "invocationId")
    if (
        execution_identity is not None
        and execution_identity.invocation_ref != invocation_id
    ):
        raise DispatchError("v2 invocation reference does not match invocation ID")
    errors = verify_agent_invocation(
        metadata_path,
        project_root=project_root,
        expected_run_manifest_path=run_manifest_path,
        expected_assignment=assignment,
        expected_invocation_id=invocation_id,
        expected_worker_id=worker_id,
        expected_assignment_ref=assignment_ref,
        expected_audience=audience,
        expected_participant_ref=(
            execution_identity.participant_ref if execution_identity else None
        ),
        expected_role_execution_ref=(
            execution_identity.role_execution_ref if execution_identity else None
        ),
        expected_invocation_ref=(
            execution_identity.invocation_ref if execution_identity else None
        ),
    )
    if errors:
        raise DispatchError("agent invocation verification failed: " + "; ".join(errors))
    _require_prompt_model_matches_assignment(metadata_path, assignment)
    if (
        enforcement_mode == "host-native-spec-link-gate"
        and assignment.runner != "native-session"
        and not _is_current_session_lead(
            run_manifest_path, execution_identity, assignment
        )
    ):
        raise DispatchError(
            "host-native enforcement requires a native-session assignment"
        )
    if (
        enforcement_mode == "core-pre-dispatch"
        and assignment.runner == "native-session"
        and not allow_native_core
    ):
        raise DispatchError(
            "native-session assignment must use host-native-spec-link-gate"
        )
    digests = metadata.get("digests")
    if not isinstance(digests, Mapping):
        raise DispatchError("agent invocation digests are missing")
    digest_values = {
        key: require_string(digests, key) for key in _AGENT_DIGEST_KEYS
    }
    prompt = metadata.get("prompt")
    if not isinstance(prompt, Mapping):
        raise DispatchError("agent invocation prompt is missing")
    prompt_path = require_string(prompt, "path")
    # Identity-bearing records carry the stored attempt. A path without
    # identity still records the first attempt; the conflict guard below
    # refuses a second write under that id.
    dispatch_id = build_dispatch_id(
        invocation_id,
        execution_identity.attempt if execution_identity else 1,
    )
    record = {
        "dispatchId": dispatch_id,
        "audience": audience,
        "invocationId": invocation_id,
        "assignmentRef": assignment_ref,
        "promptPath": prompt_path,
        "promptMetadataPath": _relative_project_path(project_root, metadata_path),
        **digest_values,
        "modelExecutionValue": assignment.model_execution_value,
        "hostModelValue": assignment.host_model_value,
        "enforcementMode": enforcement_mode,
        "promptDeliveryVerified": False,
        "status": "dispatched",
    }
    if execution_identity is None:
        record["workerId"] = worker_id
    else:
        record.update({
            "schemaVersion": "2.0",
            "executionIdentityVersion": 2,
            "invocationRef": execution_identity.invocation_ref,
            "participantRef": execution_identity.participant_ref,
            "roleExecutionRef": execution_identity.role_execution_ref,
            "executionLabel": execution_identity.execution_label,
            "dutyId": execution_identity.duty_id,
            "attempt": execution_identity.attempt,
            "dispatchKind": require_string(metadata, "dispatchKind"),
        })
    team_state_path = resolve_required_path(project_root, manifest, "teamStatePath")
    write_policy = None
    write_enforcement = None
    mutation_snapshot_path = metadata_path.with_suffix(
        metadata_path.suffix + ".mutation-audit.json"
    )
    if execution_identity is not None:
        write_policy, write_enforcement = _agent_write_contract(
            project_root,
            run_manifest_path,
            manifest,
            execution_identity.role_execution_ref,
            resolve_project_path(project_root, prompt_path),
            assignment_ref,
        )
        record.update({
            "writePolicyDigest": write_policy.digest,
            "writeEnforcement": write_enforcement.to_payload(),
            "mutationAuditSnapshotPath": (
                _relative_project_path(project_root, mutation_snapshot_path)
                if write_enforcement.mutation_audit == "batch"
                else ""
            ),
        })
    existing_state = load_json_object(team_state_path, "team-state")
    existing = [
        item
        for item in existing_state.get("agentDispatches") or []
        if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id
    ]
    if existing:
        if len(existing) != 1 or dict(existing[0]) != record:
            raise DispatchError(f"agent dispatch ID conflicts: {dispatch_id}")
        return record
    if execution_identity is not None:
        dispatch_kind = require_string(metadata, "dispatchKind")
        if write_policy is None or write_enforcement is None:
            raise DispatchError("agent write contract was not prepared")
        record_invocation_attempt(
            run_manifest_path,
            Invocation(
                invocation_ref=execution_identity.invocation_ref,
                participant_ref=execution_identity.participant_ref,
                role_execution_ref=execution_identity.role_execution_ref,
                source_invocation_ref=None,
                recovery_ref=None,
                duty_id=execution_identity.duty_id,
                dispatch_kind=dispatch_kind,
                round=_execution_dispatch_round(dispatch_kind),
                input_digest=invocation_input_digest(metadata, project_root),
            ),
            Attempt(
                invocation_ref=execution_identity.invocation_ref,
                attempt=execution_identity.attempt,
                started_at=utc_now(),
                finished_at=None,
                status="started",
                result_path=None,
                error_path=None,
                change_summary={},
                evidence_seal_ref=None,
                write_policy=write_policy.to_payload(),
                write_policy_digest=write_policy.digest,
                write_enforcement=write_enforcement.to_payload(),
            ),
            task_key=require_string(manifest, "taskKey"),
        )
        if write_enforcement.mutation_audit == "batch":
            snapshot = ExecutionMutationAudit().snapshot(
                (write_policy,),
                orchestrator_paths=(
                    # run 산출물 트리 전체. 같은 run 의 다른 라운드가 감사 창
                    # 안에 남기는 프롬프트·로그·사이드카를 파일 단위로 미리 셀
                    # 수 없고, 그 아래는 정의상 소스가 아니다.
                    Path(run_manifest_path).resolve().parents[1],
                    run_manifest_path,
                    team_state_path,
                    Path(f"{team_state_path}.lock"),
                    mutation_snapshot_path,
                ),
            )
            write_json_atomic(mutation_snapshot_path, snapshot.to_payload())
    with _team_state_lock(team_state_path):
        team_state = load_json_object(team_state_path, "team-state")
        dispatches = team_state.setdefault("agentDispatches", [])
        if not isinstance(dispatches, list):
            raise DispatchError("team-state agentDispatches must be an array")
        existing = [
            item for item in dispatches
            if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id
        ]
        if existing:
            if len(existing) != 1 or dict(existing[0]) != record:
                raise DispatchError(f"agent dispatch ID conflicts: {dispatch_id}")
            return record
        dispatches.append(record)
        write_json(team_state_path, team_state)
    return record


def _is_current_session_lead(
    manifest_path: Path,
    execution_identity: InvocationMetadataIdentity | None,
    assignment: AgentModelAssignment,
) -> bool:
    """Report whether this dispatch is the lead attesting its own session.

    A current-session lead has no model binding, so the run manifest projects
    its assignment as ``cli-wrapper`` while the execution manifest records the
    participant as ``current-session``. It runs through no dispatch boundary at
    all: the spec link is the lead associating the already-running session with
    its verified invocation specification, which is what the host-native gate
    records. Keyed on the execution manifest's own participant row rather than
    on the projected runner string, so a genuine cli-wrapper worker never
    reaches the native gate.
    """
    if execution_identity is None or assignment.runner == "native-session":
        return False
    manifest = read_execution_manifest(manifest_path)
    if manifest.legacy:
        return False
    participant = next(
        (
            row for row in manifest.participant_assignments
            if row.participant_ref == execution_identity.participant_ref
        ),
        None,
    )
    execution = next(
        (
            row for row in manifest.role_executions
            if row.role_execution_ref == execution_identity.role_execution_ref
        ),
        None,
    )
    return (
        participant is not None
        and execution is not None
        and execution.role == "leader"
        and participant.runner == "current-session"
        and participant.entry_mode == "current-session"
    )


def _agent_write_contract(
    project_root: Path,
    manifest_path: Path,
    authority: Mapping[str, Any],
    role_execution_ref: str,
    prompt_path: Path,
    assignment_ref: str,
):
    execution_manifest = read_execution_manifest(manifest_path)
    execution = next(
        (
            row for row in execution_manifest.role_executions
            if row.role_execution_ref == role_execution_ref
        ),
        None,
    )
    capability = (
        execution.binding.worker_write_capability
        if execution is not None and execution.binding is not None
        else None
    )
    if execution is None:
        raise DispatchError("agent write policy has no role execution")
    if execution.role == "leader":
        artifacts = _leader_artifact_paths(project_root, authority)
        worktree = None
        maximum_precision = "none"
    elif capability is not None:
        try:
            artifacts, worktree = _prompt_write_paths(project_root, prompt_path)
        except IncompletePromptWriteArtifacts:
            artifacts, worktree = _manifest_worker_write_paths(
                project_root, authority, assignment_ref, prompt_path
            )
        maximum_precision = capability.max_boundary_precision
    else:
        raise DispatchError("agent write policy has no runner capability")
    planned = (
        planned_paths_from_run_manifest(project_root, authority)
        if execution.role == "implementer"
        else ((), True)
    )
    try:
        return build_invocation_write_contract(
            role=execution.role,
            project_root=project_root,
            worktree=worktree,
            artifact_paths=artifacts,
            task_root=task_root_from_run_manifest(project_root, authority),
            maximum_precision=maximum_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"agent write policy is invalid: {exc}") from exc


def _leader_artifact_paths(
    project_root: Path, authority: Mapping[str, Any]
) -> tuple[Path, ...]:
    value = authority.get("taskRootPath") or authority.get("runDirectoryPath")
    if not isinstance(value, str) or not value.strip():
        raise DispatchError("leader write policy has no artifact root")
    artifact_home = project_root / ".okstra"
    if not artifact_home.is_dir():
        raise DispatchError("leader write policy artifact home does not exist")
    return (artifact_home,)


def _manifest_worker_write_paths(
    project_root: Path,
    authority: Mapping[str, Any],
    assignment_ref: str,
    prompt_path: Path,
) -> tuple[tuple[Path, ...], Path | None]:
    prefix = "initial/"
    if not assignment_ref.startswith(prefix):
        raise DispatchError("agent prompt write artifact headers are incomplete")
    worker_id = assignment_ref.removeprefix(prefix)
    team_state_path = resolve_required_path(
        project_root, authority, "teamStatePath"
    )
    team_state = load_json_object(team_state_path, "team-state")
    matches = [
        row for row in team_state.get("workers") or []
        if isinstance(row, Mapping) and row.get("workerId") == worker_id
    ]
    if len(matches) != 1:
        raise DispatchError("agent prompt has no unique worker artifact authority")
    result = resolve_project_path(
        project_root, require_string(matches[0], "resultPath")
    )
    paths = {
        *dispatch_completion_paths(worker_id, result, authority, project_root),
        result,
        Path(audit_sidecar_rel(str(result))),
        status_path_for_prompt(prompt_path),
        log_path_for_prompt(prompt_path),
    }
    active_value = authority.get("activeRunContextPath")
    active = (
        load_json_object(
            resolve_project_path(project_root, active_value),
            "active run context",
        )
        if isinstance(active_value, str) and active_value
        else {}
    )
    error_logs = active.get("errorLogs")
    sidecars = (
        error_logs.get("sidecarsByWorkerId")
        if isinstance(error_logs, Mapping)
        else None
    )
    error_value = sidecars.get(worker_id) if isinstance(sidecars, Mapping) else None
    if isinstance(error_value, str) and error_value:
        paths.add(resolve_project_path(project_root, error_value))
    worktree_value = worktree_path(authority, active)
    worktree = Path(worktree_value) if worktree_value else None
    return tuple(sorted(paths, key=str)), worktree


def _prompt_write_paths(
    project_root: Path, prompt_path: Path
) -> tuple[tuple[Path, ...], Path | None]:
    try:
        lines = prompt_path.read_text(encoding="utf-8").splitlines()
    except (OSError, UnicodeError) as exc:
        raise DispatchError("agent prompt cannot be read for write policy") from exc
    values: dict[str, str] = {}
    labels = {
        "Result Path",
        "Worker Result Path",
        "Audit sidecar path",
        "Errors sidecar path",
        "Worktree",
    }
    for line in lines:
        for label in labels:
            prefix = f"**{label}:** "
            if line.startswith(prefix):
                values[label] = line.removeprefix(prefix).strip()
    required = {"Result Path", "Audit sidecar path", "Errors sidecar path"}
    if not required <= values.keys():
        raise IncompletePromptWriteArtifacts(
            "agent prompt write artifact headers are incomplete"
        )
    artifact_values = [values[label] for label in sorted(required)]
    if "Worker Result Path" in values:
        artifact_values.append(values["Worker Result Path"])
    artifacts = {
        _project_or_absolute(project_root, value) for value in artifact_values
    }
    artifacts.update(prompt_derived_paths(prompt_path))
    worktree_value = values.get("Worktree")
    worktree = (
        _project_or_absolute(project_root, worktree_value)
        if worktree_value
        else None
    )
    return tuple(sorted(artifacts, key=str)), worktree


def _project_or_absolute(project_root: Path, value: str) -> Path:
    path = Path(value)
    return path if path.is_absolute() else project_root / path


def prompt_anchor_values(
    prompt_path: Path, labels: Iterable[str],
) -> dict[str, str]:
    """`**<label>:** <value>` 앵커 줄을 라벨별로 읽는다. 첫 번째 값이 이긴다.

    앵커는 materialize 가 문서 머리에 쓰고, 본문(리드 지시문)이 같은 라벨을
    예시로 되풀이할 수 있다. `_prompt_write_paths` 는 마지막 값을 취하는데
    그것은 write-policy 의 허용 집합을 넓힐 뿐이지만, 여기서 읽는 값은 워커가
    실제로 쓰는 경로의 판정 기준이므로 머리의 것이어야 한다.
    """
    values: dict[str, str] = {}
    for line in prompt_path.read_text(encoding="utf-8").splitlines():
        for label in labels:
            prefix = f"**{label}:** "
            if label not in values and line.startswith(prefix):
                values[label] = line.removeprefix(prefix).strip()
    return values


def _execution_dispatch_round(dispatch_kind: str) -> int:
    """attempt 행에 적는 라운드 번호. 예약(`agent-prompt materialize` 의
    `_reservation_round`)과 같은 계산을 써야 한다."""
    if not is_verification_dispatch_kind(dispatch_kind):
        return 1
    round_number = verification_dispatch_round(dispatch_kind)
    if round_number is None:
        raise DispatchError(f"invalid verification dispatch kind: {dispatch_kind}")
    return round_number


def link_agent_dispatch_result(
    *,
    project_root: Path,
    run_manifest_path: Path,
    dispatch_id: str,
    result_path: Path,
) -> dict[str, Any]:
    """Link an existing result artifact to exactly one verified dispatch."""
    project_root = project_root.resolve()
    manifest_path = resolve_project_path(
        project_root, str(run_manifest_path)
    ).resolve(strict=True)
    manifest = load_json_object(manifest_path, "run manifest")
    run_root = resolve_required_path(
        project_root, manifest, "runDirectoryPath"
    ).resolve(strict=True)
    resolved_result = resolve_project_path(
        project_root, str(result_path)
    ).resolve(strict=True)
    if not resolved_result.is_relative_to(run_root):
        raise DispatchError("agent result link must stay inside the current run")
    result_relative = _relative_project_path(project_root, resolved_result)
    team_state_path = resolve_required_path(project_root, manifest, "teamStatePath")
    preview = load_json_object(team_state_path, "team-state")
    native_dispatches = [
        item for item in preview.get("agentDispatches") or []
        if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id
    ]
    if len(native_dispatches) == 1:
        _finish_native_agent_attempt(
            project_root=project_root,
            manifest_path=manifest_path,
            manifest=manifest,
            dispatch=native_dispatches[0],
            result_path=resolved_result,
            result_relative=result_relative,
        )
    with _team_state_lock(team_state_path):
        team_state = load_json_object(team_state_path, "team-state")
        dispatches = [
            item
            for collection in (
                team_state.get("agentDispatches") or [],
                team_state.get("workerDispatches") or [],
            )
            for item in collection
            if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id
        ]
        if len(dispatches) != 1:
            raise DispatchError(
                f"agent result link requires exactly one dispatch record: {dispatch_id}"
            )
        dispatch = dispatches[0]
        _require_recorded_result_path(
            project_root, dispatch, resolved_result, result_relative
        )
        link: dict[str, Any] = {
            "dispatchId": dispatch_id,
            "resultPath": result_relative,
        }
        identity = worker_execution_identity(dispatch)
        if identity is not None:
            link.update({
                "participantRef": identity["participant_ref"],
                "roleExecutionRef": identity["role_execution_ref"],
                "executionLabel": identity["execution_label"],
                "dutyId": identity["duty_id"],
                "invocationRef": identity["invocation_ref"],
                "attempt": identity["attempt"],
            })
        links = team_state.setdefault("agentResultLinks", [])
        if not isinstance(links, list):
            raise DispatchError("team-state agentResultLinks must be an array")
        same_result = [
            item for item in links
            if isinstance(item, Mapping) and item.get("resultPath") == result_relative
        ]
        # A superseded link is the record of a result the lead rejected, kept so
        # the chain shows the corrective round happened rather than hiding it.
        # It no longer owns the path, so the re-dispatch may claim it.
        live_conflicts = [
            item for item in same_result
            if not item.get("supersededBy") and dict(item) != link
        ]
        if live_conflicts:
            raise DispatchError(
                f"agent result is already linked to another dispatch: "
                f"{result_relative}. If that result was rejected and re-dispatched, "
                f"record the rejection with `okstra agent-prompt reject-result "
                f"--dispatch-id {live_conflicts[0].get('dispatchId')} "
                f"--superseded-by {dispatch_id}` first"
            )
        if any(
            isinstance(item, Mapping)
            and item.get("dispatchId") == dispatch_id
            and item.get("resultPath") != result_relative
            for item in links
        ):
            raise DispatchError(
                f"agent dispatch is already linked to another result: {dispatch_id}"
            )
        if link not in links:
            links.append(link)
            write_json(team_state_path, team_state)
    return link


def _finish_native_agent_attempt(
    *,
    project_root: Path,
    manifest_path: Path,
    manifest: Mapping[str, Any],
    dispatch: Mapping[str, Any],
    result_path: Path,
    result_relative: str,
) -> None:
    identity = worker_execution_identity(dispatch)
    if identity is None:
        return
    execution = read_execution_manifest(manifest_path)
    attempt = next(
        (
            row for row in execution.attempts
            if row.invocation_ref == identity["invocation_ref"]
            and row.attempt == identity["attempt"]
        ),
        None,
    )
    if attempt is None:
        raise DispatchError("agent result has no canonical attempt policy")
    if dispatch.get("writePolicyDigest") != attempt.write_policy_digest:
        raise DispatchError("agent dispatch write policy digest changed")
    if dispatch.get("writeEnforcement") != attempt.write_enforcement:
        raise DispatchError("agent dispatch write enforcement changed")
    policy = write_policy_from_payload(attempt.write_policy)
    status, summary = _native_mutation_result(
        project_root=project_root,
        dispatch=dispatch,
        result_path=result_path,
        policy=policy,
    )
    _close_native_attempt(
        manifest_path,
        manifest,
        identity,
        status,
        result_relative,
        summary,
    )
    if status != "ok":
        violations = summary.get("violations")
        detail = "; ".join(violations) if isinstance(violations, list) else status
        raise DispatchError(f"agent mutation audit failed: {detail or status}")


def _native_mutation_result(
    *,
    project_root: Path,
    dispatch: Mapping[str, Any],
    result_path: Path,
    policy: WritePolicy,
) -> tuple[str, Mapping[str, Any]]:
    snapshot_value = string_value(dispatch.get("mutationAuditSnapshotPath"))
    if snapshot_value:
        snapshot_payload = load_json_object(
            resolve_project_path(project_root, snapshot_value),
            "mutation audit snapshot",
        )
        try:
            snapshot = MutationSnapshot.from_payload(snapshot_payload)
        except (KeyError, TypeError, ValueError) as exc:
            raise DispatchError("mutation audit snapshot is invalid") from exc
        result = ExecutionMutationAudit().compare(
            snapshot,
            (policy,),
            out_of_plan_edits=_native_out_of_plan_paths(
                result_path, project_root=project_root
            ),
            attempt_succeeded=True,
            result_present=True,
        )
        summary = result.change_summary()
        status = result.status
    else:
        status = "ok"
        summary = {
            "mutationStatus": "ok",
            "changedPaths": [],
            "sourceChanged": False,
            "gitChanged": False,
            "attribution": "call",
            "retryAllowed": False,
            "violations": [],
            "beforeDigest": "",
            "afterDigest": "",
        }
    return status, summary


def _close_native_attempt(
    manifest_path: Path,
    manifest: Mapping[str, Any],
    identity: Mapping[str, Any],
    status: str,
    result_path: str,
    summary: Mapping[str, Any],
) -> None:
    execution = read_execution_manifest(manifest_path)
    existing = next(
        (
            row for row in execution.attempts
            if row.invocation_ref == identity["invocation_ref"]
            and row.attempt == identity["attempt"]
        ),
        None,
    )
    if existing is not None and existing.finished_at is not None:
        if (
            existing.status == status
            and existing.result_path == result_path
            and dict(existing.change_summary) == dict(summary)
        ):
            return
        raise DispatchError("agent attempt terminal mutation result drift")
    finish_attempt_mutation(
        manifest_path,
        invocation_ref=identity["invocation_ref"],
        attempt=identity["attempt"],
        finished_at=utc_now(),
        status=status,
        result_path=result_path,
        error_path=None,
        change_summary=summary,
        task_key=require_string(manifest, "taskKey"),
    )


def _native_out_of_plan_paths(
    result_path: Path, *, project_root: Path
) -> tuple[str, ...]:
    if result_path.suffix != ".json":
        return ()
    try:
        # 외부 입력: 네이티브 모델 결과에서 선택적 수정 경로만 투영한다.
        payload = load_external_json(
            external_worker_json_source(
                result_path,
                trusted_root=project_root,
                lane_root=result_path.parent,
            ),
            artifact="native 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 reject_agent_dispatch_result(
    *,
    project_root: Path,
    run_manifest_path: Path,
    dispatch_id: str,
    superseded_by: str,
    reason: str,
) -> dict[str, str]:
    """Record that a linked result was rejected and re-dispatched.

    The contract tells the lead to re-dispatch a worker whose answer violated the
    response format, with a correction paragraph appended. That path did not
    finish: the prompt is immutable and already dispatched, so
    `--replace-undispatched` refuses (correctly), and a fresh invocation id
    produces a worker that writes the same result path — where `link-result`
    refused because the path already belonged to the first dispatch. The worker
    ran, wrote a good result, and the ledger could not accept it.

    Rejection is a fact worth recording rather than routing around, so it is
    written rather than allowed implicitly: the first link stays in the ledger
    marked `supersededBy` with the lead's reason, and only then may the
    re-dispatch claim the path. Nothing is deleted, so the chain still shows both
    attempts and why the second exists.
    """
    if not superseded_by.strip():
        raise DispatchError("superseding dispatch ID is required")
    if not reason.strip():
        raise DispatchError(
            "a rejection reason is required — the ledger has to say why a "
            "returned result was not accepted"
        )
    project_root = project_root.resolve()
    manifest_path = resolve_project_path(
        project_root, str(run_manifest_path)
    ).resolve(strict=True)
    manifest = load_json_object(manifest_path, "run manifest")
    team_state_path = resolve_required_path(project_root, manifest, "teamStatePath")
    with _team_state_lock(team_state_path):
        team_state = load_json_object(team_state_path, "team-state")
        links = team_state.get("agentResultLinks")
        if not isinstance(links, list):
            raise DispatchError("team-state agentResultLinks must be an array")
        matches = [
            item for item in links
            if isinstance(item, Mapping) and item.get("dispatchId") == dispatch_id
        ]
        if len(matches) != 1:
            raise DispatchError(
                f"expected exactly one agent result link for {dispatch_id}, "
                f"found {len(matches)}"
            )
        row = matches[0]
        if row.get("supersededBy"):
            raise DispatchError(
                f"agent result link is already superseded by "
                f"{row['supersededBy']}: {dispatch_id}"
            )
        row["supersededBy"] = superseded_by
        row["rejectionReason"] = reason
        write_json(team_state_path, team_state)
    return dict(row)


def _require_recorded_result_path(
    project_root: Path,
    dispatch: Mapping[str, Any],
    resolved_result: Path,
    result_relative: str,
) -> None:
    """링크되는 결과가 이 디스패치가 실제로 기다린 산출물인가.

    비교 대상을 여기서 다시 계산하지 않는다. `record-dispatch` 가 이미
    `completionPaths` 로 그 목록을 적어 두었고(리포트 작성자는 narrative 와
    포인터 기록 둘), 그것이 이 디스패치가 완료로 인정하는 경로의 정본이다.

    이 대조가 없으면 잘못된 경로를 링크해도 그 자리에서는 통과하고, 조립이
    `required input is missing` 으로 죽을 때에야 드러난다 — 그때는 워커가 이미
    돌고 난 뒤다. 실측(2026-08-26): 리드가 지어낸 seq(`-002.md`)를 넘겼고
    매니페스트의 선언은 `-004.md` 였다. 여기서 막으면 워커를 다시 돌릴 필요 없이
    같은 명령을 옳은 인자로 다시 부르면 된다.

    `completionPaths` 를 기록하지 않은 행은 판정하지 않는다 — 목록이 없으면
    무엇이 정당한지 알 수 없고, 모르면서 막는 것은 추측이다.
    """
    recorded = dispatch.get("completionPaths")
    if not isinstance(recorded, list) or not recorded:
        return
    allowed = {
        resolve_project_path(project_root, str(value)).resolve()
        for value in recorded
        if isinstance(value, str) and value.strip()
    }
    if not allowed or resolved_result in allowed:
        return
    expected = ", ".join(sorted(_relative_project_path(project_root, path) for path in allowed))
    raise DispatchError(
        f"agent result link is not one of this dispatch's completion paths: "
        f"linked {result_relative}, expected one of {expected}"
    )


def abandon_agent_dispatch_attempt(
    *,
    project_root: Path,
    run_manifest_path: Path,
    invocation_ref: str,
    reason: str,
) -> dict[str, Any]:
    """Close a started attempt whose worker died without producing a result.

    Every other path that closes an attempt needs something the dead worker
    never made: `link-result` needs its result file, and
    `_abandon_unstarted_attempt` only runs when the dispatch itself failed
    before the process existed. A worker that started and then died — its host
    killed, its pane closed, an import error before any log — leaves the attempt
    `started` forever, and `_validate_next_attempt` then refuses every retry with
    `previous attempt is not terminal`. The run has to be re-rendered from
    scratch to make progress.

    **Only a `source-readonly` attempt may be abandoned.** The manifest accepts a
    following attempt only after `failed-no-mutation`, so closing an attempt is
    also an assertion that nothing was written. For a read-only worker that
    assertion is true by policy — it never held permission to write — so no
    guess is involved. For a `project-mutation` attempt it would be a guess, and
    a wrong one lets the retry run on top of a half-mutated tree, which is the
    invariant this manifest exists to protect. That case is refused here and
    belongs to the mutation audit.
    """
    if not reason.strip():
        raise DispatchError(
            "an abandonment reason is required — the ledger has to say why an "
            "attempt was closed without a result"
        )
    project_root = project_root.resolve()
    manifest_path = resolve_project_path(
        project_root, str(run_manifest_path)
    ).resolve(strict=True)
    manifest = load_json_object(manifest_path, "run manifest")
    execution = read_execution_manifest(manifest_path)
    attempts = [
        row for row in execution.attempts if row.invocation_ref == invocation_ref
    ]
    if not attempts:
        raise DispatchError(f"no attempt for invocation {invocation_ref}")
    row = max(attempts, key=lambda item: item.attempt)
    if row.finished_at is not None:
        raise DispatchError(
            f"attempt {invocation_ref} #{row.attempt} is already terminal "
            f"({row.status})"
        )
    mode = _attempt_source_mode(row)
    if mode != "source-readonly":
        raise DispatchError(
            f"attempt {invocation_ref} #{row.attempt} runs under sourcePolicy "
            f"`{mode}` — only a source-readonly attempt may be abandoned, "
            "because closing one asserts that nothing was written. Resolve a "
            "mutating attempt through its mutation audit."
        )
    finish_attempt_mutation(
        manifest_path,
        invocation_ref=invocation_ref,
        attempt=row.attempt,
        finished_at=utc_now(),
        status="failed-no-mutation",
        result_path=None,
        error_path=None,
        change_summary={"abandonedReason": reason.strip()},
        task_key=require_string(manifest, "taskKey"),
    )
    return {
        "invocationRef": invocation_ref,
        "attempt": row.attempt,
        "status": "failed-no-mutation",
        "reason": reason.strip(),
    }


def _attempt_source_mode(row: Any) -> str:
    policy = getattr(row, "write_policy", None)
    source = policy.get("sourcePolicy") if isinstance(policy, Mapping) else None
    mode = source.get("mode") if isinstance(source, Mapping) else None
    return str(mode or "unknown")


def _relative_project_path(project_root: Path, path: Path) -> str:
    try:
        return path.resolve(strict=False).relative_to(project_root).as_posix()
    except ValueError as exc:
        raise DispatchError(f"path is outside project root: {path}") from exc


def worker_state(team_state: Mapping[str, Any], worker_id: str) -> Mapping[str, Any]:
    workers = team_state.get("workers")
    if not isinstance(workers, list):
        raise DispatchError("team-state workers must be an array")
    for worker in workers:
        if isinstance(worker, Mapping) and worker.get("workerId") == worker_id:
            return worker
    raise DispatchError(f"team-state has no workerId={worker_id}")


def _task_project_root(team_state_path: Path) -> Path | None:
    """team-state 경로에서 프로젝트 루트를 되짚는다 (`.okstra` 를 기준점으로).

    stage run 은 `runs/<phase>/stage-N/state/` 로 한 층이 더 깊으므로 부모를
    세는 방식은 빗나간다. 기준점을 못 찾으면 None — 예상 밖 배치에서 판정을
    포기하는 쪽이, 실행을 통째로 막는 쪽보다 낫다. 최종 검증이 같은 모순을
    다시 잡는다.
    """
    for parent in team_state_path.resolve().parents:
        if parent.name == ".okstra":
            return parent.parent
    return None


class CompletedWithoutResultError(DispatchError):
    """명부의 `resultPath` 파일 없이 `completed` 로 적으려 했다.

    `_finish_attempt` 가 이 거절을 데이터로 돌려세우기 위해 구분한다 — 디스패치
    자신의 산출물은 있는데 명부가 가리키는 파일이 다른 경우다.
    """


def _reject_completed_without_result(
    team_state_path: Path, worker_id: str, worker: Mapping[str, Any]
) -> None:
    """산출물이 없는 역할을 완료로 적지 못하게 한다.

    okstra 는 디스패치 결과를 스스로 기록한다. 그 원장이 두 시도 모두 `error`
    라고 적은 역할이 명부에는 `completed` 로 남을 수 있었다 — 명부 상태가
    원장에서 파생되지 않기 때문이다. 모순은 phase 끝 `validate-run` 이
    "completed but worker result file is missing" 으로 잡지만, 그때는 워커를
    다시 돌릴 수 없어 실행 전체가 막힌다.

    조인 대신 명부 자신의 `resultPath` 를 본다. 그 값이 없는 역할(critic scope
    등)은 판정 대상이 아니다.

    실측(2026-08-27, `fontsninja-v3-site` `dev-10341`): 명부 5건 중 `resultPath`
    를 가진 4건에서 3건은 파일이 있었고, grok-planner 만 없었다.
    """
    relative = str(worker.get("resultPath") or "").strip()
    if not relative:
        return
    candidate = Path(relative)
    if not candidate.is_absolute():
        project_root = _task_project_root(team_state_path)
        if project_root is None:
            return
        candidate = project_root / candidate
    if candidate.is_file():
        return
    raise CompletedWithoutResultError(
        f"worker {worker_id} cannot be recorded completed: its result "
        f"{relative} does not exist. Record the terminal status the dispatch "
        "actually reached (`error` / `timeout` / `not-run`) with a reason."
    )


def transition_worker_status(
    team_state_path: Path,
    worker_id: str,
    status: str,
    reason: str,
    *,
    model_execution_value: str = "",
    at: str | datetime | None = None,
) -> None:
    if status not in WORKER_STATUSES:
        allowed = ", ".join(sorted(WORKER_STATUSES))
        raise DispatchError(
            f"unsupported worker status `{status}`; expected one of: {allowed}"
        )
    reason = reason.strip()
    if status in REASON_REQUIRED_STATUSES and not reason:
        raise DispatchError(f"worker status `{status}` requires a non-empty reason")
    timestamp = _utc_timestamp(at)
    with _team_state_lock(team_state_path):
        payload = load_json_object(team_state_path, "team-state")
        workers = payload.get("workers")
        if not isinstance(workers, list):
            raise DispatchError(f"team-state workers must be an array: {team_state_path}")
        for worker in workers:
            if isinstance(worker, dict) and worker.get("workerId") == worker_id:
                if status == "completed":
                    _reject_completed_without_result(
                        team_state_path, worker_id, worker
                    )
                worker["status"] = status
                worker["reason"] = reason
                if status == "in-progress":
                    worker["startedAt"] = timestamp
                    worker.pop("endedAt", None)
                elif status == "not-run":
                    worker.pop("startedAt", None)
                    worker.pop("endedAt", None)
                else:
                    worker["endedAt"] = timestamp
                if model_execution_value:
                    # `model` is the catalog display name the task-manifest
                    # declares; only the execution identifier belongs here. Writing
                    # both from one value destroys the display name on every worker
                    # whose two differ, and `validate-run` compares team-state's
                    # `model` against the manifest.
                    worker["modelExecutionValue"] = model_execution_value
                write_json(team_state_path, payload)
                return
    raise DispatchError(f"team-state has no workerId={worker_id}: {team_state_path}")


def _utc_timestamp(value: str | datetime | None) -> str:
    if value is None:
        instant = datetime.now(timezone.utc)
    elif isinstance(value, datetime):
        instant = value
    elif isinstance(value, str):
        try:
            instant = datetime.fromisoformat(value.replace("Z", "+00:00"))
        except ValueError as exc:
            raise DispatchError(f"invalid UTC timestamp `{value}`") from exc
    else:
        raise DispatchError(f"invalid UTC timestamp `{value}`")
    if instant.tzinfo is None or instant.utcoffset() != timezone.utc.utcoffset(instant):
        raise DispatchError(f"invalid UTC timestamp `{value}`")
    return instant.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def record_dispatch_facts(team_state_path: Path, dispatch_mode: str) -> None:
    """Persist the team-state facts okstra owns at dispatch time.

    `dispatchMode` names the backend the workers actually went out on.
    `teamCreate` is the implicit-team audit marker: Claude Code v2.1.178 removed
    the TeamCreate tool, so `{attempted: false, status: "implicit"}` is a
    constant of a non-concurrent run rather than a lead judgment. validate-run
    requires it once any worker has been dispatched, and Phase 7 token
    attribution reads it to locate worker sessions — leaving it for the lead to
    hand-write failed every dispatch path whose host adapter does not spell the
    rule out. A marker the launch prompt pre-recorded (the concurrent-run
    `skipped` decision) belongs to that run and is never overwritten.
    """
    with _team_state_lock(team_state_path):
        payload = load_json_object(team_state_path, "team-state")
        payload["dispatchMode"] = dispatch_mode
        if not _has_recorded_team_create(payload):
            payload["teamCreate"] = {"attempted": False, "status": "implicit"}
        write_json(team_state_path, payload)


def _has_recorded_team_create(team_state: Mapping[str, Any]) -> bool:
    existing = team_state.get("teamCreate")
    return (isinstance(existing, dict)
            and bool(str(existing.get("status") or "").strip()))


# --- job facts ----------------------------------------------------------------

def dispatch_mode(jobs: Sequence[WorkerJob]) -> str:
    backends = {job.backend for job in jobs}
    if len(backends) == 1:
        return next(iter(backends))
    return BACKEND_MIXED


def unusable_result_defect(
    worker_id: str, result_path: Path, dispatch_kind: str = "",
) -> str | None:
    """산출물이 있어도 소비자가 읽을 수 없으면 없는 것이다 — 서사와 재검증 표.

    report-writer 의 서사가 줄 문법을 어기면(frontmatter·헤딩으로 된 보통
    보고서) 조립이 Phase 7 에서 거절하고, 그때는 배치의 재시도가 이미 지나
    리드가 손으로 재저작을 띄워야 한다 — 실측(2026-09-09, jobs implementation
    stage-2)에서 리드는 그것을 하지 않고 run 을 닫았다. 수집 시점에 "없는
    산출물" 로 세면 `_should_retry` 가 같은 배치 안에서 다시 띄운다.

    재검증(`reverify-r<N>`) 결과도 같은 자리에 있다. `okstra convergence
    collect-results` 는 표로 읽히지 않는 결과를 거절하는데, 원장은 그 attempt 를
    `ok` 로 닫아 두므로 그 워커를 빼고 수집하면 `apply-round` 가 "missing vote
    for completed worker" 로 막는다. 즉 리드에게 남는 수가 없다 — 실측
    (2026-09-10, fontsninja-v3-site dev-10631 implementation-option-selection):
    antigravity 가 35건 중 34건의 `**Explanation**` 을 빼먹었고 run 이 그 자리에
    멈췄다. 여기서 결함으로 세면 재시도가 배치 안에서 돌고, 그마저 실패하면
    attempt 가 실패로 닫혀 `collect-results` 가 그 워커를 `error` 로 적는다 —
    엔진은 그 표를 `verification-error` 로 기록하고 라운드는 진행한다.
    """
    if not result_path.is_file():
        return None
    if dispatch_kind.startswith("reverify-r"):
        try:
            text = result_path.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError) as exc:
            return f"reverify result is unreadable: {exc}"
        defect = finding_vote_defect(text)
        if defect is None:
            return None
        return f"reverify result does not parse: {defect}"
    if worker_id != REPORT_WRITER_WORKER_ID:
        return None
    try:
        text = result_path.read_text(encoding="utf-8")
    except (OSError, UnicodeDecodeError) as exc:
        return f"narrative is unreadable: {exc}"
    defect = narrative_structure_defect(text)
    if defect is None:
        return None
    return f"narrative does not parse: {defect}"


def missing_completion_paths(job: WorkerJob) -> tuple[Path, ...]:
    missing: list[Path] = []
    for path in job.completion_paths:
        if path.is_file():
            if path == job.result_path and unusable_result_defect(
                job.worker_id, path, job.dispatch_kind,
            ):
                missing.append(path)
            continue
        # reports seq 와 workerResults seq 가 갈라지면 워커는 다른 쪽
        # 파일명으로 쓴다. 둘 중 하나가 있으면 산출물은 있는 것이다.
        if path in {job.result_path, job.worker_result_path} and any(
            alias.is_file() for alias in job.result_aliases
        ):
            continue
        missing.append(path)
    return tuple(missing)


def dispatch_session_id(provider: str) -> str:
    """The session this dispatch mints for *provider*, or '' for one that has no
    such flag.

    Only the claude CLI reads `--session-id`; handing the value to another
    wrapper would start it with an argument it does not know. An empty value is
    the same as not issuing one — `WorkerJob.command` appends the flag only when
    it is set.

    This lives beside `worker_jobs_from_file` for the same reason
    `dispatch_result_path` does: both job constructors apply it, and a rule
    written twice is a rule that drifts.
    """
    return generate_claude_session_id() if provider == "claude" else ""


def worker_session_ids(
    team_state: Mapping[str, Any], worker_id: str | None = None
) -> list[str]:
    """The claude sessions this run's dispatches issued, in the order recorded.

    A worker started in a cmux pane is its own `claude -p` process and tags its
    session jsonl with neither `agentName` nor `teamName`, so nothing in the
    transcript says which worker wrote it. The id `dispatch_session_id` minted
    and `_dispatch_record` wrote is the only handle on it.

    Every id, not one: a retry opens a new session, so one worker legitimately
    owns several rows with different `sessionId` values, and taking only the
    first or last bills a single attempt of a retried worker.

    Pass ``worker_id`` for one worker's sessions, or omit it for the run's.
    This reads a `workerDispatches[]` row shape, so it belongs beside the code
    that writes that shape rather than in each of the two modules that read it.

    A non-list log yields nothing rather than raising: `session.observe_lead_session`
    calls this for its exclusion set and catches only OSError, so a `TypeError`
    from iterating a number here would take down `team dispatch` / `team await` /
    `report-finalize`. Observation not breaking its caller is a contract kept
    here, the same way `add_observed_session` keeps it for the keys it appends to.
    """
    session_ids: list[str] = []
    for record in worker_dispatch_records(team_state, worker_id):
        session_id = str(record.get("sessionId") or "").strip()
        if session_id and session_id not in session_ids:
            session_ids.append(session_id)
    return session_ids


def worker_dispatch_records(
    team_state: Mapping[str, Any], worker_id: str | None = None
) -> list[Mapping[str, Any]]:
    """한 워커(또는 run 전체)의 `workerDispatches[]` 행을 기록된 순서대로.

    사용량 수집이 워커의 실행 증거를 찾는 열쇠는 이 행들이다 — 프롬프트 경로
    (status 사이드카 → 래퍼 창), 세션 id(claude 트랜스크립트), 워크트리 경로
    (grok·kimi 는 그 안에서 돌아 세션 디렉터리가 그 경로로 인코딩된다).
    `workers[]` 행의 `promptPath` 는 첫 dispatch 하나만 가리키므로 재검증·
    critic-gap·plan-verify 로 다시 띄운 세션은 거기서 보이지 않는다(실측
    2026-09-08 jobs implementation-planning r01: codex 5회 dispatch 중 1회만
    집계, v2 명부 행 `scope` 는 `promptPath` 가 비어 0회).

    v1 행은 `workerId`, v2 행은 `assignmentRef` 마지막 마디로 워커에 묶인다
    (`_dispatch_worker_key`). 리스트가 아니면 빈 결과 — `worker_session_ids`
    와 같은 이유로 호출자를 깨지 않는다.
    """
    dispatches = team_state.get("workerDispatches")
    records: list[Mapping[str, Any]] = []
    for record in dispatches if isinstance(dispatches, list) else []:
        if not isinstance(record, Mapping):
            continue
        if worker_id is not None and _dispatch_worker_key(record) != worker_id:
            continue
        records.append(record)
    return records


def _dispatch_worker_key(record: Mapping[str, Any]) -> str:
    """The `workers[]` row a dispatch record belongs to.

    A v1 record names it as `workerId`. A v2 record carries execution identity
    instead and no `workerId` (`_dispatch_record`), so its row is the one the
    roster keys off the assignment ref's last segment — `critic/scope` is the
    row `scope`, `initial/claude-analyser` the row `claude-analyser` — the same
    projection `v2_worker_state_key` makes for the jobs file. Matching on
    `workerId` alone found no v2 record, so a pane worker's session id was never
    read and every v2 pane worker stayed `unavailable` (observed 2026-09-02,
    dev-10626 r04: the critic's session was on disk with 79 usage records).
    """
    worker_id = str(record.get("workerId") or "").strip()
    if worker_id:
        return worker_id
    assignment_ref = record.get("assignmentRef")
    if not isinstance(assignment_ref, str):
        return ""
    return assignment_ref.rsplit("/", 1)[-1].strip()


def dispatch_result_path(
    worker_id: str,
    worker_result_path: Path,
    manifest: Mapping[str, Any],
    project_root: Path,
) -> Path:
    """What `resultPath` means for this worker.

    For everyone but the report writer it is the worker-result file. The report
    writer authors three artifacts and its canonical result is the final-report
    data.json, not its own `.md` pointer — a distinction that is not cosmetic:
    `dispatch_core` hands `job.result_path` to report finalization as the data
    path, so a `.md` there is parsed as JSON and a complete report settles
    `error`.

    This lives beside `worker_jobs_from_file` because both job constructors must
    apply it. It was `dispatch_core`-private while the roster path applied it and
    the `--jobs-file` path took whatever the file said, which is how the two
    produced different jobs for the same worker.
    """
    if worker_id != REPORT_WRITER_WORKER_ID:
        return worker_result_path
    if uses_report_contract_v3(manifest):
        return report_narrative_path(project_root, manifest)
    return final_report_data_path(
        resolve_required_path(project_root, manifest, "expectedReportRecordPath")
    )


def dispatch_completion_paths(
    worker_id: str,
    worker_result_path: Path,
    manifest: Mapping[str, Any],
    project_root: Path,
) -> tuple[Path, ...]:
    """Every artifact that must exist before this worker counts as done.

    The report writer's two are the report record and the worker-result
    pointer. The full reading copy is rendered on demand and is not a
    completion artifact.
    """
    if worker_id != REPORT_WRITER_WORKER_ID:
        return (worker_result_path,)
    if uses_report_contract_v3(manifest):
        return (report_narrative_path(project_root, manifest), worker_result_path)
    data_json = final_report_data_path(
        resolve_required_path(project_root, manifest, "expectedReportRecordPath")
    )
    return (data_json, worker_result_path)


def _plan_verify_result_aliases(
    worker_result_path: Path, manifest: Mapping[str, Any],
) -> tuple[Path, ...]:
    """reports seq 파일의 workerResults seq 짝.

    plan-verify 결과 파일명만 해당한다. 다른 산출물의 seq 를 바꾸면 안 된다.
    """
    if "-plan-verify-r" not in worker_result_path.name:
        return ()
    sequences = manifest.get("runSequencesByCategory")
    if not isinstance(sequences, Mapping):
        return ()
    reports = str(sequences.get("reports") or "").strip()
    workers = str(sequences.get("workerResults") or "").strip()
    if not reports or not workers or reports == workers:
        return ()
    name = worker_result_path.name
    if name.endswith(f"-{reports}.md"):
        other = name[: -len(f"-{reports}.md")] + f"-{workers}.md"
    elif name.endswith(f"-{workers}.md"):
        other = name[: -len(f"-{workers}.md")] + f"-{reports}.md"
    else:
        return ()
    return (worker_result_path.with_name(other),)


def validate_initial_prompts(
    manifest: Mapping[str, Any], jobs: Sequence[WorkerJob]
) -> None:
    records = [
        PromptRecord(
            job.worker_id,
            job.dispatch_kind,
            job.prompt_path,
            metadata_path=(job.prompt_metadata_path if job.invocation_id else None),
            expected_duty_audience=(job.audience if job.invocation_id else None),
        )
        for job in jobs
    ]
    errors = validate_initial_prompt_records(
        manifest=manifest,
        records=records,
        require_evidence_ledger=True,
    )
    if errors:
        task_type = require_string(manifest, "taskType")
        raise DispatchError(f"{task_type} prompt contract: " + "; ".join(errors))


def validate_dispatch_prompts(
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any],
    jobs: Sequence[WorkerJob],
) -> None:
    if isinstance(manifest.get("agentContract"), Mapping):
        _validate_agent_invocations(manifest, jobs)
    initial_jobs = [
        job for job in jobs if not is_verification_dispatch_kind(job.dispatch_kind)
    ]
    if initial_jobs:
        validate_initial_prompts(manifest, initial_jobs)

    reverify_jobs = [
        job for job in jobs if is_verification_dispatch_kind(job.dispatch_kind)
    ]
    if not reverify_jobs:
        return
    task_type = require_string(manifest, "taskType")
    workflow = active_context.get("workflow")
    if not isinstance(workflow, Mapping):
        raise DispatchError("reverify prompt contract: active run workflow is missing")
    forbidden_actions = string_value(workflow.get("forbiddenActions"))
    if not forbidden_actions:
        raise DispatchError(
            "reverify prompt contract: workflow.forbiddenActions is missing"
        )

    errors: list[str] = []
    for job in reverify_jobs:
        try:
            text = job.prompt_path.read_text(encoding="utf-8")
        except OSError as exc:
            errors.append(f"{job.worker_id}: cannot read prompt {job.prompt_path}: {exc}")
            continue
        errors.extend(
            f"{job.worker_id}: {error}"
            # The job carries the model this dispatch will actually run, so the
            # hand-written `**Model:**` header is checkable here — and must be,
            # because a header naming a model the runtime does not serve kills
            # the worker with a provider 400 minutes later, where it reads as a
            # worker fault rather than the typo it is.
            for error in validate_reverify_prompt(
                text,
                task_type=task_type,
                forbidden_actions=forbidden_actions,
                expected_model=job.model_execution_value or None,
                dispatch_kind=job.dispatch_kind,
            )
        )
    if errors:
        raise DispatchError("reverify prompt contract: " + "; ".join(errors))


def _validate_agent_invocations(
    manifest: Mapping[str, Any],
    jobs: Sequence[WorkerJob],
) -> None:
    manifest_path_value = require_string(manifest, "runManifestPath")
    assignments = manifest.get("invocationAssignments")
    if not isinstance(assignments, Mapping):
        raise DispatchError("run manifest has no invocationAssignments object")
    errors: list[str] = []
    for job in jobs:
        missing = [
            name
            for name, value in (
                ("invocationId", job.invocation_id),
                ("audience", job.audience),
                ("assignmentRef", job.assignment_ref),
                ("promptMetadataPath", str(job.prompt_metadata_path)),
                ("catalogDigest", job.catalog_digest),
                ("assignmentDigest", job.assignment_digest),
                ("dutyDigest", job.duty_digest),
                ("instructionDigest", job.instruction_digest),
                ("promptDigest", job.prompt_digest),
                ("enforcementMode", job.enforcement_mode),
            )
            if not value or value == "."
        ]
        if missing:
            errors.append(
                f"{job.worker_id}: agent invocation metadata fields are missing: "
                + ", ".join(missing)
            )
            continue
        if job.enforcement_mode != "core-pre-dispatch":
            errors.append(
                f"{job.worker_id}: code-owned dispatch requires core-pre-dispatch"
            )
        assignment_payload = assignments.get(job.assignment_ref)
        try:
            assignment = agent_model_assignment_from_payload(assignment_payload)
        except AgentInvocationError as exc:
            errors.append(f"{job.worker_id}: {exc}")
            continue
        if assignment.model_execution_value != job.model_execution_value:
            errors.append(
                f"{job.worker_id}: model execution value does not match invocation assignment"
            )
        if assignment.host_model_value != job.host_model_value:
            errors.append(
                f"{job.worker_id}: host model value does not match invocation assignment"
            )
        manifest_path = resolve_project_path(job.project_root, manifest_path_value)
        verification = verify_agent_invocation(
            job.prompt_metadata_path,
            project_root=job.project_root,
            expected_run_manifest_path=manifest_path,
            expected_assignment=assignment,
            expected_invocation_id=job.invocation_id,
            expected_worker_id=job.worker_id,
            expected_assignment_ref=job.assignment_ref,
            expected_audience=job.audience,
        )
        errors.extend(f"{job.worker_id}: {error}" for error in verification)
        try:
            metadata = load_json_object(job.prompt_metadata_path, "agent metadata")
        except DispatchError as exc:
            errors.append(f"{job.worker_id}: {exc}")
            continue
        digests = metadata.get("digests")
        if not isinstance(digests, Mapping) or dict(digests) != job.digests:
            errors.append(f"{job.worker_id}: job digests do not match agent metadata")
    if errors:
        raise DispatchError("agent invocation contract: " + "; ".join(errors))


def divergent_fields(
    pairs: Sequence[tuple[str, Any, Any]],
    *,
    left: str,
    right: str,
) -> list[str]:
    """어긋난 필드만 이름과 두 값으로 적는다.

    이 두 검사는 각각 일곱 개와 다섯 개의 조건을 `or` 로 묶어 한 문장으로
    보고했다. 그래서 재시도 프롬프트의 `dispatchKind` 하나가 틀린 경우와
    provider 가 틀린 경우가 같은 문장으로 나왔고, 읽는 쪽이 어느 값을 봐야 할지
    알 수 없었다. 판정에 쓴 값은 이미 손에 있으므로 감출 이유가 없다.
    """
    return [
        f"{name}: {left}={found!r} {right}={expected!r}"
        for name, found, expected in pairs
        if found != expected
    ]


def _v2_job_authority_errors(
    job: WorkerJob,
    metadata: Mapping[str, Any],
    manifest: Mapping[str, Any],
) -> list[str]:
    try:
        identity = invocation_metadata_identity(metadata)
    except AgentInvocationError as exc:
        return [str(exc)]
    if identity is None:
        return ["jobs file v2 identity has v1 invocation metadata"]
    errors: list[str] = []
    if job.invocation_ref != identity.invocation_ref:
        errors.append("jobs file invocationRef does not match invocation metadata")
    divergent = divergent_fields(
        (
            ("participantRef", identity.participant_ref, job.participant_ref),
            ("roleExecutionRef", identity.role_execution_ref, job.role_execution_ref),
            ("executionLabel", identity.execution_label, job.execution_label),
            ("dutyId", identity.duty_id, job.duty_id),
            ("attempt", identity.attempt, job.attempt),
        ),
        left="metadata",
        right="job",
    )
    if divergent:
        errors.append(
            "jobs file v2 identity does not match invocation metadata — "
            + "; ".join(divergent)
        )
    prompt = metadata.get("prompt")
    metadata_prompt_value = (
        string_value(prompt.get("path")) if isinstance(prompt, Mapping) else ""
    )
    metadata_prompt_path = resolve_project_path(
        job.project_root, metadata_prompt_value
    )
    if (
        not metadata_prompt_value
        or job.prompt_path.resolve() != metadata_prompt_path.resolve()
    ):
        errors.append("jobs file promptPath does not match invocation metadata")
    role_executions = manifest.get("roleExecutions")
    execution = next((
        row for row in role_executions
        if isinstance(row, Mapping)
        and row.get("roleExecutionRef") == identity.role_execution_ref
    ), None) if isinstance(role_executions, list) else None
    participants = manifest.get("participantAssignments")
    participant = next((
        row for row in participants
        if isinstance(row, Mapping)
        and row.get("participantRef") == identity.participant_ref
    ), None) if isinstance(participants, list) else None
    assignment = metadata.get("modelAssignment")
    try:
        duty_role = role_for_duty(job.duty_id)
    except RoleCatalogError:
        duty_role = ""
    if isinstance(assignment, Mapping):
        try:
            parsed_assignment = agent_model_assignment_from_payload(assignment)
        except AgentInvocationError as exc:
            errors.append(str(exc))
        else:
            errors.extend(
                "jobs file " + error
                for error in v2_role_assignment_authority_errors(
                    manifest,
                    role_execution_ref=identity.role_execution_ref,
                    participant_ref=identity.participant_ref,
                    assignment=parsed_assignment,
                )
            )
    missing = [
        name for name, value in (
            ("roleExecution", execution),
            ("participant", participant),
            ("modelAssignment", assignment),
        )
        if not isinstance(value, Mapping)
    ]
    if missing:
        errors.append(
            "jobs file v2 identity does not match role execution authority: "
            f"the run manifest has no {', '.join(missing)} for this job"
        )
        return errors
    # 어긋난 필드를 이름으로 댄다. 일곱 조건을 한 문장으로 접던 동안, dispatchKind
    # 하나가 틀린 경우와 provider 가 틀린 경우가 같은 문장으로 보고됐다 —
    # 재시도 프롬프트를 만들다 `initial-retry` 로 적은 리드가 원인을 좁히지
    # 못했다. 판정에 쓴 값은 여기 다 있으므로 굳이 감출 이유가 없다.
    divergent = divergent_fields(
        (
            ("participantRef",
             execution.get("participantRef"), job.participant_ref),
            ("executionLabel",
             execution.get("executionLabel"), job.execution_label),
            ("role", execution.get("role"), job.role),
            ("modelExecutionValue",
             assignment.get("modelExecutionValue"), job.model_execution_value),
            ("dispatchKind",
             metadata.get("dispatchKind"), job.dispatch_kind),
            ("provider", execution.get("provider"), job.provider),
        ),
        left="manifest",
        right="job",
    )
    if execution.get("role") != duty_role:
        divergent.append(
            f"role: roleExecution={execution.get('role')!r} "
            f"duty {job.duty_id!r} requires {duty_role!r}"
        )
    if divergent:
        errors.append(
            "jobs file v2 identity does not match role execution authority — "
            + "; ".join(divergent)
        )
    return errors


def _validate_v2_jobs_file_authority(
    manifest: Mapping[str, Any], jobs: Sequence[WorkerJob]
) -> None:
    if (
        manifest.get("schemaVersion") != "2.0"
        or manifest.get("executionIdentityVersion") != 2
    ):
        return
    _validate_agent_invocations(manifest, jobs)
    errors: list[str] = []
    for job in jobs:
        metadata = load_json_object(job.prompt_metadata_path, "agent metadata")
        errors.extend(
            f"{job.worker_id}: {error}"
            for error in _v2_job_authority_errors(job, metadata, manifest)
        )
    if errors:
        raise DispatchError("jobs file v2 authority: " + "; ".join(errors))
    role_executions = {
        string_value(row.get("roleExecutionRef")): row
        for row in manifest.get("roleExecutions") or []
        if isinstance(row, Mapping)
    }
    if any(
        isinstance(role_executions.get(job.role_execution_ref), Mapping)
        and role_executions[job.role_execution_ref].get("role") == "translator"
        for job in jobs
    ):
        raise DispatchError(
            "jobs file translator must use canonical reservation dispatch"
        )


def worker_jobs_from_file(
    project_root: Path,
    jobs_file: Path,
    *,
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any] | None = None,
    backend: str,
    idle_timeout_seconds: int | None,
    default_dispatch_kind: str,
    resolve_wrapper: Callable[[str], Path],
    default_provider: Callable[[str], str],
) -> list[WorkerJob]:
    payload = load_json_object(
        resolve_project_path(project_root, str(jobs_file)),
        "jobs file",
    )
    return worker_jobs_from_payload(
        project_root, payload, manifest=manifest, active_context=active_context,
        backend=backend, idle_timeout_seconds=idle_timeout_seconds,
        default_dispatch_kind=default_dispatch_kind, resolve_wrapper=resolve_wrapper,
        default_provider=default_provider,
    )


def worker_jobs_from_payload(
    project_root: Path, payload: Mapping[str, Any], *,
    manifest: Mapping[str, Any], active_context: Mapping[str, Any] | None = None,
    backend: str, idle_timeout_seconds: int | None, default_dispatch_kind: str,
    resolve_wrapper: Callable[[str], Path], default_provider: Callable[[str], str],
) -> list[WorkerJob]:
    """파일 소비자와 생성기가 동일한 배치 계약을 검사한다."""
    dispatch_kind = string_value(payload.get("dispatchKind")) or default_dispatch_kind
    workers = payload.get("workers")
    if not isinstance(workers, list):
        raise DispatchError("jobs file workers must be an array")
    jobs = [
        _worker_job_from_file(
            project_root,
            item,
            manifest=manifest,
            active_context=active_context,
            backend=backend,
            idle_timeout_seconds=idle_timeout_seconds,
            dispatch_kind=dispatch_kind,
            resolve_wrapper=resolve_wrapper,
            default_provider=default_provider,
        )
        for item in workers
        if isinstance(item, Mapping)
    ]
    _validate_jobs_file_prompt_anchors(jobs)
    _validate_v2_jobs_file_authority(manifest, jobs)
    return jobs


def _normalized_path(path: Path) -> Path:
    return Path(os.path.normpath(path))


def _validate_jobs_file_prompt_anchors(jobs: Sequence[WorkerJob]) -> None:
    """jobs-file 의 결과 경로가 프롬프트 앵커와 같은 파일인지 본다.

    워커는 프롬프트의 `**Result Path:**` 에 쓰고(포인터를 가진 report-writer 는
    `**Worker Result Path:**` 에 포인터를), 수집기는 jobs-file 의
    `workerResultPath` 를 기다린다. 두 값은 리드가 따로 적으므로 어긋날 수 있고,
    어긋나면 워커가 결과를 다 써도 `required worker artifact was not produced`
    로 끝나 같은 프롬프트가 한 번 더 돈다. 실측(2026-09-02, fontsninja-v3-site
    dev-10626-1 error-analysis r04): 검증 디스패치 9건 중 8건이 이 형태였고
    재시도 6건이 같은 이유로 실패했다.

    report-writer 는 한 가지를 더 본다 — 프롬프트의 `**Result Path:**`(서술문)가
    `dispatch_result_path` 가 정한 조립 입력과 같은가. 다르면 서술문은 써지지만
    `report-finalize` 가 읽는 자리에는 아무것도 없다.

    프롬프트 파일이 아직 없는 잡은 건너뛴다. 실제 디스패치는 같은 파일을
    write-policy 검사(`_prompt_write_paths`)가 반드시 읽으므로 빠져나갈 자리가
    없고, 단위 테스트의 빈 경로만 여기서 면제된다.
    """
    errors: list[str] = []
    for job in jobs:
        if not job.prompt_path.is_file():
            continue
        anchors = prompt_anchor_values(
            job.prompt_path, ("Result Path", "Worker Result Path"),
        )
        label = (
            "Worker Result Path"
            if anchors.get("Worker Result Path")
            else "Result Path"
        )
        declared = anchors.get(label)
        if not declared:
            continue
        declared_path = _normalized_path(
            _project_or_absolute(job.project_root, declared)
        )
        accepted = {
            _normalized_path(path)
            for path in (job.worker_result_path, *job.result_aliases)
        }
        if declared_path not in accepted:
            errors.append(
                f"{job.worker_id}: workerResultPath {job.worker_result_path} "
                f"differs from the prompt's **{label}:** {declared_path}"
            )
        narrative = anchors.get("Result Path")
        if (
            label == "Worker Result Path"
            and narrative
            and _normalized_path(job.result_path)
            != _normalized_path(job.worker_result_path)
            and _normalized_path(_project_or_absolute(job.project_root, narrative))
            != _normalized_path(job.result_path)
        ):
            errors.append(
                f"{job.worker_id}: the prompt's **Result Path:** {narrative} is "
                f"not the result this run assembles from ({job.result_path})"
            )
    if errors:
        raise DispatchError(
            "jobs file result paths differ from the prompt anchors — the worker "
            "writes where the prompt says, so the jobs file must name that path: "
            + "; ".join(errors)
        )


def _worker_job_from_file(
    project_root: Path,
    item: Mapping[str, Any],
    *,
    manifest: Mapping[str, Any],
    active_context: Mapping[str, Any] | None = None,
    backend: str,
    idle_timeout_seconds: int | None,
    dispatch_kind: str,
    resolve_wrapper: Callable[[str], Path],
    default_provider: Callable[[str], str],
) -> WorkerJob:
    # 빠진 필드를 한꺼번에 댄다. 하나씩 거절하던 동안 세 워커짜리 파일 하나를
    # 통과시키는 데 dry-run 세 번이 들었다(2026-09-02 실측).
    missing = _missing_required_strings(item)
    if missing:
        raise DispatchError(
            "jobs file worker is missing required string fields: "
            + ", ".join(missing)
        )
    identity = worker_execution_identity(item)
    worker_id = (
        require_string(item, "workerId")
        if identity is None
        else v2_worker_state_key(item)
    )
    provider = (
        string_value(item.get("provider")) or default_provider(worker_id)
        if identity is None
        else require_string(item, "provider")
    )
    prompt_path = resolve_project_path(
        project_root, require_string(item, "promptPath")
    )
    worker_result_path = resolve_project_path(
        project_root, require_string(item, "workerResultPath")
    )
    # The file's own `resultPath` / `completionPaths` are advisory: the same
    # derivation the roster path applies decides them, so a jobs-file dispatch
    # and a roster dispatch of one worker cannot disagree about which artifact
    # is the result. A file that names the report writer's `.md` as its result
    # used to reach report finalization as the data path.
    result_path = dispatch_result_path(
        worker_id, worker_result_path, manifest, project_root
    )
    completion_paths = dispatch_completion_paths(
        worker_id, worker_result_path, manifest, project_root
    )
    result_aliases = _plan_verify_result_aliases(worker_result_path, manifest)
    digests = item.get("digests")
    digest_values = digests if isinstance(digests, Mapping) else {}
    host_model_value = item.get("hostModelValue")
    if host_model_value is not None and not isinstance(host_model_value, str):
        raise DispatchError("jobs file hostModelValue must be a string or null")
    return WorkerJob(
        worker_id=worker_id,
        provider=provider,
        backend=backend,
        project_root=project_root,
        model_execution_value=require_string(item, "modelExecutionValue"),
        wrapper_path=resolve_wrapper(provider),
        prompt_path=prompt_path,
        result_path=result_path,
        worker_result_path=worker_result_path,
        completion_paths=completion_paths,
        # 파일이 적지 않았으면 run authority 에서 해소한다. 로스터 경로는
        # 언제나 그렇게 하고(`dispatch_core._jobs_from_roster`), 예약 쪽
        # (`agent_prompt_cli._run_worktree`)도 같은 seam 을 읽는다. 이 자리만
        # 손으로 적은 값을 유일한 출처로 삼던 동안, 같은 run 이 어떤 경로로
        # 디스패치됐는지에 따라 워커의 소스 루트가 달라졌다 — 그 값은
        # writePolicy 의 `sourcePolicy.allowedRoot` 이므로, 예약본과 갈리면
        # 같은 invocationRef 가 `invocationRef drift` 로 거부된다.
        worktree_path=(
            string_value(item.get("worktreePath"))
            or worktree_path(manifest, active_context or {})
        ),
        role=require_string(item, "role"),
        idle_timeout_seconds=idle_timeout_seconds,
        dispatch_kind=dispatch_kind,
        # Minted here, not read from the file: the lead writes this file by
        # hand, and a session id it typed would be unfalsifiable — nothing can
        # tell an id that names no session from one that names the wrong one.
        session_id=dispatch_session_id(provider),
        invocation_id=string_value(item.get("invocationId")),
        audience=string_value(item.get("audience")),
        assignment_ref=string_value(item.get("assignmentRef")),
        prompt_metadata_path=resolve_project_path(
            project_root,
            string_value(item.get("promptMetadataPath")),
        ) if string_value(item.get("promptMetadataPath")) else Path(),
        catalog_digest=string_value(digest_values.get("catalogDigest")),
        assignment_digest=string_value(digest_values.get("assignmentDigest")),
        duty_digest=string_value(digest_values.get("dutyDigest")),
        instruction_digest=string_value(digest_values.get("instructionDigest")),
        prompt_digest=string_value(digest_values.get("promptDigest")),
        host_model_value=host_model_value,
        enforcement_mode=string_value(item.get("enforcementMode")),
        result_aliases=result_aliases,
        **(identity or {}),
    )


_V1_JOB_REQUIRED_STRINGS = (
    "workerId", "promptPath", "workerResultPath", "modelExecutionValue", "role",
)
_V2_JOB_REQUIRED_STRINGS = (
    "provider", "assignmentRef", "participantRef", "roleExecutionRef",
    "executionLabel", "dutyId", "invocationRef", "promptPath",
    "workerResultPath", "modelExecutionValue", "role",
)


def _missing_required_strings(item: Mapping[str, Any]) -> list[str]:
    keys = (
        _V2_JOB_REQUIRED_STRINGS
        if item.get("schemaVersion") == "2.0"
        else _V1_JOB_REQUIRED_STRINGS
    )
    return [key for key in keys if not string_value(item.get(key))]


def worker_execution_identity(
    item: Mapping[str, Any],
) -> dict[str, Any] | None:
    identity_keys = {
        "executionIdentityVersion",
        "participantRef",
        "roleExecutionRef",
        "executionLabel",
        "dutyId",
        "invocationRef",
    }
    schema = item.get("schemaVersion")
    has_v2_fields = bool(identity_keys.intersection(item))
    if schema in (None, 1, "1", "1.0"):
        if has_v2_fields:
            raise DispatchError("jobs file worker mixes v1 and v2 identity")
        return None
    if schema != "2.0" or item.get("executionIdentityVersion") != 2:
        raise DispatchError("jobs file worker has unsupported identity version")
    if "workerId" in item:
        raise DispatchError("jobs file worker mixes v1 and v2 identity")
    attempt = item.get("attempt")
    if not isinstance(attempt, int) or isinstance(attempt, bool) or attempt < 1:
        raise DispatchError("jobs file v2 attempt must be positive")
    return {
        "participant_ref": require_string(item, "participantRef"),
        "role_execution_ref": require_string(item, "roleExecutionRef"),
        "execution_label": require_string(item, "executionLabel"),
        "duty_id": require_string(item, "dutyId"),
        "invocation_ref": require_string(item, "invocationRef"),
        "attempt": attempt,
    }


def v2_worker_state_key(item: Mapping[str, Any]) -> str:
    assignment_ref = require_string(item, "assignmentRef")
    worker_id = assignment_ref.rsplit("/", 1)[-1]
    if not worker_id:
        raise DispatchError("jobs file v2 assignmentRef has no worker projection")
    return worker_id


def utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def dispatch_record_matches(record: object, prompt_path: Path, attempt: int) -> bool:
    """Whether this team-state row is the dispatch handed `prompt_path`.

    The prompt file identifies a dispatch: each round writes its own, and a
    second run of the same one is a retry, which `attempt` separates. Worker id
    and dispatch kind cannot carry that — `--dispatch-kind` is the lead's to
    pick and it has reused one across two rounds, which put the finished
    round's completion on the older namesake row and left the row that actually
    ran marked running, so an await re-collected that worker forever.
    """
    return (
        isinstance(record, dict)
        and record.get("promptPath") == str(prompt_path)
        and record.get("attempt") == attempt
    )