"""Append-only persistence and validation for execution identity v2."""
from __future__ import annotations

import json
import os
import tempfile
from collections.abc import Mapping, Sequence
from functools import lru_cache
from pathlib import Path
from typing import Any, Callable

from .domain.provider import HostModelBinding, ServedModelAttestation
from .domain.worker_exec import WorkerWriteCapability
from .execution_identity import (
    Attempt,
    EXECUTION_IDENTITY_VERSION,
    ExecutionManifest,
    ExecutionManifestError,
    Invocation,
    ParticipantAssignment,
    RoleExecution,
    V2_SCHEMA_VERSION,
    execution_label,
    identity_surface_from_payload,
    read_identity_surface,
    synthesize_v1_execution_view,
    validate_task_role_execution_index,
)
from .final_report_schema import SchemaError, load_schema, validate as validate_schema
from .json_boundary import JsonBoundaryError, load_owned_object, serialize_owned_object
from .run_context import task_mutex
from .write_policy import (
    WritePolicyError,
    validate_write_enforcement_payload,
    validate_write_policy_payload,
    write_policy_digest,
)


_IDENTITY_KEYS = (
    "schemaVersion",
    "executionIdentityVersion",
    "entryMode",
    "participantAssignments",
    "roleExecutions",
    "invocations",
    "attempts",
    "attemptEvidenceSeals",
    "mutationRecoveryDecisions",
)


def read_execution_manifest(path: Path) -> ExecutionManifest:
    manifest, _payload = read_execution_manifest_view(path)
    return manifest


def read_execution_manifest_view(
    path: Path,
) -> tuple[ExecutionManifest, Mapping[str, Any]]:
    """Read one validated manifest snapshot with its enclosing run authority."""
    surface = read_identity_surface(path, surface="run-manifest")
    if surface.legacy:
        return synthesize_v1_execution_view(surface.payload), surface.payload
    validate_execution_manifest_payload(surface.payload)
    return _execution_manifest_from_payload(surface.payload), surface.payload


def write_execution_manifest(path: Path, manifest: ExecutionManifest) -> None:
    if manifest.legacy:
        raise ExecutionManifestError("cannot write a v1 memory view")
    if path.exists():
        raise ExecutionManifestError(f"execution manifest already exists: {path}")
    payload = manifest.to_payload()
    validate_execution_manifest_payload(payload)
    _write_json_atomic(path, payload, create_only=True)


def append_role_execution(
    path: Path,
    row: RoleExecution,
    *,
    task_key: str,
    task_manifest_path: Path | None = None,
    run_ref: str | None = None,
) -> None:
    with task_mutex(task_key):
        manifest = read_execution_manifest(path)
        if manifest.legacy:
            raise ExecutionManifestError("cannot append execution identity to v1 data")
        task_payload = None
        if task_manifest_path is not None and run_ref is not None:
            task_payload = _read_json_object(task_manifest_path)
            _validated_existing_task_index(task_manifest_path, task_payload)
            _validate_run_ref_for_manifest(
                path,
                task_manifest_path,
                task_payload,
                run_ref,
            )
        _validate_next_role_execution(manifest, row)
        updated = _replace_manifest(
            manifest,
            role_executions=(*manifest.role_executions, row),
        )
        if task_payload is not None:
            _validate_new_task_index_row(updated, row)
        _write_manifest_over_existing(path, updated)
        if (
            task_manifest_path is not None
            and run_ref is not None
            and task_payload is not None
        ):
            _append_task_index_unlocked(
                task_manifest_path,
                run_ref,
                row,
                task_payload,
            )


def _drift_detail(existing: Invocation, incoming: Invocation) -> str:
    """어긋난 필드를 이름으로 돌려준다.

    `invocationRef drift: <ref>` 만으로는 무엇이 달라졌는지 알 수 없어, 리드가
    새 invocation id 를 몇 개씩 만들어 보는 것 말고 할 수 있는 일이 없었다.
    같은 ref 로 두 번째 예약이 오는 것 자체는 정상 경로(재materialize)이고,
    거절해야 하는 것은 '내용이 달라진' 경우뿐이므로 그 내용을 이름 붙인다.
    """
    before = existing.to_payload()
    after = incoming.to_payload()
    changed = sorted(
        key for key in (*before, *after) if before.get(key) != after.get(key)
    )
    return ", ".join(changed) if changed else "(no field differs)"


