"""Compose and verify auditable LLM invocation specifications."""
from __future__ import annotations

import contextlib
from dataclasses import dataclass
from datetime import datetime, timezone
import fcntl
import hashlib
import json
import os
from pathlib import Path
from pathlib import PurePosixPath
import re
import tempfile
from typing import Iterator, Literal, Mapping, get_args

from .domain.role import RoleCatalogError, role_for_duty
from .json_boundary import (
    JsonBoundaryError,
    external_invocation_json_source,
    load_external_json,
    load_owned_object,
)


AgentAudience = Literal[
    "lead",
    "analysis-worker",
    "discovery-worker",
    "diagnosis-worker",
    "planning-worker",
    "direction-selection-worker",
    "implementation-executor",
    "implementation-verifier",
    "acceptance-verifier",
    "reverification-worker",
    "scope-critic",
    "acceptance-critic",
    "report-writer",
    "translator",
    "code-reviewer",
    "schedule-verifier",
]

_SUPPORTED_AUDIENCES = frozenset(get_args(AgentAudience))
# The section set IS the duty contract's shape: a duty author adding a role file
# reads these names, and `_validate_duty_sections` refuses a file that misses one.
# The check is structural — it proves every required section exists and carries
# text, NOT that the text is a real contract. A one-line placeholder passes it;
# what keeps a section substantive is review, and the rule that earns a section a
# place here at all: a sentence that reads the same in another duty file belongs
# in `common.md` or nowhere.
COMMON_DUTY_SECTIONS = (
    "Assignment fidelity",
    "Required inputs",
    "Evidence first",
    "Authority and scope",
    "Collaboration and independence",
    "Instruction precedence",
    "Conflict handling",
    "Completion honesty",
)
ROLE_DUTY_SECTIONS = (
    "Responsibility",
    "Required conduct",
    "Decision principles",
    "Authority and boundaries",
    "Evidence standard",
    "Collaboration contract",
    "Completion criteria",
    "Forbidden conduct",
    "Blocked-state reporting",
)
_SLUG_RE = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
_DUTY_SECTION_RE = re.compile(r"(?m)^## ([^\n]+)\s*$")
_TOP_LEVEL_KEYS = {
    "schemaVersion",
    "invocationId",
    "audience",
    "dispatchKind",
    "workerId",
    "assignmentRef",
    "contractSource",
    "modelAssignment",
    "dutyContract",
    "instruction",
    "prompt",
    "digests",
}
_V2_TOP_LEVEL_KEYS = (_TOP_LEVEL_KEYS - {"workerId"}) | {
    "executionIdentityVersion",
    "invocationRef",
    "participantRef",
    "roleExecutionRef",
    "executionLabel",
    "dutyId",
    "attempt",
}
_NESTED_KEYS = {
    "contractSource": {"mode", "runManifestPath", "dutyRootPath"},
    "modelAssignment": {
        "provider",
        "model",
        "modelExecutionValue",
        "runner",
        "hostRuntime",
        "hostModelValue",
    },
    "dutyContract": {"id", "version"},
    "instruction": {"sourcePaths"},
    "prompt": {"path"},
    "digests": {
        "catalogDigest",
        "assignmentDigest",
        "dutyDigest",
        "instructionDigest",
        "promptDigest",
    },
}


class AgentInvocationError(RuntimeError):
    """Raised when an invocation contract cannot be safely materialized."""

    def __init__(self, message: str, *, reason: str = "invalid_agent_invocation"):
        super().__init__(message)
        self.reason = reason


@dataclass(frozen=True)
class DutyContract:
    id: str
    version: int
    kind: Literal["common", "role"]
    applies_to: AgentAudience | None
    body: str
    source_path: Path


@dataclass(frozen=True)
class AgentModelAssignment:
    provider: str
    model: str
    model_execution_value: str
    runner: str
    host_runtime: str
    host_model_value: str | None


@dataclass(frozen=True)
class InvocationExecutionIdentity:
    participant_ref: str
    role_execution_ref: str
    execution_label: str
    duty_id: str


@dataclass(frozen=True)
class InvocationMetadataIdentity:
    invocation_ref: str
    participant_ref: str
    role_execution_ref: str
    execution_label: str
    duty_id: str
    attempt: int


@dataclass(frozen=True)
class AgentInstructionSource:
    kind: Literal["project", "runtime"]
    path: str


@dataclass(frozen=True)
class AgentInstruction:
    anchor_lines: tuple[str, ...]
    body: str
    source_paths: tuple[AgentInstructionSource, ...]


@dataclass(frozen=True)
class AgentInvocationRequest:
    invocation_id: str
    worker_id: str | None
    audience: AgentAudience
    assignment_ref: str | None
    purpose: str | None
    assignment: AgentModelAssignment
    instruction: AgentInstruction
    project_root: Path
    run_manifest_path: Path | None
    duty_root: Path
    prompt_path: Path
    metadata_path: Path
    dispatch_kind: str
    participant_ref: str | None
    role_execution_ref: str | None
    duty_id: str | None
    invocation_ref: str | None
    attempt: int
    # Republish a prompt this invocation id already wrote, but only while no
    # dispatch has used it. See `_publish_or_reuse`.
    replace_undispatched: bool = False


@dataclass(frozen=True)
class PreparedAgentInvocation:
    invocation_id: str
    worker_id: str | None
    assignment_ref: str | None
    prompt_path: Path
    metadata_path: Path
    assignment: AgentModelAssignment
    duty_id: str
    duty_version: int
    catalog_digest: str
    assignment_digest: str
    duty_digest: str
    instruction_digest: str
    prompt_digest: str
    participant_ref: str | None = None
    role_execution_ref: str | None = None
    invocation_ref: str | None = None
    attempt: int = 1


@dataclass(frozen=True)
class MaterializedStandaloneResult:
    purpose: str
    invocation_id: str
    prompt_metadata_path: Path
    result_path: Path
    result_envelope_digest: str


@dataclass(frozen=True)
class RetryAgentInvocation:
    invocation_id: str
    invocation_ref: str
    prompt_path: Path
    metadata_path: Path
    attempt: int


@dataclass(frozen=True)
class VerifiedStandaloneResult:
    purpose: str
    invocation_id: str
    prompt_metadata_path: Path
    result_path: Path
    completion_path: Path
    result_envelope_digest: str
    returned_body: str


def agent_model_assignment_from_payload(payload: object) -> AgentModelAssignment:
    """Parse the exact persisted six-field model assignment contract."""
    if not isinstance(payload, Mapping) or set(payload) != _NESTED_KEYS["modelAssignment"]:
        raise AgentInvocationError("model assignment payload is invalid")
    string_fields = (
        "provider",
        "model",
        "modelExecutionValue",
        "runner",
        "hostRuntime",
    )
    if any(not isinstance(payload[key], str) for key in string_fields):
        raise AgentInvocationError("model assignment payload is invalid")
    host_model_value = payload["hostModelValue"]
    if host_model_value is not None and not isinstance(host_model_value, str):
        raise AgentInvocationError("model assignment payload is invalid")
    assignment = AgentModelAssignment(
        provider=payload["provider"],
        model=payload["model"],
        model_execution_value=payload["modelExecutionValue"],
        runner=payload["runner"],
        host_runtime=payload["hostRuntime"],
        host_model_value=host_model_value,
    )
    _validate_assignment(assignment)
    return assignment


def invocation_execution_identity_from_manifest(
    manifest: Mapping[str, object],
    *,
    assignment: AgentModelAssignment,
    assignment_ref: str,
    duty_id: str,
    role_execution_ref: str | None = None,
) -> InvocationExecutionIdentity | None:
    """Resolve v2 refs from Task 8 without duplicating them in model assignments."""
    schema = manifest.get("schemaVersion")
    identity_version = manifest.get("executionIdentityVersion")
    if schema in (None, 1, "1", "1.0") and identity_version is None:
        return None
    if schema != "2.0" or identity_version != 2:
        raise AgentInvocationError("run manifest mixes v1 and v2 execution identity")
    try:
        role = role_for_duty(duty_id)
    except RoleCatalogError as exc:
        raise AgentInvocationError(str(exc)) from exc
    role_executions = manifest.get("roleExecutions")
    candidates = [
        row for row in role_executions
        if isinstance(row, Mapping)
        and row.get("role") == role
        and row.get("provider") == assignment.provider
        and _role_execution_matches_assignment(row, assignment)
    ] if isinstance(role_executions, list) else []
    if role_execution_ref is None:
        selected = _select_role_execution_candidate(
            manifest,
            candidates,
            assignment_ref=assignment_ref,
            role=role,
            assignment=assignment,
        )
    else:
        selected = next(
            (
                row
                for row in candidates
                if row.get("roleExecutionRef") == role_execution_ref
            ),
            None,
        )
    if selected is None:
        raise AgentInvocationError(
            "v2 execution identity does not match a canonical role execution"
        )
    return InvocationExecutionIdentity(
        participant_ref=_required_metadata_string(selected, "participantRef"),
        role_execution_ref=_required_metadata_string(selected, "roleExecutionRef"),
        execution_label=_required_metadata_string(selected, "executionLabel"),
        duty_id=duty_id,
    )