def reserve_derived_role_invocation(
    path: Path,
    *,
    source_role_execution_ref: str,
    role: str,
    invocation_factory: Callable[[RoleExecution], Invocation],
    task_key: str,
    task_manifest_path: Path,
    run_ref: str,
) -> tuple[RoleExecution, Invocation]:
    """Reserve one derived role and logical invocation under the task lock."""
    with task_mutex(task_key):
        manifest = read_execution_manifest(path)
        if manifest.legacy:
            raise ExecutionManifestError("cannot append execution identity to v1 data")
        task_payload = _read_json_object(task_manifest_path)
        task_payload_before_recovery = dict(task_payload)
        _rebuild_task_identity_index_unlocked(task_manifest_path, task_payload)
        _validate_run_ref_for_manifest(
            path,
            task_manifest_path,
            task_payload,
            run_ref,
        )
        source = next(
            (
                item
                for item in manifest.role_executions
                if item.role_execution_ref == source_role_execution_ref
            ),
            None,
        )
        if source is None:
            raise ExecutionManifestError(
                f"source role execution is unknown: {source_role_execution_ref}"
            )
        derived, created = _derived_role_execution(manifest, source, role)
        invocation = invocation_factory(derived)
        if not isinstance(invocation, Invocation):
            raise ExecutionManifestError("invocation factory must return Invocation")
        existing_invocation = next(
            (
                item
                for item in manifest.invocations
                if item.invocation_ref == invocation.invocation_ref
            ),
            None,
        )
        if existing_invocation is not None:
            if existing_invocation != invocation:
                raise ExecutionManifestError(
                    f"invocationRef drift: {invocation.invocation_ref} "
                    f"(differs in: {_drift_detail(existing_invocation, invocation)})"
                )
            if created:
                raise ExecutionManifestError(
                    "existing invocation cannot reference an unpersisted role execution"
                )
            recovered_task_payload = _task_payload_with_role_index(
                task_payload,
                run_ref,
                derived,
            )
            if recovered_task_payload != task_payload_before_recovery:
                _write_json_atomic(task_manifest_path, recovered_task_payload)
            return derived, existing_invocation

        roles = (
            (*manifest.role_executions, derived)
            if created
            else manifest.role_executions
        )
        with_role = _replace_manifest(manifest, role_executions=roles)
        if created:
            _validate_next_role_execution(manifest, derived)
        _validate_next_invocation(with_role, invocation)
        updated = _replace_manifest(
            with_role,
            invocations=(*with_role.invocations, invocation),
        )
        validate_execution_manifest_payload(updated.to_payload())
        updated_task_payload = _task_payload_with_role_index(
            task_payload,
            run_ref,
            derived,
        )
        _write_manifest_and_task_index(
            path,
            updated,
            task_manifest_path,
            updated_task_payload,
        )
        return derived, invocation


def append_invocation(path: Path, row: Invocation, *, task_key: str) -> None:
    with task_mutex(task_key):
        manifest = read_execution_manifest(path)
        if manifest.legacy:
            raise ExecutionManifestError("cannot append invocation to v1 data")
        _validate_next_invocation(manifest, row)
        updated = _replace_manifest(
            manifest,
            invocations=(*manifest.invocations, row),
        )
        _write_manifest_over_existing(path, updated)


def ensure_invocation(path: Path, row: Invocation, *, task_key: str) -> Invocation:
    """Create one logical invocation or return its byte-equivalent value."""
    with task_mutex(task_key):
        manifest = read_execution_manifest(path)
        if manifest.legacy:
            raise ExecutionManifestError("cannot append invocation to v1 data")
        existing = next(
            (
                item
                for item in manifest.invocations
                if item.invocation_ref == row.invocation_ref
            ),
            None,
        )
        if existing is not None:
            if existing != row:
                raise ExecutionManifestError(
                    f"invocationRef drift: {row.invocation_ref} "
                    f"(differs in: {_drift_detail(existing, row)})"
                )
            return existing
        _validate_next_invocation(manifest, row)
        updated = _replace_manifest(
            manifest,
            invocations=(*manifest.invocations, row),
        )
        _write_manifest_over_existing(path, updated)
        return row


def append_attempt(path: Path, row: Attempt, *, task_key: str) -> None:
    with task_mutex(task_key):
        manifest = read_execution_manifest(path)
        if manifest.legacy:
            raise ExecutionManifestError("cannot append attempt to v1 data")
        _validate_next_attempt(manifest, row)
        updated = _replace_manifest(
            manifest,
            attempts=(*manifest.attempts, row),
        )
        _write_manifest_over_existing(path, updated)


def record_invocation_attempt(
    path: Path,
    invocation: Invocation,
    attempt: Attempt,
    *,
    task_key: str,
) -> None:
    """Ensure one logical invocation and append its next attempt atomically."""
    with task_mutex(task_key):
        manifest = read_execution_manifest(path)
        if manifest.legacy:
            raise ExecutionManifestError("cannot append execution identity to v1 data")
        existing = next(
            (
                item
                for item in manifest.invocations
                if item.invocation_ref == invocation.invocation_ref
            ),
            None,
        )
        if existing is not None and existing != invocation:
            raise ExecutionManifestError(
                f"invocationRef drift: {invocation.invocation_ref} "
                f"(differs in: {_drift_detail(existing, invocation)})"
            )
        invocations = manifest.invocations
        if existing is None:
            _validate_next_invocation(manifest, invocation)
            invocations = (*invocations, invocation)
        with_invocation = _replace_manifest(manifest, invocations=invocations)
        _validate_next_attempt(with_invocation, attempt)
        updated = _replace_manifest(
            with_invocation,
            attempts=(*with_invocation.attempts, attempt),
        )
        _write_manifest_over_existing(path, updated)


def finish_attempt_mutation(
    path: Path,
    *,
    invocation_ref: str,
    attempt: int,
    finished_at: str,
    status: str,
    result_path: str | None,
    error_path: str | None,
    change_summary: Mapping[str, Any],
    task_key: str,
) -> Attempt:
    """Close one started attempt with Task 11 mutation evidence only."""
    with task_mutex(task_key):
        manifest = read_execution_manifest(path)
        index = next(
            (
                offset for offset, row in enumerate(manifest.attempts)
                if row.invocation_ref == invocation_ref and row.attempt == attempt
            ),
            None,
        )
        if index is None:
            raise ExecutionManifestError("attempt to finish is unknown")
        existing = manifest.attempts[index]
        closed = Attempt(
            invocation_ref=invocation_ref,
            attempt=attempt,
            started_at=existing.started_at,
            finished_at=finished_at,
            status=status,
            result_path=result_path,
            error_path=error_path,
            change_summary=dict(change_summary),
            evidence_seal_ref=None,
        )
        if existing.finished_at is not None:
            if existing != closed:
                raise ExecutionManifestError("attempt terminal mutation result drift")
            return existing
        attempts = list(manifest.attempts)
        attempts[index] = closed
        updated = _replace_manifest(manifest, attempts=tuple(attempts))
        _write_manifest_over_existing(path, updated)
        return closed


def persist_static_execution_identity(
    *,
    run_manifest_path: Path,
    task_manifest_path: Path,
    input_snapshot_path: Path,
    manifest: ExecutionManifest,
    task_key: str,
    run_ref: str,
    dynamic_roles: Sequence[str],
) -> None:
    """Publish v2 snapshots and their task index under the existing task mutex."""
    payload = manifest.to_payload()
    validate_execution_manifest_payload(payload)
    with task_mutex(task_key):
        run_payload = _read_json_object(run_manifest_path)
        input_payload = _read_json_object(input_snapshot_path)
        input_identity = _input_identity_payload(manifest, dynamic_roles)
        _assert_input_identity_immutable(input_payload, input_identity)
        task_payload = _read_json_object(task_manifest_path)
        identity_surface_from_payload(task_payload, surface="task-manifest")
        _validated_existing_task_index(task_manifest_path, task_payload)

        run_payload.update(payload)
        input_payload.update(input_identity)

        _write_json_atomic(run_manifest_path, run_payload)
        _write_json_atomic(input_snapshot_path, input_payload)
        _rebuild_task_identity_index_unlocked(task_manifest_path, task_payload)
        _write_json_atomic(task_manifest_path, task_payload)


def rebuild_task_role_execution_index(
    task_manifest_path: Path,
    *,
    task_key: str,
) -> list[dict[str, str]]:
    with task_mutex(task_key):
        payload = _read_json_object(task_manifest_path)
        rows = _rebuild_task_identity_index_unlocked(task_manifest_path, payload)
        _write_json_atomic(task_manifest_path, payload)
    return rows


def validate_execution_manifest_payload(payload: Mapping[str, Any]) -> None:
    _require_v2_manifest_keys(payload)
    _validate_execution_manifest_schema(payload)
    participants = _indexed_rows(payload["participantAssignments"], "participantRef")
    roles = _indexed_rows(payload["roleExecutions"], "roleExecutionRef")
    invocations = _indexed_rows(payload["invocations"], "invocationRef")
    _validate_roles(payload, participants)
    _validate_invocations(participants, roles, invocations)
    _validate_attempts(payload["attempts"], invocations)
    _validate_seals_and_recoveries(payload)


def _validate_execution_manifest_schema(payload: Mapping[str, Any]) -> None:
    try:
        errors = validate_schema(payload, _execution_manifest_schema())
    except SchemaError as exc:
        raise ExecutionManifestError(f"invalid execution manifest schema: {exc}") from exc
    if errors:
        raise ExecutionManifestError(errors[0])


@lru_cache(maxsize=1)
def _execution_manifest_schema() -> dict[str, Any]:
    here = Path(__file__).resolve()
    for parent in [here.parent, *here.parents]:
        candidate = parent / "schemas/execution-manifest-v2.schema.json"
        if candidate.is_file():
            try:
                return load_schema(candidate)
            except (OSError, json.JSONDecodeError) as exc:
                raise ExecutionManifestError(
                    f"cannot load execution manifest schema: {candidate}"
                ) from exc
    raise ExecutionManifestError("cannot locate schemas/execution-manifest-v2.schema.json")


def _require_v2_manifest_keys(payload: Mapping[str, Any]) -> None:
    if payload.get("schemaVersion") != V2_SCHEMA_VERSION:
        raise ExecutionManifestError("execution manifest schemaVersion must be 2.0")
    if payload.get("executionIdentityVersion") != EXECUTION_IDENTITY_VERSION:
        raise ExecutionManifestError("executionIdentityVersion must be 2")
    required = _IDENTITY_KEYS[2:]
    missing = [key for key in required if key not in payload]
    if missing:
        raise ExecutionManifestError(
            "execution manifest is missing required fields: " + ", ".join(missing)
        )
    for key in required[1:]:
        if not isinstance(payload[key], list):
            raise ExecutionManifestError(f"execution manifest {key} must be an array")


def _validate_roles(
    payload: Mapping[str, Any],
    participants: Mapping[str, Mapping[str, Any]],
) -> None:
    ordinals: dict[str, list[int]] = {}
    earlier_roles: dict[str, Mapping[str, Any]] = {}
    participant_roles: set[tuple[str, str]] = set()
    for row in payload["roleExecutions"]:
        role_ref = _required_text(row, "roleExecutionRef")
        participant_ref = _required_text(row, "participantRef")
        if participant_ref not in participants:
            raise ExecutionManifestError(
                f"unknown participantRef in role execution: {participant_ref}"
            )
        _validate_role_identity(payload, row, participants[participant_ref])
        role = _required_text(row, "role")
        participant_role = (participant_ref, role)
        if participant_role in participant_roles:
            raise ExecutionManifestError(
                "role execution participantRef and role must be unique"
            )
        participant_roles.add(participant_role)
        ordinal = row.get("ordinal")
        if not isinstance(ordinal, int) or ordinal < 1:
            raise ExecutionManifestError("role execution ordinal must be positive")
        ordinals.setdefault(role, []).append(ordinal)
        source = row.get("sourceRoleExecutionRef")
        if source is not None:
            source_role = earlier_roles.get(source)
            if source_role is None:
                raise ExecutionManifestError(
                    "sourceRoleExecutionRef must reference an earlier different "
                    "role execution"
                )
            if source_role.get("participantRef") != participant_ref:
                raise ExecutionManifestError(
                    "sourceRoleExecutionRef participant must match derived role "
                    "participant"
                )
        earlier_roles[role_ref] = row
    for role, values in ordinals.items():
        if sorted(values) != list(range(1, len(values) + 1)):
            raise ExecutionManifestError(f"role ordinals are not continuous: {role}")