def v2_role_assignment_authority_errors(
    manifest: Mapping[str, object],
    *,
    role_execution_ref: str,
    participant_ref: str,
    assignment: AgentModelAssignment,
) -> list[str]:
    """Compare one v2 role and participant with the provider-neutral assignment."""
    role_executions = manifest.get("roleExecutions")
    execution = next((
        row for row in role_executions
        if isinstance(row, Mapping)
        and row.get("roleExecutionRef") == role_execution_ref
    ), None) if isinstance(role_executions, list) else None
    participants = manifest.get("participantAssignments")
    participant = next((
        row for row in participants
        if isinstance(row, Mapping)
        and row.get("participantRef") == participant_ref
    ), None) if isinstance(participants, list) else None
    if not isinstance(execution, Mapping) or not isinstance(participant, Mapping):
        return ["v2 assignment does not reference canonical execution identity"]
    errors: list[str] = []
    if participant.get("hostModelValue") != assignment.host_model_value:
        errors.append(
            "participant hostModelValue does not match invocation metadata"
        )
    if (
        participant.get("hostRuntime") != assignment.host_runtime
        or (
            manifest.get("leadRuntime") is not None
            and participant.get("hostRuntime") != manifest.get("leadRuntime")
        )
    ):
        errors.append(
            "participant hostRuntime does not match invocation metadata"
        )
    binding = execution.get("binding")
    binding_matches = (
        isinstance(binding, Mapping)
        and binding.get("runner") == assignment.runner
        and binding.get("resolvedExecutionValue")
        == assignment.model_execution_value
        and binding.get("hostModelValue") == assignment.host_model_value
        and binding.get("hostModelValue") == participant.get("hostModelValue")
        and binding.get("workerWriteCapability")
        == participant.get("workerWriteCapability")
        and participant.get("runner") == binding.get("runner")
    ) or (
        binding is None
        and assignment.model_execution_value == "unknown"
        and participant.get("runner") == assignment.runner
    )
    if (
        execution.get("participantRef") != participant_ref
        or execution.get("provider") != assignment.provider
        or execution.get("provider") != participant.get("provider")
        or execution.get("modelRef") != participant.get("modelRef")
        or execution.get("modelId") != assignment.model
        or execution.get("modelId") != participant.get("modelId")
        or not binding_matches
    ):
        errors.append("v2 assignment does not match role execution authority")
    return errors


def _role_execution_matches_assignment(
    row: Mapping[str, object], assignment: AgentModelAssignment,
) -> bool:
    binding = row.get("binding")
    if isinstance(binding, Mapping):
        return binding.get("resolvedExecutionValue") == assignment.model_execution_value
    return assignment.model_execution_value == "unknown"


def _select_role_execution_candidate(
    manifest: Mapping[str, object],
    candidates: list[Mapping[str, object]],
    *,
    assignment_ref: str,
    role: str,
    assignment: AgentModelAssignment,
) -> Mapping[str, object] | None:
    if len(candidates) == 1:
        return candidates[0]
    if not assignment_ref.startswith("initial/"):
        return None
    worker_id = assignment_ref.partition("/")[2]
    workers = manifest.get("workerAssignments")
    matching_workers = [
        row for row in workers
        if isinstance(row, Mapping)
        and row.get("role") == role
        and row.get("provider") == assignment.provider
        and row.get("modelExecutionValue") == assignment.model_execution_value
    ] if isinstance(workers, list) else []
    worker_index = next((
        index for index, row in enumerate(matching_workers)
        if row.get("workerId") == worker_id
    ), None)
    ordered = sorted(candidates, key=lambda row: int(row.get("ordinal", 0)))
    if worker_index is None or worker_index >= len(ordered):
        return None
    return ordered[worker_index]


def _required_metadata_string(
    payload: Mapping[str, object], key: str,
) -> str:
    value = payload.get(key)
    if not isinstance(value, str) or not value:
        raise AgentInvocationError(f"v2 execution identity field is invalid: {key}")
    return value


def invocation_metadata_identity(
    payload: Mapping[str, object],
) -> InvocationMetadataIdentity | None:
    """Read one exact metadata version without guessing from partial fields."""
    schema = payload.get("schemaVersion")
    identity_version = payload.get("executionIdentityVersion")
    if schema == 1 and identity_version is None:
        if not _metadata_schema_is_exact(payload):
            raise AgentInvocationError("agent invocation metadata schema is invalid")
        return None
    if schema != "2.0" or identity_version != 2:
        raise AgentInvocationError("agent invocation metadata mixes v1 and v2 identity")
    if not _metadata_schema_is_exact(payload):
        raise AgentInvocationError("v2 execution identity metadata is invalid")
    return InvocationMetadataIdentity(
        invocation_ref=payload["invocationRef"],
        participant_ref=payload["participantRef"],
        role_execution_ref=payload["roleExecutionRef"],
        execution_label=payload["executionLabel"],
        duty_id=payload["dutyId"],
        attempt=payload["attempt"],
    )


@dataclass(frozen=True)
class _MaterializedInvocation:
    prepared: PreparedAgentInvocation
    prompt_bytes: bytes
    metadata: dict[str, object]
    metadata_bytes: bytes
    reservation_root: Path | None
    reservation: dict[str, object] | None


def load_common_duty_contract(duty_root: Path) -> DutyContract:
    """Load the common fragment, which is deliberately outside the role catalog."""
    path = duty_root / "common.md"
    fields, body = _parse_duty_file(path)
    if set(fields) != {"id", "version", "kind"}:
        raise AgentInvocationError(f"invalid common duty frontmatter: {path}")
    if fields["id"] != "common" or fields["kind"] != "common":
        raise AgentInvocationError(f"invalid common duty frontmatter: {path}")
    _validate_duty_sections(body, COMMON_DUTY_SECTIONS, "common", path)
    return DutyContract(
        id="common",
        version=_parse_version(fields["version"], path),
        kind="common",
        applies_to=None,
        body=body,
        source_path=path,
    )


def load_duty_catalog(duty_root: Path) -> dict[AgentAudience, DutyContract]:
    """Load one role contract for every supported invocation audience."""
    catalog: dict[AgentAudience, DutyContract] = {}
    seen_ids: set[str] = set()
    for path in sorted(duty_root.glob("*.md")):
        if path.name == "common.md":
            continue
        duty = _load_role_duty(path)
        if duty.id in seen_ids:
            raise AgentInvocationError(f"duplicate duty id: {duty.id}")
        if duty.applies_to in catalog:
            raise AgentInvocationError(f"duplicate duty audience: {duty.applies_to}")
        seen_ids.add(duty.id)
        catalog[duty.applies_to] = duty
    missing = sorted(_SUPPORTED_AUDIENCES - set(catalog))
    if missing:
        raise AgentInvocationError(f"missing duty audiences: {', '.join(missing)}")
    return catalog


def digest_duty_catalog(duty_root: Path) -> str:
    """Return the canonical digest of every duty file in a snapshot."""
    names = [path.relative_to(duty_root).as_posix() for path in duty_root.glob("*.md")]
    return _digest_framed_files(duty_root, names)


def prepare_agent_invocation(
    request: AgentInvocationRequest,
) -> PreparedAgentInvocation:
    """Materialize one immutable prompt and its adjacent completion metadata."""
    _validate_request(request)
    materialized = _materialize(request)
    if materialized.reservation_root is None:
        with _exclusive_lock(_prompt_lock_path(request.prompt_path)):
            _publish_or_reuse(request, materialized)
    else:
        reservation_lock = materialized.reservation_root / "publish.lock"
        with _exclusive_lock(reservation_lock):
            _publish_or_reuse_reservation(materialized)
            with _exclusive_lock(_prompt_lock_path(request.prompt_path)):
                _publish_or_reuse(request, materialized)
    errors = verify_agent_invocation(
        request.metadata_path,
        project_root=request.project_root,
        expected_run_manifest_path=request.run_manifest_path,
        expected_assignment=request.assignment,
        expected_invocation_id=request.invocation_id,
        expected_worker_id=request.worker_id,
        expected_assignment_ref=request.assignment_ref,
        expected_audience=request.audience,
        expected_participant_ref=request.participant_ref,
        expected_role_execution_ref=request.role_execution_ref,
        expected_invocation_ref=request.invocation_ref,
    )
    if errors:
        raise AgentInvocationError("; ".join(errors))
    return materialized.prepared


def materialize_retry_invocation(
    metadata_path: Path,
    *,
    project_root: Path,
    attempt: int,
) -> RetryAgentInvocation:
    """Publish an attempt-specific prompt/metadata pair for one v2 retry."""
    metadata = _load_json_object(metadata_path, "agent invocation metadata")
    identity = invocation_metadata_identity(metadata)
    if identity is None:
        raise AgentInvocationError("v1 invocation metadata cannot be retried as v2")
    if attempt != identity.attempt + 1:
        raise AgentInvocationError(
            f"retry attempt must be {identity.attempt + 1}"
        )
    source = metadata.get("contractSource")
    prompt_spec = metadata.get("prompt")
    if not isinstance(source, dict) or not isinstance(prompt_spec, dict):
        raise AgentInvocationError("retry invocation metadata is incomplete")
    manifest_path = _project_path(
        project_root, source.get("runManifestPath"), must_exist=True
    )
    prompt_path = _project_path(
        project_root, prompt_spec.get("path"), must_exist=True
    )
    retry_prompt_path = prompt_path.with_name(
        f"{prompt_path.stem}-attempt-{attempt}{prompt_path.suffix}"
    )
    retry_metadata_path = retry_prompt_path.with_name(
        retry_prompt_path.name + ".meta.json"
    )
    retry_invocation_id = f"{metadata['invocationId']}-attempt-{attempt}"
    retry_metadata = dict(metadata)
    retry_metadata["invocationId"] = retry_invocation_id
    retry_metadata["attempt"] = attempt
    retry_metadata["prompt"] = {
        "path": _project_relative(
            retry_prompt_path, project_root, must_exist=False
        )
    }
    manifest = _load_json_object(manifest_path, "run manifest")
    contract = manifest.get("agentContract")
    if not isinstance(contract, dict):
        raise AgentInvocationError("run manifest has no agent contract")
    reservation_root = _project_path(
        project_root,
        contract.get("invocationReservationRootPath"),
        must_exist=True,
    )
    reservation = {
        "schemaVersion": "2.0",
        "executionIdentityVersion": 2,
        "invocationId": retry_invocation_id,
        "assignmentRef": metadata["assignmentRef"],
        "audience": metadata["audience"],
        "dispatchKind": metadata["dispatchKind"],
        "promptPath": retry_metadata["prompt"]["path"],
        "metadataPath": _project_relative(
            retry_metadata_path, project_root, must_exist=False
        ),
        "invocationRef": identity.invocation_ref,
        "participantRef": identity.participant_ref,
        "roleExecutionRef": identity.role_execution_ref,
        "dutyId": identity.duty_id,
        "attempt": attempt,
    }
    with _exclusive_lock(reservation_root / "publish.lock"):
        _publish_or_reuse_exact(
            reservation_root / f"{retry_invocation_id}.json",
            _pretty_json(reservation),
        )
        with _exclusive_lock(_prompt_lock_path(retry_prompt_path)):
            _publish_or_reuse_exact(retry_prompt_path, prompt_path.read_bytes())
            _publish_or_reuse_exact(
                retry_metadata_path, _pretty_json(retry_metadata)
            )
    errors = verify_agent_invocation(
        retry_metadata_path,
        project_root=project_root,
        expected_run_manifest_path=manifest_path,
        expected_assignment=agent_model_assignment_from_payload(
            metadata["modelAssignment"]
        ),
        expected_invocation_id=retry_invocation_id,
        expected_assignment_ref=str(metadata["assignmentRef"]),
        expected_audience=str(metadata["audience"]),
        expected_participant_ref=identity.participant_ref,
        expected_role_execution_ref=identity.role_execution_ref,
        expected_invocation_ref=identity.invocation_ref,
    )
    if errors:
        raise AgentInvocationError("; ".join(errors))
    return RetryAgentInvocation(
        invocation_id=retry_invocation_id,
        invocation_ref=identity.invocation_ref,
        prompt_path=retry_prompt_path,
        metadata_path=retry_metadata_path,
        attempt=attempt,
    )


def compose_agent_prompt(request: AgentInvocationRequest) -> str:
    """Render and validate a prompt candidate without publishing artifacts."""
    _validate_request(request)
    return _materialize(request).prompt_bytes.decode("utf-8")


def compose_unbound_run_prompt(request: AgentInvocationRequest) -> bytes:
    """Compose v2 run prompt bytes before a dynamic role identity is reserved."""
    _validate_common_request(request)
    _validate_unbound_v2_run_request(request)
    common = load_common_duty_contract(request.duty_root)
    catalog = load_duty_catalog(request.duty_root)
    duty = catalog[request.audience]
    return _render_prompt(request, common, duty).encode("utf-8")


def verify_agent_invocation(
    metadata_path: Path,
    *,
    project_root: Path,
    expected_run_manifest_path: Path | None,
    expected_assignment: AgentModelAssignment | None = None,
    expected_invocation_id: str | None = None,
    expected_worker_id: str | None = None,
    expected_assignment_ref: str | None = None,
    expected_audience: AgentAudience | None = None,
    expected_participant_ref: str | None = None,
    expected_role_execution_ref: str | None = None,
    expected_invocation_ref: str | None = None,
) -> list[str]:
    """Return every deterministic violation found in a published invocation."""
    metadata = _read_metadata(metadata_path)
    if metadata is None or not _metadata_schema_is_exact(metadata):
        return ["agent invocation metadata schema is invalid"]
    errors = _verify_expected_identity(
        metadata,
        expected_invocation_id=expected_invocation_id,
        expected_worker_id=expected_worker_id,
        expected_assignment_ref=expected_assignment_ref,
        expected_audience=expected_audience,
        expected_participant_ref=expected_participant_ref,
        expected_role_execution_ref=expected_role_execution_ref,
        expected_invocation_ref=expected_invocation_ref,
    )
    errors.extend(_verify_prompt_file(metadata, project_root))
    errors.extend(_verify_assignment(metadata, expected_assignment))
    errors.extend(
        _verify_contract_source(
            metadata,
            metadata_path=metadata_path,
            project_root=project_root,
            expected_run_manifest_path=expected_run_manifest_path,
        )
    )
    return _deduplicate(errors)


def materialize_standalone_result(
    *,
    project_root: Path,
    purpose: str,
    metadata_path: Path,
    returned_body: bytes,
) -> MaterializedStandaloneResult:
    """Publish one immutable raw-return envelope for a standalone invocation."""
    try:
        returned_text = returned_body.decode("utf-8")
    except UnicodeDecodeError as exc:
        raise AgentInvocationError("returned body must be valid UTF-8") from exc
    invocation_id, root = _standalone_lock_identity(
        project_root=project_root, purpose=purpose, metadata_path=metadata_path,
    )
    with _exclusive_lock(root / f"{invocation_id}.result.publish.lock"):
        invocation_id, root, canonical_metadata = _standalone_authority(
            project_root=project_root,
            purpose=purpose,
            metadata_path=metadata_path,
        )
        result_path = root / f"{invocation_id}.result.json"
        envelope = {
            "schemaVersion": 1,
            "purpose": purpose,
            "invocationId": invocation_id,
            "promptMetadataPath": _project_relative(
                canonical_metadata,
                project_root,
                must_exist=True,
            ),
            "returnedBody": returned_text,
        }
        envelope_bytes = _canonical_json(envelope)
        _publish_exclusive(result_path, envelope_bytes)
    return MaterializedStandaloneResult(
        purpose=purpose,
        invocation_id=invocation_id,
        prompt_metadata_path=canonical_metadata,
        result_path=result_path,
        result_envelope_digest=_sha256(envelope_bytes),
    )


def publish_standalone_completion(
    *,
    project_root: Path,
    purpose: str,
    metadata_path: Path,
    completed_at: datetime,
) -> Path:
    """Publish the completion marker last, after re-verifying prompt and result."""
    completed = _utc_seconds(completed_at)
    invocation_id, root = _standalone_lock_identity(
        project_root=project_root, purpose=purpose, metadata_path=metadata_path,
    )
    with _exclusive_lock(root / f"{invocation_id}.result.publish.lock"):
        invocation_id, root, canonical_metadata = _standalone_authority(
            project_root=project_root,
            purpose=purpose,
            metadata_path=metadata_path,
        )
        result_path = root / f"{invocation_id}.result.json"
        completion_path = root / f"{invocation_id}.prompt.md.completion.json"
        envelope_bytes, _returned_body = _verified_result_envelope(
            result_path,
            project_root=project_root,
            purpose=purpose,
            invocation_id=invocation_id,
            metadata_path=canonical_metadata,
        )
        completion = {
            "schemaVersion": 1,
            "purpose": purpose,
            "invocationId": invocation_id,
            "status": "completed",
            "promptMetadataPath": _project_relative(
                canonical_metadata,
                project_root,
                must_exist=True,
            ),
            "resultPath": _project_relative(
                result_path,
                project_root,
                must_exist=True,
            ),
            "resultEnvelopeDigest": _sha256(envelope_bytes),
            "completedAt": completed,
        }
        _publish_exclusive(completion_path, _canonical_json(completion))
    return completion_path


def verify_standalone_completion(
    completion_path: Path,
    *,
    project_root: Path,
    expected_purpose: str,
) -> VerifiedStandaloneResult:
    """Verify a completion marker and return its already-read immutable body."""
    _validate_slug(expected_purpose, "standalone purpose")
    completion = _load_json_object(completion_path, "standalone completion")
    expected_keys = {
        "schemaVersion",
        "purpose",
        "invocationId",
        "status",
        "promptMetadataPath",
        "resultPath",
        "resultEnvelopeDigest",
        "completedAt",
    }
    if set(completion) != expected_keys or completion.get("schemaVersion") != 1:
        raise AgentInvocationError("standalone completion schema is invalid")
    if completion.get("purpose") != expected_purpose:
        raise AgentInvocationError("standalone completion purpose does not match")
    invocation_id = completion.get("invocationId")
    if not isinstance(invocation_id, str):
        raise AgentInvocationError("standalone completion invocation ID is invalid")
    _validate_slug(invocation_id, "standalone invocation ID")
    root = project_root / ".okstra" / "agent-invocations" / expected_purpose
    canonical_completion = root / f"{invocation_id}.prompt.md.completion.json"
    if completion_path.resolve(strict=True) != canonical_completion.resolve(strict=True):
        raise AgentInvocationError("standalone completion path is not canonical")
    metadata_path = root / f"{invocation_id}.prompt.md.meta.json"
    result_path = root / f"{invocation_id}.result.json"
    if completion.get("status") != "completed":
        raise AgentInvocationError("standalone completion status is invalid")
    if completion.get("promptMetadataPath") != _project_relative(
        metadata_path,
        project_root,
        must_exist=True,
    ):
        raise AgentInvocationError("standalone completion metadata path does not match")
    if completion.get("resultPath") != _project_relative(
        result_path,
        project_root,
        must_exist=True,
    ):
        raise AgentInvocationError("standalone completion result path does not match")
    with _exclusive_lock(root / f"{invocation_id}.result.publish.lock"):
        authority_id, _authority_root, canonical_metadata = _standalone_authority(
            project_root=project_root,
            purpose=expected_purpose,
            metadata_path=metadata_path,
        )
        if authority_id != invocation_id:
            raise AgentInvocationError("standalone completion identity does not match")
        envelope_bytes, returned_body = _verified_result_envelope(
            result_path,
            project_root=project_root,
            purpose=expected_purpose,
            invocation_id=invocation_id,
            metadata_path=canonical_metadata,
        )
        digest = _sha256(envelope_bytes)
        if completion.get("resultEnvelopeDigest") != digest:
            raise AgentInvocationError("result envelope digest does not match completion")
    return VerifiedStandaloneResult(
        purpose=expected_purpose,
        invocation_id=invocation_id,
        prompt_metadata_path=metadata_path,
        result_path=result_path,
        completion_path=canonical_completion,
        result_envelope_digest=digest,
        returned_body=returned_body,
    )


def _standalone_authority(
    *, project_root: Path, purpose: str, metadata_path: Path,
) -> tuple[str, Path, Path]:
    _validate_slug(purpose, "standalone purpose")
    root = project_root / ".okstra" / "agent-invocations" / purpose
    metadata = _load_json_object(metadata_path, "agent invocation metadata")
    source = metadata.get("contractSource")
    if not isinstance(source, Mapping) or source.get("mode") != "standalone":
        raise AgentInvocationError("result completion requires standalone metadata")
    invocation_id = metadata.get("invocationId")
    if not isinstance(invocation_id, str):
        raise AgentInvocationError("standalone invocation ID is invalid")
    _validate_slug(invocation_id, "standalone invocation ID")
    canonical_metadata = root / f"{invocation_id}.prompt.md.meta.json"
    if metadata_path.resolve(strict=True) != canonical_metadata.resolve(strict=True):
        raise AgentInvocationError("standalone metadata path is not canonical")
    errors = verify_agent_invocation(
        canonical_metadata,
        project_root=project_root,
        expected_run_manifest_path=None,
        expected_invocation_id=invocation_id,
    )
    if errors:
        raise AgentInvocationError("; ".join(errors))
    return invocation_id, root, canonical_metadata