def _validate_role_identity(
    payload: Mapping[str, Any],
    row: Mapping[str, Any],
    participant: Mapping[str, Any],
) -> None:
    role = _required_text(row, "role")
    provider = _required_text(row, "provider")
    model_id = _required_text(row, "modelId")
    unknown = row.get("modelRef") is None or row.get("modelSpecDigest") is None
    if unknown and not _is_current_session_unknown_leader(payload, row, participant):
        raise ExecutionManifestError("unknown model is allowed only for current-session leader")
    if not unknown and row.get("binding") is None:
        raise ExecutionManifestError("binding is required for a known model")
    expected_label = execution_label(role, provider, model_id, row.get("ordinal"))
    if row.get("executionLabel") != expected_label:
        raise ExecutionManifestError("executionLabel does not match structured identity")
    _validate_role_participant_match(payload, row, participant)
    source = row.get("sourceRoleExecutionRef")
    if source is not None and not isinstance(source, str):
        raise ExecutionManifestError("sourceRoleExecutionRef must be a string or null")


def _validate_role_participant_match(
    payload: Mapping[str, Any],
    row: Mapping[str, Any],
    participant: Mapping[str, Any],
) -> None:
    for key in ("provider", "modelRef", "modelId"):
        if row.get(key) != participant.get(key):
            raise ExecutionManifestError(
                f"role execution {key} does not match participant assignment"
            )
    binding = row.get("binding")
    if binding is None:
        if participant.get("hostModelValue") is not None:
            raise ExecutionManifestError(
                "role execution hostModelValue does not match participant"
            )
        _validate_participant_host_runtime(payload, participant)
        return
    if not isinstance(binding, Mapping):
        raise ExecutionManifestError("role execution binding must be an object or null")
    if binding.get("runner") != participant.get("runner"):
        raise ExecutionManifestError("role execution runner does not match participant")
    if binding.get("workerWriteCapability") != participant.get("workerWriteCapability"):
        raise ExecutionManifestError(
            "role execution workerWriteCapability does not match participant"
        )
    if binding.get("hostModelValue") != participant.get("hostModelValue"):
        raise ExecutionManifestError(
            "role execution hostModelValue does not match participant"
        )
    _validate_participant_host_runtime(payload, participant)


def _validate_participant_host_runtime(
    payload: Mapping[str, Any],
    participant: Mapping[str, Any],
) -> None:
    run_host = payload.get("leadRuntime")
    if run_host is not None and participant.get("hostRuntime") != run_host:
        raise ExecutionManifestError(
            "participant hostRuntime does not match run host"
        )


def _is_current_session_unknown_leader(
    payload: Mapping[str, Any],
    row: Mapping[str, Any],
    participant: Mapping[str, Any],
) -> bool:
    attestation = row.get("servedModelAttestation")
    return (
        payload.get("entryMode") == "current-session"
        and row.get("role") == "leader"
        and row.get("modelRef") is None
        and row.get("modelId") == "unknown"
        and row.get("modelSpecDigest") is None
        and row.get("binding") is None
        and isinstance(attestation, Mapping)
        and attestation.get("level") == "unknown"
        and participant.get("entryMode") == "current-session"
        and participant.get("runner") == "current-session"
        and participant.get("workerWriteCapability") is None
    )


def _validate_invocations(
    participants: Mapping[str, Mapping[str, Any]],
    roles: Mapping[str, Mapping[str, Any]],
    invocations: Mapping[str, Mapping[str, Any]],
) -> None:
    for row in invocations.values():
        participant_ref = _required_text(row, "participantRef")
        role_ref = _required_text(row, "roleExecutionRef")
        if participant_ref not in participants or role_ref not in roles:
            raise ExecutionManifestError("invocation references unknown execution identity")
        if roles[role_ref].get("participantRef") != participant_ref:
            raise ExecutionManifestError("invocation participant does not own role execution")
        source = row.get("sourceInvocationRef")
        if source is not None and source not in invocations:
            raise ExecutionManifestError("invocation sourceInvocationRef is unknown")
        _validate_invocation_write_contract(row, participants[participant_ref], roles[role_ref])


def _validate_invocation_write_contract(
    row: Mapping[str, Any],
    participant: Mapping[str, Any],
    role: Mapping[str, Any],
) -> None:
    policy = row.get("writePolicy")
    enforcement = row.get("writeEnforcement")
    if not isinstance(policy, Mapping) or not isinstance(enforcement, Mapping):
        raise ExecutionManifestError("invocation write contract must be structured")
    try:
        validate_write_policy_payload(policy)
        validate_write_enforcement_payload(enforcement)
        expected_digest = write_policy_digest(policy)
    except WritePolicyError as exc:
        raise ExecutionManifestError(f"invocation write contract: {exc}") from exc
    if row.get("writePolicyDigest") != expected_digest:
        raise ExecutionManifestError("invocation writePolicyDigest does not match")
    source = policy.get("sourcePolicy")
    expected_mode = (
        "project-mutation" if role.get("role") == "implementer" else "source-readonly"
    )
    if not isinstance(source, Mapping) or source.get("mode") != expected_mode:
        raise ExecutionManifestError("invocation sourcePolicy does not match role")
    git_policy = policy.get("gitPolicy")
    expected_git_mode = (
        "fast-forward-descendant-chain"
        if expected_mode == "project-mutation"
        else "disabled"
    )
    if (
        not isinstance(git_policy, Mapping)
        or git_policy.get("mode") != expected_git_mode
    ):
        raise ExecutionManifestError("invocation gitPolicy does not match role")
    if expected_mode == "source-readonly" and source.get("plannedPaths"):
        raise ExecutionManifestError(
            "source-readonly invocation cannot declare planned paths"
        )
    _validate_enforcement_capability(enforcement, participant, role)