def _standalone_lock_identity(
    *, project_root: Path, purpose: str, metadata_path: Path,
) -> tuple[str, Path]:
    """Derive only the stable lock name; verify all authority after locking."""
    _validate_slug(purpose, "standalone purpose")
    suffix = ".prompt.md.meta.json"
    if not metadata_path.name.endswith(suffix):
        raise AgentInvocationError("standalone metadata path is not canonical")
    invocation_id = metadata_path.name.removesuffix(suffix)
    _validate_slug(invocation_id, "standalone invocation ID")
    return invocation_id, project_root / ".okstra" / "agent-invocations" / purpose


def _verified_result_envelope(
    result_path: Path,
    *,
    project_root: Path,
    purpose: str,
    invocation_id: str,
    metadata_path: Path,
) -> tuple[bytes, str]:
    try:
        body = result_path.read_bytes()
        envelope = load_external_json(
            external_invocation_json_source(
                result_path,
                project_root=project_root,
                purpose=purpose,
            ),
            artifact="standalone worker result",
        )
    except (OSError, UnicodeDecodeError, json.JSONDecodeError, JsonBoundaryError) as exc:
        raise AgentInvocationError(
            f"standalone result envelope is missing or invalid: {result_path}"
        ) from exc
    expected_keys = {
        "schemaVersion", "purpose", "invocationId", "promptMetadataPath", "returnedBody",
    }
    expected_metadata = _project_relative(
        metadata_path,
        project_root,
        must_exist=True,
    )
    if (
        not isinstance(envelope, dict)
        or set(envelope) != expected_keys
        or envelope.get("schemaVersion") != 1
        or envelope.get("purpose") != purpose
        or envelope.get("invocationId") != invocation_id
        or envelope.get("promptMetadataPath") != expected_metadata
        or not isinstance(envelope.get("returnedBody"), str)
    ):
        raise AgentInvocationError("standalone result envelope contract does not match")
    if body != _canonical_json(envelope):
        raise AgentInvocationError("standalone result envelope is not canonical JSON")
    return body, envelope["returnedBody"]


def _validate_slug(value: str, label: str) -> None:
    if not _SLUG_RE.fullmatch(value):
        raise AgentInvocationError(f"{label} must be a slug")


def _utc_seconds(value: datetime) -> str:
    if value.tzinfo is None or value.utcoffset() != timezone.utc.utcoffset(value):
        raise AgentInvocationError("completedAt must be a UTC datetime")
    return value.astimezone(timezone.utc).replace(microsecond=0).strftime(
        "%Y-%m-%dT%H:%M:%SZ"
    )


def _materialize(request: AgentInvocationRequest) -> _MaterializedInvocation:
    common = load_common_duty_contract(request.duty_root)
    catalog = load_duty_catalog(request.duty_root)
    duty = catalog[request.audience]
    if _request_identity_version(request) == 2 and request.duty_id != duty.id:
        raise AgentInvocationError("v2 execution identity duty does not match audience")
    prompt_bytes = _render_prompt(request, common, duty).encode("utf-8")
    digests = _invocation_digests(request, duty, prompt_bytes)
    metadata = _metadata_payload(request, duty, digests)
    prepared = PreparedAgentInvocation(
        invocation_id=request.invocation_id,
        worker_id=request.worker_id,
        assignment_ref=request.assignment_ref,
        prompt_path=request.prompt_path,
        metadata_path=request.metadata_path,
        assignment=request.assignment,
        duty_id=duty.id,
        duty_version=duty.version,
        catalog_digest=digests["catalogDigest"],
        assignment_digest=digests["assignmentDigest"],
        duty_digest=digests["dutyDigest"],
        instruction_digest=digests["instructionDigest"],
        prompt_digest=digests["promptDigest"],
        participant_ref=request.participant_ref,
        role_execution_ref=request.role_execution_ref,
        invocation_ref=request.invocation_ref,
        attempt=request.attempt,
    )
    reservation_root, reservation = _reservation_spec(request)
    return _MaterializedInvocation(
        prepared=prepared,
        prompt_bytes=prompt_bytes,
        metadata=metadata,
        metadata_bytes=_pretty_json(metadata),
        reservation_root=reservation_root,
        reservation=reservation,
    )


def _validate_request(request: AgentInvocationRequest) -> None:
    _validate_common_request(request)
    if request.run_manifest_path is None:
        _validate_standalone_identity(request)
    else:
        _validate_run_identity(request)


def _validate_common_request(request: AgentInvocationRequest) -> None:
    if request.audience not in _SUPPORTED_AUDIENCES:
        raise AgentInvocationError(f"unknown duty audience: {request.audience}")
    adjacent = request.prompt_path.with_name(request.prompt_path.name + ".meta.json")
    if request.metadata_path != adjacent:
        raise AgentInvocationError("metadata path must be adjacent to prompt path")
    _validate_assignment(request.assignment)
    _validate_instruction(request.instruction)
    _project_relative(request.duty_root, request.project_root, must_exist=True)
    _project_relative(request.prompt_path, request.project_root, must_exist=False)
    _project_relative(request.metadata_path, request.project_root, must_exist=False)


def _validate_unbound_v2_run_request(request: AgentInvocationRequest) -> None:
    if request.run_manifest_path is None or request.purpose is not None:
        raise AgentInvocationError("unbound prompt requires a run invocation")
    if _request_identity_version(request) != 2:
        raise AgentInvocationError("unbound prompt requires v2 execution identity")
    if not request.assignment_ref:
        raise AgentInvocationError("run assignment reference is required")
    if any((
        request.participant_ref,
        request.role_execution_ref,
        request.duty_id,
        request.invocation_ref,
    )):
        raise AgentInvocationError("unbound prompt cannot carry execution identity")
    if not request.worker_id or not _SLUG_RE.fullmatch(request.worker_id):
        raise AgentInvocationError("run worker ID must be a non-empty slug")
    suffix = request.assignment_ref.partition("/")[2]
    if suffix and request.assignment_ref.split("/", 1)[0] in {"initial", "reverify"}:
        if suffix != request.worker_id:
            raise AgentInvocationError(
                "run worker ID does not match assignment reference"
            )
    manifest = _load_json_object(request.run_manifest_path, "run manifest")
    contract = manifest.get("agentContract")
    if not isinstance(contract, Mapping):
        raise AgentInvocationError("run manifest has no agent contract")
    _validate_manifest_contract(request, manifest, contract)


def _validate_assignment(assignment: AgentModelAssignment) -> None:
    values = (
        assignment.provider,
        assignment.model,
        assignment.model_execution_value,
        assignment.runner,
        assignment.host_runtime,
    )
    if any(not value.strip() for value in values):
        raise AgentInvocationError("model assignment fields must be non-empty")
    if assignment.runner == "native-session" and not assignment.host_model_value:
        raise AgentInvocationError("native assignment requires a host model value")
    if assignment.runner == "cli-wrapper" and assignment.host_model_value is not None:
        raise AgentInvocationError("CLI assignment must not define a host model value")


def _validate_instruction(instruction: AgentInstruction) -> None:
    if re.search(r"(?m)^## (?:Duty Contract|Task Instructions)\s*$", instruction.body):
        raise AgentInvocationError("task instruction contains a reserved duty heading")
    if any("\n" in line for line in instruction.anchor_lines):
        raise AgentInvocationError("instruction anchorLines must be single lines")
    normalized = [_source_payload(source) for source in instruction.source_paths]
    if not normalized or len({_canonical_json(item) for item in normalized}) != len(normalized):
        raise AgentInvocationError("instruction sourcePaths must be non-empty and unique")


def _validate_standalone_identity(request: AgentInvocationRequest) -> None:
    if not request.purpose or not _SLUG_RE.fullmatch(request.purpose):
        raise AgentInvocationError("standalone purpose must be a slug")
    if not _SLUG_RE.fullmatch(request.invocation_id):
        raise AgentInvocationError("standalone invocation ID must be a slug")
    if request.worker_id is not None or request.assignment_ref is not None:
        raise AgentInvocationError("standalone invocation cannot bind a run worker")
    if any((
        request.participant_ref,
        request.role_execution_ref,
        request.duty_id,
        request.invocation_ref,
    )):
        raise AgentInvocationError("standalone invocation cannot bind v2 execution identity")
    root = (
        request.project_root
        / ".okstra"
        / "agent-invocations"
        / request.purpose
    )
    expected_prompt = root / f"{request.invocation_id}.prompt.md"
    expected_duty = root / f"{request.invocation_id}.duty-contracts"
    if request.prompt_path != expected_prompt or request.duty_root != expected_duty:
        raise AgentInvocationError("prompt does not use canonical standalone path")


def _validate_run_identity(request: AgentInvocationRequest) -> None:
    if request.purpose is not None:
        raise AgentInvocationError("run invocation cannot define standalone purpose")
    if not request.assignment_ref:
        raise AgentInvocationError("run assignment reference is required")
    _project_relative(
        request.run_manifest_path,
        request.project_root,
        must_exist=True,
    )
    if _request_identity_version(request) == 2:
        _validate_v2_run_identity(request)
        return
    if any((
        request.participant_ref,
        request.role_execution_ref,
        request.duty_id,
        request.invocation_ref,
    )):
        raise AgentInvocationError("v1 run invocation cannot bind v2 execution identity")
    if not request.worker_id or not _SLUG_RE.fullmatch(request.worker_id):
        raise AgentInvocationError("run worker ID must be a non-empty slug")
    suffix = request.assignment_ref.partition("/")[2]
    if suffix and request.assignment_ref.split("/", 1)[0] in {"initial", "reverify"}:
        if suffix != request.worker_id:
            raise AgentInvocationError("run worker ID does not match assignment reference")


def _request_identity_version(request: AgentInvocationRequest) -> int:
    if request.run_manifest_path is None:
        return 1
    manifest = _load_json_object(request.run_manifest_path, "run manifest")
    schema = manifest.get("schemaVersion")
    identity = manifest.get("executionIdentityVersion")
    if schema == "2.0" and identity == 2:
        return 2
    if schema == "2.0" or identity is not None:
        raise AgentInvocationError("run manifest mixes v1 and v2 execution identity")
    if schema not in (None, 1, "1", "1.0"):
        raise AgentInvocationError("run manifest execution identity version is unsupported")
    return 1


def _validate_v2_run_identity(request: AgentInvocationRequest) -> None:
    fields = (
        request.participant_ref,
        request.role_execution_ref,
        request.duty_id,
        request.invocation_ref,
    )
    if any(not isinstance(value, str) or not value.strip() for value in fields):
        raise AgentInvocationError("v2 execution identity references are required")
    if request.invocation_ref != request.invocation_id:
        raise AgentInvocationError("v2 execution identity invocation reference is invalid")
    if (
        not isinstance(request.attempt, int)
        or isinstance(request.attempt, bool)
        or request.attempt < 1
    ):
        raise AgentInvocationError("v2 execution identity attempt must be positive")
    manifest = _load_json_object(request.run_manifest_path, "run manifest")
    identity = invocation_execution_identity_from_manifest(
        manifest,
        assignment=request.assignment,
        assignment_ref=request.assignment_ref,
        duty_id=request.duty_id,
        role_execution_ref=request.role_execution_ref,
    )
    if (
        identity is None
        or identity.participant_ref != request.participant_ref
        or identity.role_execution_ref != request.role_execution_ref
    ):
        raise AgentInvocationError(
            "v2 execution identity does not match a canonical role execution"
        )


def _render_prompt(
    request: AgentInvocationRequest,
    common: DutyContract,
    duty: DutyContract,
) -> str:
    assignment = request.assignment
    header = [
        *request.instruction.anchor_lines,
        f"**Provider:** {assignment.provider}",
        f"**Model:** {assignment.model}",
        f"**Model execution value:** {assignment.model_execution_value}",
        f"**Runner:** {assignment.runner}",
        f"**Host runtime:** {assignment.host_runtime}",
    ]
    # The dispatch prompt contract requires exactly one of these on every prompt
    # it checks, and a critic pass is checked under the same contract as the
    # initial analysis it audits — but only the initial-prompt path emitted one,
    # so a dynamically materialized critic was refused. A dynamic prompt carries
    # its instruction inline, which is what `eager-include` states; the initial
    # path states the host's `initialPromptDeliveryMode` instead and puts it in
    # `anchor_lines`, so defer to it when it is already there.
    delivery_mode = "**Prompt Delivery Mode:**"
    already_declared = any(
        line.strip().startswith(delivery_mode)
        for line in (
            *request.instruction.anchor_lines,
            *request.instruction.body.splitlines(),
        )
    )
    if not already_declared:
        header.append(f"{delivery_mode} eager-include")
    if assignment.host_model_value is not None:
        header.append(f"**Host model value:** {assignment.host_model_value}")
    duty_body = f"{common.body.rstrip()}\n\n{duty.body.rstrip()}"
    task_body = request.instruction.body.rstrip("\n")
    return (
        "\n".join(header)
        + "\n\n## Duty Contract\n\n"
        + duty_body
        + "\n\n## Task Instructions\n\n"
        + task_body
        + "\n"
    )


def _invocation_digests(
    request: AgentInvocationRequest,
    duty: DutyContract,
    prompt_bytes: bytes,
) -> dict[str, str]:
    assignment = _assignment_payload(request.assignment)
    instruction = {
        "anchorLines": list(request.instruction.anchor_lines),
        # Digest what the prompt can actually carry. Composition writes
        # `body.rstrip("\n")`, so trailing blank lines never reach the file —
        # and `verify` can only rebuild the body from the prompt. Hashing the
        # raw text made an instruction file ending in a blank line unverifiable
        # forever: neither candidate `verify` reconstructs could equal a digest
        # over bytes the prompt does not contain.
        "body": request.instruction.body.rstrip("\n"),
        "sourcePaths": [
            _source_payload(source) for source in request.instruction.source_paths
        ],
    }
    selected_names = ["common.md", duty.source_path.name]
    return {
        "catalogDigest": digest_duty_catalog(request.duty_root),
        "assignmentDigest": _sha256(_canonical_json(assignment)),
        "dutyDigest": _digest_framed_files(request.duty_root, selected_names),
        "instructionDigest": _sha256(_canonical_json(instruction)),
        "promptDigest": _sha256(prompt_bytes),
    }


def _metadata_payload(
    request: AgentInvocationRequest,
    duty: DutyContract,
    digests: Mapping[str, str],
) -> dict[str, object]:
    mode = "run" if request.run_manifest_path is not None else "standalone"
    manifest_path = (
        _project_relative(
            request.run_manifest_path,
            request.project_root,
            must_exist=True,
        )
        if request.run_manifest_path is not None
        else None
    )
    payload: dict[str, object] = {
        "schemaVersion": 1,
        "invocationId": request.invocation_id,
        "audience": request.audience,
        "dispatchKind": request.dispatch_kind,
        "workerId": request.worker_id,
        "assignmentRef": request.assignment_ref,
        "contractSource": {
            "mode": mode,
            "runManifestPath": manifest_path,
            "dutyRootPath": _project_relative(
                request.duty_root,
                request.project_root,
                must_exist=True,
            ),
        },
        "modelAssignment": _assignment_payload(request.assignment),
        "dutyContract": {"id": duty.id, "version": duty.version},
        "instruction": {
            "sourcePaths": [
                _source_payload(source)
                for source in request.instruction.source_paths
            ]
        },
        "prompt": {
            "path": _project_relative(
                request.prompt_path,
                request.project_root,
                must_exist=False,
            )
        },
        "digests": dict(digests),
    }
    if _request_identity_version(request) == 2:
        manifest = _load_json_object(request.run_manifest_path, "run manifest")
        identity = invocation_execution_identity_from_manifest(
            manifest,
            assignment=request.assignment,
            assignment_ref=request.assignment_ref,
            duty_id=request.duty_id,
            role_execution_ref=request.role_execution_ref,
        )
        if identity is None:
            raise AgentInvocationError(
                "v2 execution identity does not match a canonical role execution"
            )
        payload.update({
            "schemaVersion": "2.0",
            "executionIdentityVersion": 2,
            "invocationRef": request.invocation_ref,
            "participantRef": request.participant_ref,
            "roleExecutionRef": request.role_execution_ref,
            "executionLabel": identity.execution_label,
            "dutyId": request.duty_id,
            "attempt": request.attempt,
        })
        payload.pop("workerId")
    return payload


def _reservation_spec(
    request: AgentInvocationRequest,
) -> tuple[Path | None, dict[str, object] | None]:
    if request.run_manifest_path is None:
        return None, None
    manifest = _load_json_object(request.run_manifest_path, "run manifest")
    contract = manifest.get("agentContract")
    if not isinstance(contract, dict):
        raise AgentInvocationError("run manifest has no agent contract")
    _validate_manifest_contract(request, manifest, contract)
    reservation_value = contract.get("invocationReservationRootPath")
    reservation_root = _project_path(
        request.project_root,
        reservation_value,
        must_exist=False,
    )
    reservation: dict[str, object] = {
        "schemaVersion": 1,
        "invocationId": request.invocation_id,
        "workerId": request.worker_id,
        "assignmentRef": request.assignment_ref,
        "audience": request.audience,
        "dispatchKind": request.dispatch_kind,
        "promptPath": _project_relative(
            request.prompt_path,
            request.project_root,
            must_exist=False,
        ),
        "metadataPath": _project_relative(
            request.metadata_path,
            request.project_root,
            must_exist=False,
        ),
    }
    if _request_identity_version(request) == 2:
        reservation.update({
            "schemaVersion": "2.0",
            "executionIdentityVersion": 2,
            "invocationRef": request.invocation_ref,
            "participantRef": request.participant_ref,
            "roleExecutionRef": request.role_execution_ref,
            "dutyId": request.duty_id,
            "attempt": request.attempt,
        })
        reservation.pop("workerId")
    return reservation_root, reservation


def _validate_manifest_contract(
    request: AgentInvocationRequest,
    manifest: Mapping[str, object],
    contract: Mapping[str, object],
) -> None:
    duty_path = _project_relative(
        request.duty_root,
        request.project_root,
        must_exist=True,
    )
    if contract.get("dutyRootPath") != duty_path:
        raise AgentInvocationError("duty root does not match run manifest")
    if contract.get("catalogDigest") != digest_duty_catalog(request.duty_root):
        raise AgentInvocationError("catalog digest does not match run duty snapshot")
    allowed = contract.get("allowedAudiences")
    if not isinstance(allowed, list) or request.audience not in allowed:
        raise AgentInvocationError("audience is not allowed by run manifest")
    assignments = manifest.get("invocationAssignments")
    manifest_assignment = (
        assignments.get(request.assignment_ref)
        if isinstance(assignments, dict)
        else None
    )
    if _model_assignment_projection(manifest_assignment) != _assignment_payload(
        request.assignment
    ):
        raise AgentInvocationError("model assignment does not match run manifest")
    _validate_authorized_prompt_path(request, contract)


def _validate_authorized_prompt_path(
    request: AgentInvocationRequest,
    contract: Mapping[str, object],
) -> None:
    authorized = contract.get("authorizedPaths")
    roots = authorized.get("promptRoots") if isinstance(authorized, dict) else None
    if not isinstance(roots, list) or not roots:
        raise AgentInvocationError("run manifest has no authorized prompt roots")
    prompt = request.prompt_path.resolve(strict=False)
    allowed = [
        _project_path(request.project_root, value, must_exist=True)
        for value in roots
    ]
    if not any(_is_relative_to(prompt, root) for root in allowed):
        raise AgentInvocationError("prompt path is outside authorized prompt roots")