def _validate_enforcement_capability(
    enforcement: Mapping[str, Any],
    participant: Mapping[str, Any],
    role: Mapping[str, Any],
) -> None:
    capability = participant.get("workerWriteCapability")
    if role.get("role") == "leader" and capability is None:
        if enforcement.get("boundaryPrecision") != "none":
            raise ExecutionManifestError(
                "leader writeEnforcement must use boundaryPrecision none"
            )
        if enforcement.get("mutationAudit") != "none":
            raise ExecutionManifestError(
                "leader writeEnforcement must use mutationAudit none"
            )
        return
    if not isinstance(capability, Mapping):
        raise ExecutionManifestError("worker invocation has no write capability")
    rank = {"none": 0, "directory-boundary": 1, "exact-path": 2}
    maximum = capability.get("maxBoundaryPrecision")
    actual = enforcement.get("boundaryPrecision")
    if maximum not in rank or actual not in rank or rank[actual] > rank[maximum]:
        raise ExecutionManifestError("writeEnforcement exceeds worker capability")


def _validate_attempts(
    rows: object,
    invocations: Mapping[str, Mapping[str, Any]],
) -> None:
    if not isinstance(rows, list):
        raise ExecutionManifestError("execution manifest attempts must be an array")
    attempts_by_invocation: dict[str, list[int]] = {}
    for row in rows:
        if not isinstance(row, Mapping):
            raise ExecutionManifestError("attempt row must be an object")
        invocation_ref = _required_text(row, "invocationRef")
        if invocation_ref not in invocations:
            raise ExecutionManifestError("attempt references unknown invocation")
        attempt = row.get("attempt")
        if not isinstance(attempt, int) or attempt < 1:
            raise ExecutionManifestError("attempt number must be positive")
        attempts_by_invocation.setdefault(invocation_ref, []).append(attempt)
    for invocation_ref, values in attempts_by_invocation.items():
        if sorted(values) != list(range(1, len(values) + 1)):
            raise ExecutionManifestError(
                f"attempt numbers are not continuous: {invocation_ref}"
            )


def _validate_seals_and_recoveries(payload: Mapping[str, Any]) -> None:
    attempts = payload.get("attempts")
    if not isinstance(attempts, list):
        return
    from .dispatch_state import build_dispatch_id

    attempt_refs = {
        build_dispatch_id(str(row.get("invocationRef")), int(row.get("attempt") or 0))
        for row in attempts
        if isinstance(row, Mapping)
    }
    seals = payload.get("attemptEvidenceSeals") or []
    if not isinstance(seals, list):
        raise ExecutionManifestError("attemptEvidenceSeals must be an array")
    seen_attempts: set[str] = set()
    seen_refs: set[str] = set()
    for row in seals:
        if not isinstance(row, Mapping):
            raise ExecutionManifestError("evidence seal must be an object")
        seal_ref = str(row.get("sealRef") or "")
        attempt_ref = str(row.get("attemptRef") or "")
        if not seal_ref or seal_ref in seen_refs:
            raise ExecutionManifestError("evidence sealRef is missing or duplicated")
        if attempt_ref not in attempt_refs or attempt_ref in seen_attempts:
            raise ExecutionManifestError("evidence seal does not uniquely reference an attempt")
        seen_refs.add(seal_ref)
        seen_attempts.add(attempt_ref)
    seal_by_ref = {str(row.get("sealRef")): row for row in seals if isinstance(row, Mapping)}
    for row in attempts:
        if not isinstance(row, Mapping):
            continue
        evidence = row.get("evidenceSealRef")
        if evidence in (None, ""):
            continue
        if evidence not in seal_by_ref:
            raise ExecutionManifestError("attempt evidenceSealRef is unknown")
    recoveries = payload.get("mutationRecoveryDecisions") or []
    if not isinstance(recoveries, list):
        raise ExecutionManifestError("mutationRecoveryDecisions must be an array")


def _validate_next_role_execution(
    manifest: ExecutionManifest,
    row: RoleExecution,
) -> None:
    if row.participant_ref not in {
        item.participant_ref for item in manifest.participant_assignments
    }:
        raise ExecutionManifestError("role execution references unknown participant")
    existing = [item.ordinal for item in manifest.role_executions if item.role == row.role]
    expected = max(existing, default=0) + 1
    if row.ordinal != expected:
        raise ExecutionManifestError(f"next ordinal must be {expected}")
    if row.role_execution_ref in {
        item.role_execution_ref for item in manifest.role_executions
    }:
        raise ExecutionManifestError("roleExecutionRef is duplicated")


def _derived_role_execution(
    manifest: ExecutionManifest,
    source: RoleExecution,
    role: str,
) -> tuple[RoleExecution, bool]:
    if source.role == role:
        return source, False
    existing = [
        item
        for item in manifest.role_executions
        if item.participant_ref == source.participant_ref and item.role == role
    ]
    if len(existing) > 1:
        raise ExecutionManifestError(
            f"participant has multiple {role} role executions: {source.participant_ref}"
        )
    if existing:
        return existing[0], False
    ordinal = max(
        (item.ordinal for item in manifest.role_executions if item.role == role),
        default=0,
    ) + 1
    role_ref = f"role-exec-{role}-{ordinal:03d}"
    if role_ref in {
        item.role_execution_ref for item in manifest.role_executions
    }:
        raise ExecutionManifestError(
            f"derived roleExecutionRef is already reserved: {role_ref}"
        )
    return RoleExecution(
        role_execution_ref=role_ref,
        participant_ref=source.participant_ref,
        source_role_execution_ref=source.role_execution_ref,
        role=role,
        provider=source.provider,
        model_ref=source.model_ref,
        model_id=source.model_id,
        ordinal=ordinal,
        execution_label=execution_label(
            role,
            source.provider,
            source.model_id,
            ordinal,
        ),
        model_spec_digest=source.model_spec_digest,
        binding=source.binding,
        served_model_attestation=ServedModelAttestation.unknown(),
        status="prepared",
    ), True