def _publish_or_reuse_reservation(materialized: _MaterializedInvocation) -> None:
    root = materialized.reservation_root
    reservation = materialized.reservation
    if root is None or reservation is None:
        return
    path = root / f"{materialized.prepared.invocation_id}.json"
    if path.exists():
        if _read_json_if_object(path) != reservation:
            raise AgentInvocationError(
                "invocation reservation conflicts with existing invocation",
                reason="invocation_reservation_conflict",
            )
        return
    _publish_exclusive(path, _pretty_json(reservation))


def _publish_or_reuse(
    request: AgentInvocationRequest, materialized: _MaterializedInvocation
) -> None:
    prompt = materialized.prepared.prompt_path
    metadata = materialized.prepared.metadata_path
    prompt_exists = prompt.exists()
    metadata_exists = metadata.exists()
    if metadata_exists and not prompt_exists:
        raise _existing_conflict("metadata exists without prompt")
    if prompt_exists and prompt.read_bytes() != materialized.prompt_bytes:
        _replace_undispatched(request, materialized, _prompt_difference(prompt, materialized))
        return
    if prompt_exists and metadata_exists:
        if metadata.read_bytes() != materialized.metadata_bytes:
            # Same rule, same artifact pair. The prompt and its adjacent
            # metadata are published together and describe one call, so a
            # metadata-only difference is the identical situation as a
            # prompt difference — and it used to be the one the documented exit
            # could not reach, leaving hand-deleting the files as the only move.
            _replace_undispatched(
                request, materialized, _metadata_difference(metadata, materialized)
            )
        return
    if not prompt_exists:
        _publish_exclusive(prompt, materialized.prompt_bytes)
    _publish_exclusive(metadata, materialized.metadata_bytes)


def _metadata_difference(
    metadata: Path, materialized: _MaterializedInvocation
) -> str:
    """Which metadata fields differ, named rather than counted.

    Metadata is JSON, so the useful answer is which keys moved — a line number
    would point at whatever the serializer happened to order first. Digests are
    nested one level down and are the fields that actually differ in practice,
    so they are reported by name too.
    """
    try:
        existing = load_owned_object(metadata, artifact="agent invocation metadata")
    except JsonBoundaryError as exc:
        return f"existing metadata is unreadable: {exc}"
    composed = materialized.metadata
    if not isinstance(existing, dict):
        return "existing metadata is not an object"
    changed = sorted(
        key
        for key in set(existing) | set(composed)
        if existing.get(key) != composed.get(key)
    )
    if changed == ["digests"]:
        digests = existing.get("digests")
        composed_digests = composed.get("digests")
        if isinstance(digests, dict) and isinstance(composed_digests, dict):
            moved = sorted(
                key
                for key in set(digests) | set(composed_digests)
                if digests.get(key) != composed_digests.get(key)
            )
            return f"digests differ: {', '.join(moved)}"
    return f"fields differ: {', '.join(changed)}" if changed else "byte-level difference"


def _prompt_difference(prompt: Path, materialized: _MaterializedInvocation) -> str:
    """What actually differs between the published prompt and this one.

    "existing prompt differs" alone cannot be acted on: it does not say whether
    the file on disk is a stale sibling, a partially written artifact, or the
    same prompt built from an edited instruction file — and the evidence is
    destroyed by the very republish that unblocks the caller. One real
    investigation ended undecided for exactly this reason. Digests identify the
    two bodies, and the first differing line points at the edit.
    """
    try:
        existing = prompt.read_bytes()
    except OSError as exc:
        return f"existing prompt is unreadable: {exc}"
    composed = materialized.prompt_bytes
    detail = (
        f"existing {_sha256(existing)}, composed {_sha256(composed)}"
    )
    existing_lines = existing.decode("utf-8", errors="replace").splitlines()
    composed_lines = composed.decode("utf-8", errors="replace").splitlines()
    for index, (left, right) in enumerate(zip(existing_lines, composed_lines), start=1):
        if left != right:
            return f"{detail}, first differing line {index}"
    shorter = min(len(existing_lines), len(composed_lines))
    return f"{detail}, identical through line {shorter} then one body continues"


def _replace_undispatched(
    request: AgentInvocationRequest,
    materialized: _MaterializedInvocation,
    difference: str,
) -> None:
    """Rewrite a prompt this invocation id wrote but no dispatch ever used.

    Immutability protects the audit chain: a prompt that a worker ran must keep
    reading the way it ran. A prompt that failed a pre-dispatch gate ran
    nowhere, and there the same rule cost a lead its only clean way forward —
    fix the instruction file and the republish is refused, so the remaining
    moves were to delete the reservation by hand or to burn a second invocation
    id and blur which call is which. Both damage the chain this protects.

    The exit is explicit (`--replace-undispatched`) and it is checked, not
    trusted: a single recorded dispatch against this invocation id — worker or
    agent — puts the prompt back under the original rule.
    """
    if not request.replace_undispatched:
        raise _existing_conflict(f"existing prompt differs ({difference})")
    dispatched = _recorded_dispatch_ids(request)
    if dispatched:
        raise _existing_conflict(
            "existing prompt differs and this invocation was already "
            f"dispatched as {', '.join(sorted(dispatched))}"
        )
    _publish_exclusive(
        materialized.prepared.prompt_path, materialized.prompt_bytes, replace=True
    )
    _publish_exclusive(
        materialized.prepared.metadata_path,
        materialized.metadata_bytes,
        replace=True,
    )


def _recorded_dispatch_ids(request: AgentInvocationRequest) -> set[str]:
    """Dispatch rows in team-state that already reference this invocation id.

    Both ledgers are read: `agentDispatches` is written by the host-native
    spec-link gate, `workerDispatches` by the deterministic worker dispatch, and
    a prompt used by either one is history.
    """
    if request.run_manifest_path is None:
        return set()
    manifest = _load_json_object(request.run_manifest_path, "run manifest")
    team_state_value = manifest.get("teamStatePath")
    if not isinstance(team_state_value, str) or not team_state_value.strip():
        return set()
    team_state_path = _project_path(
        request.project_root, team_state_value, must_exist=False
    )
    team_state = _read_json_if_object(team_state_path)
    if team_state is None:
        return set()
    found: set[str] = set()
    for key in ("agentDispatches", "workerDispatches"):
        rows = team_state.get(key)
        if not isinstance(rows, list):
            continue
        for row in rows:
            if not isinstance(row, Mapping):
                continue
            if row.get("invocationId") != request.invocation_id:
                continue
            found.add(str(row.get("dispatchId") or request.invocation_id))
    return found


def _existing_conflict(detail: str) -> AgentInvocationError:
    return AgentInvocationError(
        f"existing_invocation_conflict: {detail}",
        reason="existing_invocation_conflict",
    )


@contextlib.contextmanager
def _exclusive_lock(path: Path) -> Iterator[None]:
    path.parent.mkdir(parents=True, exist_ok=True)
    handle = path.open("a+")
    try:
        fcntl.flock(handle.fileno(), fcntl.LOCK_EX)
        yield
    finally:
        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
        handle.close()


def _prompt_lock_path(prompt_path: Path) -> Path:
    return prompt_path.with_name(prompt_path.name + ".publish.lock")


def _publish_or_reuse_exact(path: Path, body: bytes) -> None:
    if path.exists():
        if path.read_bytes() != body:
            raise _existing_conflict(f"published path differs: {path}")
        return
    _publish_exclusive(path, body)


def _publish_exclusive(path: Path, body: bytes, *, replace: bool = False) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temp_path: Path | None = None
    try:
        with tempfile.NamedTemporaryFile(
            dir=path.parent,
            prefix=f".{path.name}.",
            suffix=".tmp",
            delete=False,
        ) as handle:
            temp_path = Path(handle.name)
            handle.write(body)
            handle.flush()
            os.fsync(handle.fileno())
        # `link` is what makes a first publish exclusive — it fails rather than
        # overwrite. A sanctioned replacement wants the opposite and still needs
        # to be atomic, so it swaps the same fully-written temp file into place.
        if replace:
            os.replace(temp_path, path)
            temp_path = None
        else:
            os.link(temp_path, path)
    except FileExistsError as exc:
        raise _existing_conflict(f"published path already exists: {path}") from exc
    finally:
        if temp_path is not None:
            temp_path.unlink(missing_ok=True)


def _verify_expected_identity(
    metadata: Mapping[str, object],
    *,
    expected_invocation_id: str | None,
    expected_worker_id: str | None,
    expected_assignment_ref: str | None,
    expected_audience: AgentAudience | None,
    expected_participant_ref: str | None,
    expected_role_execution_ref: str | None,
    expected_invocation_ref: str | None,
) -> list[str]:
    expected = {
        "invocationId": expected_invocation_id,
        "assignmentRef": expected_assignment_ref,
        "audience": expected_audience,
        "participantRef": expected_participant_ref,
        "roleExecutionRef": expected_role_execution_ref,
        "invocationRef": expected_invocation_ref,
    }
    if metadata.get("schemaVersion") == 1:
        expected["workerId"] = expected_worker_id
    errors = []
    for key, value in expected.items():
        if value is not None and metadata.get(key) != value:
            errors.append(f"{key} does not match expected invocation identity")
    return errors


def _verify_prompt_file(
    metadata: Mapping[str, object],
    project_root: Path,
) -> list[str]:
    prompt = metadata["prompt"]
    digests = metadata["digests"]
    assert isinstance(prompt, dict)
    assert isinstance(digests, dict)
    try:
        path = _project_path(project_root, prompt["path"], must_exist=True)
    except AgentInvocationError:
        return ["prompt path is invalid or missing"]
    body = path.read_bytes()
    if digests["promptDigest"] != _sha256(body):
        return ["prompt digest does not match prompt file"]
    errors = []
    errors.extend(_verify_instruction_digest(metadata, body))
    errors.extend(_verify_model_header(metadata, body))
    return errors


def _verify_instruction_digest(
    metadata: Mapping[str, object],
    prompt_bytes: bytes,
) -> list[str]:
    try:
        prompt = prompt_bytes.decode("utf-8")
        prefix, task_body = _split_prompt(prompt)
        anchor_lines = _extract_anchor_lines(prefix)
    except (UnicodeDecodeError, ValueError):
        return ["prompt structure is invalid"]
    instruction = metadata["instruction"]
    digests = metadata["digests"]
    assert isinstance(instruction, dict)
    assert isinstance(digests, dict)
    base = {
        "anchorLines": anchor_lines,
        "sourcePaths": instruction["sourcePaths"],
    }
    candidates = (
        {**base, "body": task_body},
        {**base, "body": task_body.rstrip("\n")},
    )
    actual = digests["instructionDigest"]
    if not any(_sha256(_canonical_json(candidate)) == actual for candidate in candidates):
        return ["instruction digest does not match prompt instructions"]
    return []