def _validate_next_invocation(manifest: ExecutionManifest, row: Invocation) -> None:
    role = next(
        (
            item
            for item in manifest.role_executions
            if item.role_execution_ref == row.role_execution_ref
        ),
        None,
    )
    if role is None or role.participant_ref != row.participant_ref:
        raise ExecutionManifestError("invocation references unknown role execution")
    if row.invocation_ref in {item.invocation_ref for item in manifest.invocations}:
        raise ExecutionManifestError("invocationRef is duplicated")
    if row.source_invocation_ref is not None and row.source_invocation_ref not in {
        item.invocation_ref for item in manifest.invocations
    }:
        raise ExecutionManifestError("sourceInvocationRef is unknown")


def _validate_next_attempt(manifest: ExecutionManifest, row: Attempt) -> None:
    if row.invocation_ref not in {
        item.invocation_ref for item in manifest.invocations
    }:
        raise ExecutionManifestError("attempt references unknown invocation")
    prior = [
        item
        for item in manifest.attempts
        if item.invocation_ref == row.invocation_ref
    ]
    expected = max((item.attempt for item in prior), default=0) + 1
    if row.attempt != expected:
        raise ExecutionManifestError(f"next attempt must be {expected}")
    if not prior:
        return
    last = max(prior, key=lambda item: item.attempt)
    if last.finished_at is None:
        raise ExecutionManifestError("previous attempt is not terminal")
    if last.status != "failed-no-mutation":
        raise ExecutionManifestError(
            "invocation cannot append another attempt after a terminal mutation"
        )


def _execution_manifest_from_payload(payload: Mapping[str, Any]) -> ExecutionManifest:
    return ExecutionManifest(
        entry_mode=str(payload["entryMode"]),
        participant_assignments=tuple(
            _participant_from_payload(row) for row in payload["participantAssignments"]
        ),
        role_executions=tuple(
            _role_from_payload(row) for row in payload["roleExecutions"]
        ),
        invocations=tuple(_invocation_from_payload(row) for row in payload["invocations"]),
        attempts=tuple(_attempt_from_payload(row) for row in payload["attempts"]),
        attempt_evidence_seals=tuple(payload["attemptEvidenceSeals"]),
        mutation_recovery_decisions=tuple(payload["mutationRecoveryDecisions"]),
    )


def _participant_from_payload(row: Mapping[str, Any]) -> ParticipantAssignment:
    return ParticipantAssignment(
        participant_ref=str(row["participantRef"]),
        provider=str(row["provider"]),
        model_ref=_optional_text(row.get("modelRef")),
        model_id=str(row["modelId"]),
        runner=str(row["runner"]),
        host_runtime=str(row["hostRuntime"]),
        host_model_value=_optional_text(row.get("hostModelValue")),
        session_ref=_optional_text(row.get("sessionRef")),
        window_ref=_optional_text(row.get("windowRef")),
        worker_write_capability=_capability_from_payload(
            row.get("workerWriteCapability")
        ),
        entry_mode=str(row["entryMode"]),
        status=str(row["status"]),
    )


def _role_from_payload(row: Mapping[str, Any]) -> RoleExecution:
    return RoleExecution(
        role_execution_ref=str(row["roleExecutionRef"]),
        participant_ref=str(row["participantRef"]),
        source_role_execution_ref=_optional_text(row.get("sourceRoleExecutionRef")),
        role=str(row["role"]),
        provider=str(row["provider"]),
        model_ref=_optional_text(row.get("modelRef")),
        model_id=str(row["modelId"]),
        ordinal=int(row["ordinal"]),
        execution_label=str(row["executionLabel"]),
        model_spec_digest=_optional_text(row.get("modelSpecDigest")),
        binding=_binding_from_payload(row.get("binding")),
        served_model_attestation=_attestation_from_payload(
            row["servedModelAttestation"]
        ),
        status=str(row["status"]),
    )


def _invocation_from_payload(row: Mapping[str, Any]) -> Invocation:
    return Invocation(
        invocation_ref=str(row["invocationRef"]),
        participant_ref=str(row["participantRef"]),
        role_execution_ref=str(row["roleExecutionRef"]),
        source_invocation_ref=_optional_text(row.get("sourceInvocationRef")),
        recovery_ref=_optional_text(row.get("recoveryRef")),
        duty_id=str(row["dutyId"]),
        dispatch_kind=str(row["dispatchKind"]),
        round=int(row["round"]),
        input_digest=str(row["inputDigest"]),
        write_policy=dict(row["writePolicy"]),
        write_policy_digest=str(row["writePolicyDigest"]),
        write_enforcement=dict(row["writeEnforcement"]),
    )


def _attempt_from_payload(row: Mapping[str, Any]) -> Attempt:
    return Attempt(
        invocation_ref=str(row["invocationRef"]),
        attempt=int(row["attempt"]),
        started_at=str(row["startedAt"]),
        finished_at=_optional_text(row.get("finishedAt")),
        status=str(row["status"]),
        result_path=_optional_text(row.get("resultPath")),
        error_path=_optional_text(row.get("errorPath")),
        change_summary=dict(row["changeSummary"]),
        evidence_seal_ref=_optional_text(row.get("evidenceSealRef")),
    )


def _binding_from_payload(value: object) -> HostModelBinding | None:
    if value is None:
        return None
    if not isinstance(value, Mapping):
        raise ExecutionManifestError("binding must be an object or null")
    return HostModelBinding(
        runner=str(value["runner"]),
        catalog_execution_value=str(value["catalogExecutionValue"]),
        resolved_execution_value=str(value["resolvedExecutionValue"]),
        host_model_value=_optional_text(value.get("hostModelValue")),
        binding_fidelity=str(value["bindingFidelity"]),
        worker_write_capability=_capability_from_payload(
            value.get("workerWriteCapability")
        ),
    )


def _attestation_from_payload(value: object) -> ServedModelAttestation:
    if not isinstance(value, Mapping):
        raise ExecutionManifestError("servedModelAttestation must be an object")
    return ServedModelAttestation(
        observed_model=_optional_text(value.get("observedModel")),
        normalized_model_ref=_optional_text(value.get("normalizedModelRef")),
        level=str(value["level"]),
        source=str(value["source"]),
    )


def _capability_from_payload(value: object) -> WorkerWriteCapability | None:
    if value is None:
        return None
    if not isinstance(value, Mapping):
        raise ExecutionManifestError("workerWriteCapability must be an object or null")
    return WorkerWriteCapability(str(value["maxBoundaryPrecision"]))


def _replace_manifest(manifest: ExecutionManifest, **changes: Any) -> ExecutionManifest:
    values = {
        "entry_mode": manifest.entry_mode,
        "participant_assignments": manifest.participant_assignments,
        "role_executions": manifest.role_executions,
        "invocations": manifest.invocations,
        "attempts": manifest.attempts,
        "attempt_evidence_seals": manifest.attempt_evidence_seals,
        "mutation_recovery_decisions": manifest.mutation_recovery_decisions,
    }
    values.update(changes)
    return ExecutionManifest(**values)


def _write_manifest_over_existing(path: Path, manifest: ExecutionManifest) -> None:
    payload = manifest.to_payload()
    validate_execution_manifest_payload(payload)
    existing = _read_json_object(path)
    existing.update(payload)
    _write_json_atomic(path, existing)


def _input_identity_payload(
    manifest: ExecutionManifest,
    dynamic_roles: Sequence[str],
) -> dict[str, Any]:
    return {
        "schemaVersion": V2_SCHEMA_VERSION,
        "executionIdentityVersion": EXECUTION_IDENTITY_VERSION,
        "staticParticipantAssignments": [
            row.to_payload() for row in manifest.participant_assignments
        ],
        "staticRoleExecutions": [row.to_payload() for row in manifest.role_executions],
        "dynamicRolePolicy": {
            "roles": list(dict.fromkeys(dynamic_roles)),
            "appendOnly": True,
        },
    }


def _assert_input_identity_immutable(
    existing: Mapping[str, Any],
    expected: Mapping[str, Any],
) -> None:
    if existing.get("schemaVersion") != V2_SCHEMA_VERSION:
        return
    keys = (
        "executionIdentityVersion",
        "staticParticipantAssignments",
        "staticRoleExecutions",
        "dynamicRolePolicy",
    )
    if any(existing.get(key) != expected.get(key) for key in keys):
        raise ExecutionManifestError("input snapshot is immutable after v2 creation")


def _rebuild_task_identity_index_unlocked(
    task_manifest_path: Path,
    payload: dict[str, Any],
) -> list[dict[str, str]]:
    existing = _validated_existing_task_index(task_manifest_path, payload)
    task_root = task_manifest_path.parent
    reference_root = _index_reference_root(task_root, payload)
    discovered = _role_execution_index_from_runs(task_root, reference_root)
    rows = list(existing)
    known = {(row["runRef"], row["roleExecutionRef"]) for row in rows}
    for row in discovered:
        pair = (row["runRef"], row["roleExecutionRef"])
        if pair not in known:
            rows.append(row)
            known.add(pair)
    payload["schemaVersion"] = V2_SCHEMA_VERSION
    payload["executionIdentityVersion"] = EXECUTION_IDENTITY_VERSION
    payload["roleExecutionIndex"] = rows
    validate_task_role_execution_index(payload)
    return rows


def _validated_existing_task_index(
    task_manifest_path: Path,
    payload: Mapping[str, Any],
) -> list[dict[str, str]]:
    surface = identity_surface_from_payload(payload, surface="task-manifest")
    if surface.legacy:
        return []
    rows = [dict(row) for row in payload["roleExecutionIndex"]]
    task_root = task_manifest_path.parent
    reference_root = _index_reference_root(task_root, payload)
    roles_by_run: dict[str, set[str]] = {}
    for row in rows:
        run_ref = row["runRef"]
        role_ref = row["roleExecutionRef"]
        if run_ref not in roles_by_run:
            roles_by_run[run_ref] = _referenced_role_refs(
                task_root,
                reference_root,
                run_ref,
            )
        if role_ref not in roles_by_run[run_ref]:
            raise ExecutionManifestError(
                "roleExecutionIndex disagrees with referenced run manifest: "
                f"{run_ref}, {role_ref}"
            )
    return rows


def _referenced_role_refs(
    task_root: Path,
    reference_root: Path,
    run_ref: str,
) -> set[str]:
    path = (reference_root / run_ref).resolve()
    if not path.is_relative_to(task_root.resolve()) or not path.is_file():
        raise ExecutionManifestError(
            f"roleExecutionIndex referenced run manifest is missing: {run_ref}"
        )
    try:
        manifest = read_execution_manifest(path)
    except ExecutionManifestError as exc:
        raise ExecutionManifestError(
            f"roleExecutionIndex referenced run manifest is corrupt: {run_ref}"
        ) from exc
    if manifest.legacy:
        raise ExecutionManifestError(
            f"roleExecutionIndex referenced run manifest is corrupt: {run_ref}"
        )
    return {row.role_execution_ref for row in manifest.role_executions}