def _verify_model_header(
    metadata: Mapping[str, object],
    prompt_bytes: bytes,
) -> list[str]:
    assignment = metadata["modelAssignment"]
    assert isinstance(assignment, dict)
    prompt = prompt_bytes.decode("utf-8")
    required = {
        f"**Provider:** {assignment['provider']}",
        f"**Model:** {assignment['model']}",
        f"**Model execution value:** {assignment['modelExecutionValue']}",
        f"**Runner:** {assignment['runner']}",
        f"**Host runtime:** {assignment['hostRuntime']}",
    }
    host_model = assignment["hostModelValue"]
    if host_model is not None:
        required.add(f"**Host model value:** {host_model}")
    prefix = prompt.split("\n\n## Duty Contract\n\n", 1)[0]
    if not required.issubset(set(prefix.splitlines())):
        return ["prompt model header does not match model assignment"]
    return []


def _split_prompt(prompt: str) -> tuple[str, str]:
    duty_marker = "\n\n## Duty Contract\n\n"
    task_marker = "\n\n## Task Instructions\n\n"
    if prompt.count(duty_marker) != 1 or prompt.count(task_marker) != 1:
        raise ValueError("prompt markers are invalid")
    prefix, remainder = prompt.split(duty_marker, 1)
    _duty, task = remainder.split(task_marker, 1)
    return prefix, task


def _extract_anchor_lines(prefix: str) -> list[str]:
    lines = prefix.splitlines()
    try:
        provider_index = next(
            index for index, line in enumerate(lines) if line.startswith("**Provider:**")
        )
    except StopIteration as exc:
        raise ValueError("model header is missing") from exc
    return lines[:provider_index]


def _verify_assignment(
    metadata: Mapping[str, object],
    expected: AgentModelAssignment | None,
) -> list[str]:
    assignment = metadata["modelAssignment"]
    digests = metadata["digests"]
    assert isinstance(assignment, dict)
    assert isinstance(digests, dict)
    errors = []
    if digests["assignmentDigest"] != _sha256(_canonical_json(assignment)):
        errors.append("assignment digest does not match model assignment")
    if expected is not None and assignment != _assignment_payload(expected):
        errors.append("model assignment does not match expected assignment")
    return errors


def _verify_contract_source(
    metadata: Mapping[str, object],
    *,
    metadata_path: Path,
    project_root: Path,
    expected_run_manifest_path: Path | None,
) -> list[str]:
    source = metadata["contractSource"]
    assert isinstance(source, dict)
    mode = source["mode"]
    if mode == "standalone":
        if expected_run_manifest_path is not None:
            return ["standalone invocation cannot use a run manifest"]
        errors = _verify_metadata_adjacency(metadata, metadata_path, project_root)
        errors.extend(_verify_duty_snapshot(metadata, project_root, None))
        return errors
    if mode != "run":
        return ["contract source mode is invalid"]
    if expected_run_manifest_path is None:
        return ["run invocation requires expected run manifest"]
    errors = _verify_metadata_adjacency(metadata, metadata_path, project_root)
    errors.extend(
        _verify_run_contract(
            metadata,
            project_root=project_root,
            expected_run_manifest_path=expected_run_manifest_path,
        )
    )
    return errors


def _verify_metadata_adjacency(
    metadata: Mapping[str, object],
    metadata_path: Path,
    project_root: Path,
) -> list[str]:
    prompt = metadata["prompt"]
    assert isinstance(prompt, dict)
    try:
        prompt_path = _project_path(project_root, prompt["path"], must_exist=False)
        expected = prompt_path.with_name(prompt_path.name + ".meta.json")
        if metadata_path.resolve(strict=True) != expected.resolve(strict=True):
            return ["metadata path is not adjacent to prompt path"]
    except (AgentInvocationError, OSError):
        return ["metadata path is invalid"]
    return []


def _verify_run_contract(
    metadata: Mapping[str, object],
    *,
    project_root: Path,
    expected_run_manifest_path: Path,
) -> list[str]:
    source = metadata["contractSource"]
    assert isinstance(source, dict)
    try:
        recorded = _project_path(
            project_root,
            source["runManifestPath"],
            must_exist=True,
        )
        expected = expected_run_manifest_path.resolve(strict=True)
    except (AgentInvocationError, OSError):
        return ["run manifest path is invalid"]
    if recorded != expected:
        return ["run manifest path does not match expected run manifest"]
    try:
        manifest = _load_json_object(expected, "run manifest")
    except AgentInvocationError:
        return ["run manifest is invalid"]
    errors = _verify_run_manifest_contract(metadata, manifest, project_root)
    errors.extend(_verify_duty_snapshot(metadata, project_root, manifest))
    errors.extend(_verify_reservation(metadata, manifest, project_root))
    return errors


def _verify_run_manifest_contract(
    metadata: Mapping[str, object],
    manifest: Mapping[str, object],
    project_root: Path,
) -> list[str]:
    source = metadata["contractSource"]
    digests = metadata["digests"]
    assert isinstance(source, dict)
    assert isinstance(digests, dict)
    contract = manifest.get("agentContract")
    if not isinstance(contract, dict):
        return ["run manifest agent contract is missing"]
    errors = []
    if contract.get("dutyRootPath") != source["dutyRootPath"]:
        errors.append("duty root does not match run manifest")
    if contract.get("catalogDigest") != digests["catalogDigest"]:
        errors.append("catalog digest does not match run manifest")
    allowed = contract.get("allowedAudiences")
    if not isinstance(allowed, list) or metadata["audience"] not in allowed:
        errors.append("audience is not allowed by run manifest")
    errors.extend(_verify_manifest_assignment(metadata, manifest))
    if metadata.get("executionIdentityVersion") == 2:
        errors.extend(_verify_manifest_execution_identity(metadata, manifest))
    return errors


def _verify_manifest_assignment(
    metadata: Mapping[str, object],
    manifest: Mapping[str, object],
) -> list[str]:
    assignments = manifest.get("invocationAssignments")
    reference = metadata["assignmentRef"]
    if not isinstance(assignments, dict) or reference not in assignments:
        return ["assignment reference is missing from run manifest"]
    if _model_assignment_projection(assignments[reference]) != metadata["modelAssignment"]:
        return ["model assignment does not match run manifest"]
    return []


def _verify_manifest_execution_identity(
    metadata: Mapping[str, object],
    manifest: Mapping[str, object],
) -> list[str]:
    if metadata["dutyContract"]["id"] != metadata["dutyId"]:
        return ["v2 duty identity does not match duty contract"]
    try:
        identity = invocation_execution_identity_from_manifest(
            manifest,
            assignment=agent_model_assignment_from_payload(
                metadata["modelAssignment"]
            ),
            assignment_ref=str(metadata["assignmentRef"]),
            duty_id=str(metadata["dutyId"]),
            role_execution_ref=str(metadata["roleExecutionRef"]),
        )
    except AgentInvocationError as exc:
        return [str(exc)]
    if (
        identity is None
        or identity.participant_ref != metadata["participantRef"]
        or identity.role_execution_ref != metadata["roleExecutionRef"]
        or identity.execution_label != metadata["executionLabel"]
    ):
        return ["v2 identity does not match a canonical role execution"]
    return []


def _verify_duty_snapshot(
    metadata: Mapping[str, object],
    project_root: Path,
    manifest: Mapping[str, object] | None,
) -> list[str]:
    source = metadata["contractSource"]
    duty_contract = metadata["dutyContract"]
    digests = metadata["digests"]
    assert isinstance(source, dict)
    assert isinstance(duty_contract, dict)
    assert isinstance(digests, dict)
    try:
        root = _project_path(project_root, source["dutyRootPath"], must_exist=True)
        common = load_common_duty_contract(root)
        catalog = load_duty_catalog(root)
        duty = catalog[duty_contract["id"]]
    except (AgentInvocationError, KeyError):
        return ["duty snapshot is invalid"]
    errors = _verify_duty_metadata(duty_contract, duty)
    actual_catalog = digest_duty_catalog(root)
    expected_catalog = (
        manifest["agentContract"]["catalogDigest"]
        if manifest is not None
        else digests["catalogDigest"]
    )
    if actual_catalog != expected_catalog or digests["catalogDigest"] != actual_catalog:
        errors.append("catalog digest does not match run duty snapshot")
        return errors
    actual_duty = _digest_framed_files(root, ["common.md", duty.source_path.name])
    if digests["dutyDigest"] != actual_duty:
        errors.append("duty digest does not match run duty snapshot")
        return errors
    errors.extend(_verify_prompt_duty_body(metadata, project_root, common, duty))
    return errors


def _verify_duty_metadata(
    recorded: Mapping[str, object],
    duty: DutyContract,
) -> list[str]:
    if recorded["id"] != duty.id or recorded["version"] != duty.version:
        return ["duty identity does not match duty snapshot"]
    return []


def _verify_prompt_duty_body(
    metadata: Mapping[str, object],
    project_root: Path,
    common: DutyContract,
    duty: DutyContract,
) -> list[str]:
    prompt_spec = metadata["prompt"]
    assert isinstance(prompt_spec, dict)
    try:
        prompt_path = _project_path(project_root, prompt_spec["path"], must_exist=True)
        prompt = prompt_path.read_text(encoding="utf-8")
        _prefix, remainder = prompt.split("\n\n## Duty Contract\n\n", 1)
        rendered_duty, _task = remainder.split("\n\n## Task Instructions\n\n", 1)
    except (AgentInvocationError, OSError, UnicodeDecodeError, ValueError):
        return ["prompt duty contract does not match duty snapshot"]
    expected = f"{common.body.rstrip()}\n\n{duty.body.rstrip()}"
    if rendered_duty != expected:
        return ["prompt duty contract does not match duty snapshot"]
    return []