def _validate_run_ref_for_manifest(
    path: Path,
    task_manifest_path: Path,
    task_payload: Mapping[str, Any],
    run_ref: str,
) -> None:
    task_root = task_manifest_path.parent.resolve()
    manifest_path = path.resolve()
    reference_root = _index_reference_root(task_root, task_payload).resolve()
    try:
        expected_ref = manifest_path.relative_to(reference_root).as_posix()
    except ValueError:
        expected_ref = ""
    if (
        not manifest_path.is_relative_to(task_root)
        or not expected_ref
        or run_ref != expected_ref
    ):
        raise ExecutionManifestError(
            "runRef does not identify execution manifest path"
        )


def _validate_new_task_index_row(
    manifest: ExecutionManifest,
    row: RoleExecution,
) -> None:
    payload = manifest.to_payload()
    validate_execution_manifest_payload(payload)
    role_refs = {
        item["roleExecutionRef"]
        for item in payload["roleExecutions"]
    }
    if row.role_execution_ref not in role_refs:
        raise ExecutionManifestError(
            "new roleExecutionIndex row is absent from updated run manifest"
        )


def _append_task_index_unlocked(
    path: Path,
    run_ref: str,
    row: RoleExecution,
    payload: dict[str, Any],
) -> None:
    identity_surface_from_payload(payload, surface="task-manifest")
    updated = _task_payload_with_role_index(payload, run_ref, row)
    _write_json_atomic(path, updated)


def _task_payload_with_role_index(
    payload: Mapping[str, Any],
    run_ref: str,
    row: RoleExecution,
) -> dict[str, Any]:
    updated = dict(payload)
    index = [dict(item) for item in payload["roleExecutionIndex"]]
    item = {"runRef": run_ref, "roleExecutionRef": row.role_execution_ref}
    if item not in index:
        index.append(item)
    updated["schemaVersion"] = V2_SCHEMA_VERSION
    updated["executionIdentityVersion"] = EXECUTION_IDENTITY_VERSION
    updated["roleExecutionIndex"] = index
    validate_task_role_execution_index(updated)
    return updated


def _write_manifest_and_task_index(
    manifest_path: Path,
    manifest: ExecutionManifest,
    task_manifest_path: Path,
    task_payload: Mapping[str, Any],
) -> None:
    manifest_payload = _read_json_object(manifest_path)
    manifest_payload.update(manifest.to_payload())
    _write_json_atomic(manifest_path, manifest_payload)
    _write_json_atomic(task_manifest_path, task_payload)


def _index_reference_root(
    task_root: Path,
    task_manifest: Mapping[str, Any],
) -> Path:
    value = task_manifest.get("projectRoot")
    if not isinstance(value, str) or not value.strip():
        return task_root.resolve()
    project_root = Path(value).resolve()
    if not task_root.resolve().is_relative_to(project_root):
        raise ExecutionManifestError(
            "task root resolves outside project root"
        )
    return project_root


def _role_execution_index_from_runs(
    task_root: Path,
    reference_root: Path,
) -> list[dict[str, str]]:
    rows: list[dict[str, str]] = []
    resolved_task_root = task_root.resolve()
    resolved_reference_root = reference_root.resolve()
    for path in sorted(task_root.glob("runs/**/manifests/run-manifest-*.json")):
        resolved_path = path.resolve()
        if (
            not resolved_path.is_relative_to(resolved_task_root)
            or not resolved_path.is_relative_to(resolved_reference_root)
        ):
            raise ExecutionManifestError(
                "discovered run manifest resolves outside task or project root"
            )
        surface = read_identity_surface(resolved_path, surface="run-manifest")
        if surface.legacy:
            continue
        validate_execution_manifest_payload(surface.payload)
        run_ref = resolved_path.relative_to(resolved_reference_root).as_posix()
        rows.extend(
            {
                "runRef": run_ref,
                "roleExecutionRef": str(role["roleExecutionRef"]),
            }
            for role in surface.payload["roleExecutions"]
        )
    return rows


def _indexed_rows(rows: object, key: str) -> dict[str, Mapping[str, Any]]:
    if not isinstance(rows, list):
        raise ExecutionManifestError(f"execution manifest {key} rows must be an array")
    indexed: dict[str, Mapping[str, Any]] = {}
    for row in rows:
        if not isinstance(row, Mapping):
            raise ExecutionManifestError(f"execution manifest {key} row must be an object")
        value = _required_text(row, key)
        if value in indexed:
            raise ExecutionManifestError(f"{key} is duplicated: {value}")
        indexed[value] = row
    return indexed


def _required_text(row: Mapping[str, Any], key: str) -> str:
    value = row.get(key)
    if not isinstance(value, str) or not value.strip():
        raise ExecutionManifestError(f"{key} must be a non-empty string")
    return value


def _read_json_object(path: Path) -> dict[str, Any]:
    try:
        payload = load_owned_object(path, artifact="execution manifest")
    except (OSError, JsonBoundaryError) as exc:
        raise ExecutionManifestError(f"cannot read JSON object: {path}") from exc
    if not isinstance(payload, dict):
        raise ExecutionManifestError(f"JSON surface must be an object: {path}")
    return payload


def _write_json_atomic(
    path: Path,
    payload: Mapping[str, Any],
    *,
    create_only: bool = False,
) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    serialized = serialize_owned_object(
        path, payload, artifact="execution manifest"
    )
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{path.name}.", suffix=".tmp", dir=path.parent
    )
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "w", encoding="utf-8") as stream:
            stream.write(serialized)
            stream.flush()
            os.fsync(stream.fileno())
        if create_only:
            try:
                os.link(temporary, path)
            except FileExistsError as exc:
                raise ExecutionManifestError(
                    f"execution manifest already exists: {path}"
                ) from exc
        else:
            os.replace(temporary, path)
    finally:
        temporary.unlink(missing_ok=True)


def _optional_text(value: object) -> str | None:
    text = str(value or "").strip()
    return text or None