def _verify_reservation(
    metadata: Mapping[str, object],
    manifest: Mapping[str, object],
    project_root: Path,
) -> list[str]:
    contract = manifest.get("agentContract")
    if not isinstance(contract, dict):
        return ["run manifest agent contract is missing"]
    try:
        root = _project_path(
            project_root,
            contract["invocationReservationRootPath"],
            must_exist=True,
        )
        path = root / f"{metadata['invocationId']}.json"
        actual = _load_json_object(path, "invocation reservation")
    except (AgentInvocationError, KeyError):
        return ["invocation reservation is missing or invalid"]
    prompt = metadata["prompt"]
    assert isinstance(prompt, dict)
    expected: dict[str, object] = {
        "schemaVersion": 1,
        "invocationId": metadata["invocationId"],
        "assignmentRef": metadata["assignmentRef"],
        "audience": metadata["audience"],
        "dispatchKind": metadata["dispatchKind"],
        "promptPath": prompt["path"],
        "metadataPath": f"{prompt['path']}.meta.json",
    }
    if metadata.get("executionIdentityVersion") == 2:
        expected.update({
            "schemaVersion": "2.0",
            "executionIdentityVersion": 2,
            "invocationRef": metadata["invocationRef"],
            "participantRef": metadata["participantRef"],
            "roleExecutionRef": metadata["roleExecutionRef"],
            "dutyId": metadata["dutyId"],
            "attempt": metadata["attempt"],
        })
    else:
        expected["workerId"] = metadata["workerId"]
    if actual != expected:
        return ["invocation reservation does not match invocation metadata"]
    return []


def _metadata_schema_is_exact(metadata: Mapping[str, object]) -> bool:
    schema = metadata.get("schemaVersion")
    if schema == 1:
        if set(metadata) != _TOP_LEVEL_KEYS:
            return False
    elif schema == "2.0":
        if (
            set(metadata) != _V2_TOP_LEVEL_KEYS
            or metadata.get("executionIdentityVersion") != 2
            or not all(
                isinstance(metadata.get(key), str) and metadata[key]
                for key in (
                    "invocationRef",
                    "participantRef",
                    "roleExecutionRef",
                    "executionLabel",
                    "dutyId",
                )
            )
            or not isinstance(metadata.get("attempt"), int)
            or isinstance(metadata.get("attempt"), bool)
            or metadata["attempt"] < 1
        ):
            return False
    else:
        return False
    for key, expected_keys in _NESTED_KEYS.items():
        value = metadata.get(key)
        if not isinstance(value, dict) or set(value) != expected_keys:
            return False
    sources = metadata["instruction"]["sourcePaths"]
    if not isinstance(sources, list) or not sources:
        return False
    if any(not isinstance(item, dict) or set(item) != {"kind", "path"} for item in sources):
        return False
    return _digest_paths(metadata) == {
        "digests.catalogDigest",
        "digests.assignmentDigest",
        "digests.dutyDigest",
        "digests.instructionDigest",
        "digests.promptDigest",
    }


def _model_assignment_projection(payload: object) -> dict[str, object] | None:
    if not isinstance(payload, Mapping):
        return None
    keys = _NESTED_KEYS["modelAssignment"]
    if not keys.issubset(payload):
        return None
    return {key: payload[key] for key in keys}


def _digest_paths(value: object, prefix: str = "") -> set[str]:
    if not isinstance(value, dict):
        return set()
    found: set[str] = set()
    for key, child in value.items():
        path = f"{prefix}.{key}" if prefix else key
        if key == "digest" or key.endswith("Digest"):
            found.add(path)
        found.update(_digest_paths(child, path))
    return found


def _assignment_payload(assignment: AgentModelAssignment) -> dict[str, object]:
    return {
        "provider": assignment.provider,
        "model": assignment.model,
        "modelExecutionValue": assignment.model_execution_value,
        "runner": assignment.runner,
        "hostRuntime": assignment.host_runtime,
        "hostModelValue": assignment.host_model_value,
    }


def _source_payload(source: AgentInstructionSource) -> dict[str, str]:
    if source.kind not in {"project", "runtime"}:
        raise AgentInvocationError("instruction sourcePaths contains an invalid kind")
    path = source.path
    pure = PurePosixPath(path)
    if (
        not path
        or "\\" in path
        or pure.is_absolute()
        or any(part in {"", ".", ".."} for part in pure.parts)
    ):
        raise AgentInvocationError("instruction sourcePaths must use relative POSIX paths")
    return {"kind": source.kind, "path": pure.as_posix()}


def _project_relative(
    path: Path,
    project_root: Path,
    *,
    must_exist: bool,
) -> str:
    root = project_root.resolve(strict=True)
    try:
        if must_exist:
            resolved = path.resolve(strict=True)
        else:
            resolved = path.parent.resolve(strict=True) / path.name
        return resolved.relative_to(root).as_posix()
    except (OSError, ValueError) as exc:
        raise AgentInvocationError(f"path escapes project root: {path}") from exc


def _project_path(
    project_root: Path,
    value: object,
    *,
    must_exist: bool,
) -> Path:
    if not isinstance(value, str) or not value:
        raise AgentInvocationError("project-relative path is invalid")
    pure = PurePosixPath(value)
    if pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts):
        raise AgentInvocationError("project-relative path is invalid")
    candidate = project_root.joinpath(*pure.parts)
    _project_relative(candidate, project_root, must_exist=must_exist)
    return candidate.resolve(strict=must_exist)


def _is_relative_to(path: Path, root: Path) -> bool:
    try:
        path.relative_to(root)
    except ValueError:
        return False
    return True


def _digest_framed_files(root: Path, relative_names: list[str]) -> str:
    framed = bytearray(b"okstra-digest-v1\0")
    for name in sorted(relative_names):
        name_bytes = name.encode("utf-8")
        body = (root / name).read_bytes()
        framed.extend(len(name_bytes).to_bytes(8, "big"))
        framed.extend(name_bytes)
        framed.extend(len(body).to_bytes(8, "big"))
        framed.extend(body)
    return _sha256(bytes(framed))


def _sha256(data: bytes) -> str:
    return f"sha256:{hashlib.sha256(data).hexdigest()}"


def _canonical_json(value: object) -> bytes:
    return json.dumps(
        value,
        ensure_ascii=False,
        sort_keys=True,
        separators=(",", ":"),
        allow_nan=False,
    ).encode("utf-8")


def _pretty_json(value: object) -> bytes:
    return (
        json.dumps(value, ensure_ascii=False, indent=2, allow_nan=False) + "\n"
    ).encode("utf-8")


def _load_json_object(path: Path, label: str) -> dict[str, object]:
    value = _read_json_if_object(path)
    if value is None:
        raise AgentInvocationError(f"{label} is missing or invalid: {path}")
    return value


def _read_json_if_object(path: Path) -> dict[str, object] | None:
    try:
        value = load_owned_object(path, artifact="agent invocation metadata")
    except JsonBoundaryError:
        return None
    return value if isinstance(value, dict) else None


def _read_metadata(path: Path) -> dict[str, object] | None:
    return _read_json_if_object(path)


def _deduplicate(errors: list[str]) -> list[str]:
    return list(dict.fromkeys(errors))


def _load_role_duty(path: Path) -> DutyContract:
    fields, body = _parse_duty_file(path)
    if set(fields) != {"id", "version", "kind", "appliesTo"}:
        raise AgentInvocationError(f"invalid role duty frontmatter: {path}")
    audience = fields["appliesTo"]
    if audience not in _SUPPORTED_AUDIENCES:
        raise AgentInvocationError(f"unknown duty audience: {audience}")
    if fields["kind"] != "role" or fields["id"] != audience:
        raise AgentInvocationError(f"invalid role duty frontmatter: {path}")
    _validate_duty_sections(body, ROLE_DUTY_SECTIONS, "role", path)
    return DutyContract(
        id=fields["id"],
        version=_parse_version(fields["version"], path),
        kind="role",
        applies_to=audience,
        body=body,
        source_path=path,
    )


def _parse_duty_file(path: Path) -> tuple[dict[str, str], str]:
    try:
        text = path.read_text(encoding="utf-8")
    except OSError as exc:
        raise AgentInvocationError(f"cannot read duty contract: {path}") from exc
    lines = text.splitlines(keepends=True)
    if not lines or lines[0].strip() != "---":
        raise AgentInvocationError(f"missing duty frontmatter: {path}")
    try:
        end = next(index for index, line in enumerate(lines[1:], 1) if line.strip() == "---")
    except StopIteration as exc:
        raise AgentInvocationError(f"unterminated duty frontmatter: {path}") from exc
    fields: dict[str, str] = {}
    for line in lines[1:end]:
        key, separator, value = line.partition(":")
        if not separator or not key.strip() or key.strip() in fields:
            raise AgentInvocationError(f"invalid duty frontmatter: {path}")
        fields[key.strip()] = value.strip()
    return fields, "".join(lines[end + 1 :]).lstrip("\n")


def _validate_duty_sections(
    body: str,
    required: tuple[str, ...],
    kind: str,
    path: Path,
) -> None:
    matches = list(_DUTY_SECTION_RE.finditer(body))
    sections: dict[str, str] = {}
    for index, match in enumerate(matches):
        name = match.group(1).strip()
        if name in sections:
            raise AgentInvocationError(f"duplicate {kind} duty section {name}: {path}")
        end = matches[index + 1].start() if index + 1 < len(matches) else len(body)
        sections[name] = body[match.end() : end].strip()
    missing = [name for name in required if name not in sections]
    if missing:
        raise AgentInvocationError(
            f"missing {kind} duty sections: {', '.join(missing)}: {path}"
        )
    empty = [name for name in required if not sections[name]]
    if empty:
        raise AgentInvocationError(
            f"empty {kind} duty sections: {', '.join(empty)}: {path}"
        )


def _parse_version(value: str, path: Path) -> int:
    try:
        version = int(value)
    except ValueError as exc:
        raise AgentInvocationError(f"invalid duty version: {path}") from exc
    if version < 1:
        raise AgentInvocationError(f"invalid duty version: {path}")
    return version