iop/scripts/agent_benchmark/attempts.py

"""Durable, append-only benchmark run and attempt state.

Attempt-directory creation is the allocation marker.  The directory remains
empty until the workspace layer has prepared it; only then is the identity
bound running record published.  This preserves the workspace API's
empty-root contract while retaining a crash-recoverable allocation boundary.
"""

from __future__ import annotations

import contextlib
import datetime as _datetime
import fcntl
import hashlib
import json
import os
import re
import secrets
import stat
import tempfile
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Callable, Iterator

from scripts.agent_benchmark.lifecycle import (
    COMPLETION_MODES,
    EVENT_FINISH,
    EVENT_IDLE,
    EVENT_QUIET,
    EVENT_SUBMITTED,
    InvocationResult,
    LifecycleRecoveryError,
    RECEIPT_VERSION,
    SUBMISSION_MODES,
    SupervisorLocator,
    TERMINAL_REASONS,
    recover_invocation,
)
from scripts.agent_benchmark.manifest import Manifest, validate_manifest_bytes
from scripts.agent_benchmark.workspace import AttemptIdentity

RUN_ID_RE = re.compile(r"^run-[0-9]{8}T[0-9]{6}Z-[0-9a-f]{12}$")
ATTEMPT_RE = re.compile(r"^attempt-([0-9]{6})$")
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
TERMINAL_STATES = frozenset(("success", "failed", "timed_out", "cancelled", "interrupted"))
NONTERMINAL_STATE = "running"
SUCCESS_EVIDENCE_KINDS = (EVENT_SUBMITTED, EVENT_FINISH, EVENT_IDLE, EVENT_QUIET)
RECEIPT_FIELDS = {
    "receipt_version", "supervisor_pid", "challenge_digest", "reason", "exit_code",
    "signal", "caller_launched", "cleanup_complete", "process_group_alive", "completed_at",
}


class AttemptError(Exception):
    """Base error whose message is safe to present to a benchmark caller."""


class RunPathError(AttemptError):
    pass


class RunBusyError(AttemptError):
    pass


class AttemptStateError(AttemptError):
    pass


class CapabilityUnavailable(AttemptError):
    pass


@dataclass(frozen=True)
class RunIdentity:
    run_id: str
    manifest_digest: str
    root: str


@dataclass(frozen=True)
class Slot:
    cell_id: str
    repetition: int


@dataclass(frozen=True)
class Attempt:
    identity: AttemptIdentity
    root: str
    state: str


def _json_bytes(value: Any) -> bytes:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") + b"\n"


def _fsync_dir(path: Path) -> None:
    fd = os.open(path, os.O_RDONLY)
    try:
        os.fsync(fd)
    finally:
        os.close(fd)


def _write_new(path: Path, data: bytes) -> None:
    fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        os.write(fd, data)
        os.fsync(fd)
    finally:
        os.close(fd)
    _fsync_dir(path.parent)


def _replace(path: Path, data: bytes) -> None:
    fd, tmp = tempfile.mkstemp(prefix=".attempt-", dir=path.parent)
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(tmp, 0o600)
        os.replace(tmp, path)
        _fsync_dir(path.parent)
    finally:
        try:
            os.unlink(tmp)
        except FileNotFoundError:
            pass


def _open_regular(path: Path, label: str, flags: int = os.O_RDONLY) -> int:
    """Open one durable file without following links, blocking or trusting its type."""
    try:
        fd = os.open(path, flags | os.O_NOFOLLOW | os.O_NONBLOCK)
    except OSError as exc:
        raise AttemptStateError(f"{label} is unavailable") from exc
    try:
        opened = os.fstat(fd)
        if not stat.S_ISREG(opened.st_mode):
            raise AttemptStateError(f"{label} must be a regular file")
    except BaseException:
        os.close(fd)
        raise
    return fd


def _read_regular_bytes(path: Path, label: str) -> bytes:
    """Read only the no-follow descriptor that was verified as regular."""
    fd = _open_regular(path, label)
    try:
        chunks: list[bytes] = []
        while True:
            chunk = os.read(fd, 1 << 20)
            if not chunk:
                return b"".join(chunks)
            chunks.append(chunk)
    finally:
        os.close(fd)


def _utc_timestamp(clock: Callable[[], _datetime.datetime]) -> str:
    return clock().astimezone(_datetime.timezone.utc).strftime("%Y%m%dT%H%M%SZ")


def _directory(path: Path, label: str) -> None:
    try:
        mode = os.lstat(path).st_mode
    except OSError as exc:
        raise RunPathError(f"{label} is unavailable") from exc
    if not stat.S_ISDIR(mode):
        raise RunPathError(f"{label} is invalid")


def _contained(path: Path, root: Path) -> bool:
    try:
        path.resolve(strict=False).relative_to(root.resolve())
        return True
    except ValueError:
        return False


class RunStore:
    """Filesystem-backed run store rooted at a validated manifest output root."""

    def __init__(
        self,
        repo_root: str | Path,
        *,
        clock: Callable[[], _datetime.datetime] = lambda: _datetime.datetime.now(_datetime.timezone.utc),
        token_hex: Callable[[int], str] = secrets.token_hex,
    ) -> None:
        self.repo_root = Path(repo_root).resolve()
        self._clock = clock
        self._token_hex = token_hex

    def _output_root(self, manifest: Manifest, *, create: bool) -> Path:
        root = self.repo_root / manifest.output_root
        if not _contained(root, self.repo_root):
            raise RunPathError("output root is invalid")
        if create:
            root.mkdir(mode=0o700, parents=True, exist_ok=True)
        _directory(root, "output root")
        if root.is_symlink():
            raise RunPathError("output root is invalid")
        return root.resolve()

    def _run_path(self, manifest: Manifest, run_id: str, *, create_root: bool) -> Path:
        if not RUN_ID_RE.fullmatch(run_id):
            raise RunPathError("run id is invalid")
        root = self._output_root(manifest, create=create_root)
        path = root / run_id
        if path.parent != root or path.is_symlink():
            raise RunPathError("run path is invalid")
        return path

    def create(self, manifest: Manifest, manifest_bytes: bytes) -> RunIdentity:
        """Create a run and atomically persist immutable source bytes first."""
        loaded = validate_manifest_bytes(manifest_bytes, repo_root=self.repo_root)
        if loaded.digest != manifest.digest:
            raise AttemptStateError("manifest snapshot does not match manifest digest")
        run_id = f"run-{_utc_timestamp(self._clock)}-{self._token_hex(6)}"
        if not RUN_ID_RE.fullmatch(run_id):
            raise AttemptStateError("generated run id is invalid")
        path = self._run_path(manifest, run_id, create_root=True)
        try:
            path.mkdir(mode=0o700)
        except FileExistsError as exc:
            raise AttemptStateError("generated run id already exists") from exc
        _write_new(path / "manifest.json", manifest_bytes)
        _write_new(path / "run.json", _json_bytes({"run_id": run_id, "manifest_digest": manifest.digest}))
        _write_new(path / "run.lock", b"")
        _fsync_dir(path)
        return RunIdentity(run_id, manifest.digest, str(path))

    def open(self, manifest: Manifest, run_id: str, manifest_bytes: bytes | None = None) -> RunIdentity:
        path = self._run_path(manifest, run_id, create_root=False)
        _directory(path, "run")
        try:
            record = json.loads(_read_regular_bytes(path / "run.json", "run record").decode("utf-8"))
            snapshot = _read_regular_bytes(path / "manifest.json", "manifest snapshot")
            _read_regular_bytes(path / "run.lock", "run lock")
        except (OSError, json.JSONDecodeError) as exc:
            raise AttemptStateError("run state is unavailable") from exc
        if record != {"run_id": run_id, "manifest_digest": manifest.digest}:
            raise AttemptStateError("run identity does not match manifest")
        if manifest_bytes is not None and snapshot != manifest_bytes:
            raise AttemptStateError("manifest bytes do not match run snapshot")
        if validate_manifest_bytes(snapshot, repo_root=self.repo_root).digest != manifest.digest:
            raise AttemptStateError("run manifest digest is invalid")
        return RunIdentity(run_id, manifest.digest, str(path))

    @contextlib.contextmanager
    def writer(self, run: RunIdentity) -> Iterator[None]:
        root = Path(run.root)
        _directory(root, "run")
        lock_path = root / "run.lock"
        fd = _open_regular(lock_path, "run lock", os.O_RDWR)
        try:
            try:
                fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
            except BlockingIOError as exc:
                raise RunBusyError("run-busy") from exc
            yield
        finally:
            try:
                fcntl.flock(fd, fcntl.LOCK_UN)
            finally:
                os.close(fd)

    @staticmethod
    def slots(manifest: Manifest) -> tuple[Slot, ...]:
        return tuple(Slot(cell.id, repetition) for cell in manifest.matrix for repetition in range(1, manifest.repetitions + 1))

    def _slot_path(self, run: RunIdentity, slot: Slot) -> Path:
        if not re.fullmatch(r"[a-z0-9][a-z0-9_-]{0,63}", slot.cell_id) or slot.repetition < 1:
            raise AttemptStateError("slot is invalid")
        return Path(run.root) / "cells" / slot.cell_id / f"repetition-{slot.repetition:04d}"

    def _ensure_slot_parent(self, run: RunIdentity, slot: Slot) -> Path:
        current = Path(run.root)
        for segment in ("cells", slot.cell_id, f"repetition-{slot.repetition:04d}"):
            current = current / segment
            if current.exists() or current.is_symlink():
                _directory(current, "slot state")
                if current.is_symlink():
                    raise AttemptStateError("slot state is invalid")
                continue
            try:
                current.mkdir(mode=0o700)
            except FileExistsError:
                _directory(current, "slot state")
                if current.is_symlink():
                    raise AttemptStateError("slot state is invalid")
        return current

    @staticmethod
    def _expected_record(run: RunIdentity, identity: AttemptIdentity, state: str) -> dict[str, Any]:
        return {
            "run_id": run.run_id,
            "manifest_digest": run.manifest_digest,
            "cell_id": identity.cell_id,
            "repetition": identity.repetition,
            "attempt": identity.attempt,
            "state": state,
        }

    def _attempt_record(self, root: Path, run: RunIdentity, identity: AttemptIdentity, *, absent_ok: bool = False) -> dict[str, Any] | None:
        path = root / "attempt.json"
        try:
            record = json.loads(_read_regular_bytes(path, "attempt record").decode("utf-8"))
        except AttemptStateError:
            if absent_ok:
                try:
                    os.lstat(path)
                except FileNotFoundError:
                    return None
            raise
        except (OSError, json.JSONDecodeError) as exc:
            raise AttemptStateError("attempt record is unavailable") from exc
        if not isinstance(record, dict):
            raise AttemptStateError("attempt record is invalid")
        expected = self._expected_record(run, identity, str(record.get("state", "")))
        if record.get("state") not in TERMINAL_STATES | {NONTERMINAL_STATE}:
            raise AttemptStateError("attempt record is invalid")
        for key, value in expected.items():
            if record.get(key) != value:
                raise AttemptStateError("attempt record identity is invalid")
        allowed = set(expected) | {"locator", "spec_digest", "lifecycle"}
        if set(record) - allowed:
            raise AttemptStateError("attempt record schema is invalid")
        if record["state"] == NONTERMINAL_STATE and "lifecycle" in record:
            raise AttemptStateError("running attempt has terminal evidence")
        if ("locator" in record) != ("spec_digest" in record):
            raise AttemptStateError("attempt invocation identity is invalid")
        if "locator" in record:
            self._locator_from_record(root, record["locator"])
            if not isinstance(record["spec_digest"], str) or not DIGEST_RE.fullmatch(record["spec_digest"]):
                raise AttemptStateError("attempt invocation digest is invalid")
        if "lifecycle" in record and (not isinstance(record["lifecycle"], dict) or set(record["lifecycle"]) != {"terminal_reason"} or not isinstance(record["lifecycle"]["terminal_reason"], str)):
            raise AttemptStateError("attempt lifecycle is invalid")
        return record

    def attempts(self, run: RunIdentity, slot: Slot) -> tuple[Attempt, ...]:
        parent = self._slot_path(run, slot)
        if not parent.exists() and not parent.is_symlink():
            return ()
        _directory(parent, "slot state")
        found: list[Attempt] = []
        for child in sorted(parent.iterdir(), key=lambda item: item.name):
            match = ATTEMPT_RE.fullmatch(child.name)
            if not match or child.is_symlink() or not child.is_dir():
                raise AttemptStateError("slot contains invalid state")
            number = int(match.group(1))
            identity = AttemptIdentity(run.run_id, slot.cell_id, slot.repetition, number)
            record = self._attempt_record(child, run, identity, absent_ok=True)
            state = NONTERMINAL_STATE if record is None else str(record["state"])
            found.append(Attempt(identity, str(child), state))
        return tuple(found)

    def allocate(self, run: RunIdentity, slot: Slot) -> Attempt:
        """Create the exclusive, deliberately empty attempt root."""
        existing = self.attempts(run, slot)
        if any(item.state not in TERMINAL_STATES for item in existing):
            raise AttemptStateError("slot already has a nonterminal attempt")
        number = max((item.identity.attempt for item in existing), default=0) + 1
        parent = self._ensure_slot_parent(run, slot)
        root = parent / f"attempt-{number:06d}"
        try:
            root.mkdir(mode=0o700)
            _fsync_dir(parent)
        except FileExistsError as exc:
            raise AttemptStateError("attempt allocation collision") from exc
        return Attempt(AttemptIdentity(run.run_id, slot.cell_id, slot.repetition, number), str(root), NONTERMINAL_STATE)

    def _bound_attempt(self, attempt: Attempt) -> tuple[RunIdentity, Path]:
        root = Path(attempt.root)
        try:
            run_root = root.parents[3]
        except IndexError as exc:
            raise AttemptStateError("attempt path is invalid") from exc
        _directory(run_root, "run")
        try:
            run_record = json.loads(_read_regular_bytes(run_root / "run.json", "run record").decode("utf-8"))
            snapshot = _read_regular_bytes(run_root / "manifest.json", "manifest snapshot")
            _read_regular_bytes(run_root / "run.lock", "run lock")
            manifest = validate_manifest_bytes(snapshot, repo_root=self.repo_root)
        except Exception as exc:
            raise AttemptStateError("attempt run binding is invalid") from exc
        if not isinstance(run_record, dict) or set(run_record) != {"run_id", "manifest_digest"}:
            raise AttemptStateError("attempt run binding is invalid")
        run = RunIdentity(str(run_record["run_id"]), str(run_record["manifest_digest"]), str(run_root))
        if not RUN_ID_RE.fullmatch(run.run_id) or run.manifest_digest != manifest.digest:
            raise AttemptStateError("attempt run binding is invalid")
        if run_root.resolve() != self._run_path(manifest, run.run_id, create_root=False).resolve():
            raise AttemptStateError("attempt run binding is invalid")
        expected = self._slot_path(run, Slot(attempt.identity.cell_id, attempt.identity.repetition)) / f"attempt-{attempt.identity.attempt:06d}"
        if root.resolve() != expected.resolve() or attempt.identity.run_id != run.run_id or root.is_symlink():
            raise AttemptStateError("attempt identity is invalid")
        _directory(root, "attempt root")
        return run, root

    def _initial_record(self, run: RunIdentity, attempt: Attempt, state: str, *, reason: str | None = None) -> dict[str, Any]:
        record = self._expected_record(run, attempt.identity, state)
        if reason is not None:
            record["lifecycle"] = {"terminal_reason": reason}
        return record

    def publish_terminal(self, attempt: Attempt, state: str, *, result: dict[str, Any] | None = None) -> Attempt:
        if state not in TERMINAL_STATES:
            raise AttemptStateError("terminal state is invalid")
        run, root = self._bound_attempt(attempt)
        record = self._attempt_record(root, run, attempt.identity, absent_ok=True)
        if record is None:
            reason = str((result or {}).get("terminal_reason") or state)
            _write_new(root / "attempt.json", _json_bytes(self._initial_record(run, attempt, state, reason=reason)))
            return Attempt(attempt.identity, attempt.root, state)
        if record["state"] in TERMINAL_STATES:
            if record["state"] != state:
                raise AttemptStateError("terminal attempt is immutable")
            return Attempt(attempt.identity, attempt.root, state)
        if record["state"] != NONTERMINAL_STATE:
            raise AttemptStateError("attempt transition is invalid")
        record["state"] = state
        record["lifecycle"] = {"terminal_reason": str((result or {}).get("terminal_reason") or state)}
        _replace(root / "attempt.json", _json_bytes(record))
        return Attempt(attempt.identity, attempt.root, state)

    def _locator_from_record(self, root: Path, raw: Any) -> SupervisorLocator:
        fields = {"supervisor_pid", "start_identity", "socket_path", "challenge", "control_dir", "created_at"}
        if not isinstance(raw, dict) or set(raw) != fields:
            raise AttemptStateError("locator is invalid")
        try:
            locator = SupervisorLocator(**raw)
        except TypeError as exc:
            raise AttemptStateError("locator is invalid") from exc
        if not isinstance(locator.supervisor_pid, int) or locator.supervisor_pid < 1 or any(not isinstance(value, str) or not value for value in (locator.start_identity, locator.socket_path, locator.challenge, locator.control_dir, locator.created_at)):
            raise AttemptStateError("locator is invalid")
        control = Path(locator.control_dir)
        socket = Path(locator.socket_path)
        if not _contained(control, root) or not _contained(socket, control) or socket.parent != control:
            raise AttemptStateError("locator escapes attempt root")
        return locator

    def record_locator(self, attempt: Attempt, locator: SupervisorLocator, invocation_digest: str) -> None:
        run, root = self._bound_attempt(attempt)
        record = self._attempt_record(root, run, attempt.identity)
        if record is None or record["state"] != NONTERMINAL_STATE or "locator" in record:
            raise AttemptStateError("locator transition is invalid")
        if not isinstance(invocation_digest, str) or not DIGEST_RE.fullmatch(invocation_digest):
            raise AttemptStateError("invocation digest is invalid")
        raw = {
            "supervisor_pid": locator.supervisor_pid,
            "start_identity": locator.start_identity,
            "socket_path": locator.socket_path,
            "challenge": locator.challenge,
            "control_dir": locator.control_dir,
            "created_at": locator.created_at,
        }
        self._locator_from_record(root, raw)
        registered = self._read_json_file(Path(locator.control_dir), "locator.json")
        if registered != raw:
            raise AttemptStateError("registered locator is invalid")
        record["locator"] = raw
        record["spec_digest"] = invocation_digest
        _replace(root / "attempt.json", _json_bytes(record))

    @staticmethod
    def _state_for_reason(reason: str) -> str:
        if reason == "success":
            return "success"
        if reason == "timed_out":
            return "timed_out"
        if reason == "cancelled":
            return "cancelled"
        return "failed"

    def _read_json_file(self, root: Path, name: str) -> dict[str, Any]:
        path = root / name
        try:
            value = json.loads(_read_regular_bytes(path, name).decode("utf-8"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise AttemptStateError(f"{name} is invalid") from exc
        if not isinstance(value, dict):
            raise AttemptStateError(f"{name} is invalid")
        return value

    @staticmethod
    def _exact_fields(value: Any, fields: set[str], label: str) -> dict[str, Any]:
        if not isinstance(value, dict) or set(value) != fields:
            raise AttemptStateError(f"{label} schema is invalid")
        return value

    @staticmethod
    def _optional_int(value: Any) -> bool:
        return value is None or (isinstance(value, int) and not isinstance(value, bool))

    def _validate_result_record(self, result: dict[str, Any], locator: SupervisorLocator, expected_digest: str) -> None:
        fields = {
            "record", "success", "terminal_reason", "exit_code", "signal", "submitted",
            "finish_then_idle_then_quiet", "cleanup_complete", "process_group_alive",
            "submission_mode", "completion_mode", "spec_digest", "locator", "started_at",
            "ended_at", "duration_ns", "stdout", "stderr", "events",
        }
        self._exact_fields(result, fields, "lifecycle result")
        required_bools = ("success", "submitted", "finish_then_idle_then_quiet", "cleanup_complete", "process_group_alive")
        if result["record"] != "result" or any(not isinstance(result[key], bool) for key in required_bools):
            raise AttemptStateError("lifecycle result is invalid")
        if result["terminal_reason"] not in TERMINAL_REASONS or result["spec_digest"] != expected_digest:
            raise AttemptStateError("lifecycle result identity is invalid")
        if not self._optional_int(result["exit_code"]) or not self._optional_int(result["signal"]):
            raise AttemptStateError("lifecycle result is invalid")
        if result["submission_mode"] not in SUBMISSION_MODES or result["completion_mode"] not in COMPLETION_MODES:
            raise AttemptStateError("lifecycle result is invalid")
        if not all(isinstance(result[key], str) for key in ("started_at", "ended_at")) or not isinstance(result["duration_ns"], int) or isinstance(result["duration_ns"], bool) or result["duration_ns"] < 0:
            raise AttemptStateError("lifecycle result is invalid")
        expected_public = self._public_locator(locator)
        if result["locator"] != expected_public:
            raise AttemptStateError("lifecycle terminal locator is invalid")
        for name, stream in (("stdout", "stdout"), ("stderr", "stderr")):
            capture = self._exact_fields(result[name], {"stream", "text", "line_count", "byte_count", "truncated"}, f"{name} capture")
            if capture["stream"] != stream or not isinstance(capture["text"], str) or not isinstance(capture["line_count"], int) or not isinstance(capture["byte_count"], int) or isinstance(capture["line_count"], bool) or isinstance(capture["byte_count"], bool) or capture["line_count"] < 0 or capture["byte_count"] < 0 or not isinstance(capture["truncated"], bool):
                raise AttemptStateError("lifecycle capture is invalid")
        if not isinstance(result["events"], list):
            raise AttemptStateError("lifecycle events are invalid")
        for event in result["events"]:
            checked = self._exact_fields(event, {"record", "kind", "source", "stream", "monotonic_ns", "source_monotonic_ns", "observed_at", "detail"}, "lifecycle event")
            if checked["record"] != "event" or not all(isinstance(checked[key], str) for key in ("kind", "source", "stream", "observed_at", "detail")) or not all(isinstance(checked[key], int) and not isinstance(checked[key], bool) and checked[key] >= 0 for key in ("monotonic_ns", "source_monotonic_ns")):
                raise AttemptStateError("lifecycle event is invalid")
        if not result["cleanup_complete"] or result["process_group_alive"] or result["success"] != (result["terminal_reason"] == "success"):
            raise AttemptStateError("lifecycle terminal outcome is invalid")
        if result["success"] and not result["finish_then_idle_then_quiet"]:
            raise AttemptStateError("lifecycle terminal outcome is invalid")

    @staticmethod
    def _instant(value: Any, label: str) -> _datetime.datetime:
        """Parse one produced ISO-8601 instant into a comparable UTC value."""
        try:
            parsed = _datetime.datetime.fromisoformat(str(value))
        except (TypeError, ValueError) as exc:
            raise AttemptStateError(f"{label} timestamp is invalid") from exc
        return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=_datetime.timezone.utc)

    @staticmethod
    def _validate_terminal_events(events: list[Any]) -> dict[str, int]:
        """Return the unique ordinal of every success-evidence event kind."""
        positions: dict[str, int] = {}
        for index, event in enumerate(events):
            kind = event["kind"]
            if kind not in SUCCESS_EVIDENCE_KINDS:
                continue
            if kind in positions:
                raise AttemptStateError("lifecycle events are invalid")
            positions[kind] = index
        return positions

    def _validate_terminal_coherence(self, result: dict[str, Any], receipt: dict[str, Any], positions: dict[str, int]) -> None:
        """Bind result, ordered events and cleanup receipt to one terminal projection."""
        submitted, finish, idle, quiet = (positions.get(kind) for kind in SUCCESS_EVIDENCE_KINDS)
        ordered = finish is not None and idle is not None and quiet is not None and finish < idle < quiet
        if ordered != result["finish_then_idle_then_quiet"]:
            raise AttemptStateError("lifecycle ordered evidence is invalid")
        if result["exit_code"] != receipt["exit_code"] or result["signal"] != receipt["signal"]:
            raise AttemptStateError("lifecycle terminal outcome is invalid")
        if result["submitted"] and not receipt["caller_launched"]:
            raise AttemptStateError("lifecycle terminal outcome is invalid")
        if result["success"] and (not result["submitted"] or submitted is None or finish is None or submitted > finish):
            raise AttemptStateError("lifecycle success evidence is invalid")
        started = self._instant(result["started_at"], "lifecycle result")
        ended = self._instant(result["ended_at"], "lifecycle result")
        completed = self._instant(receipt["completed_at"], "cleanup receipt")
        if completed < started or ended < completed:
            raise AttemptStateError("lifecycle terminal chronology is invalid")

    def _validate_receipt_record(self, receipt: dict[str, Any], locator: SupervisorLocator) -> dict[str, Any]:
        """Validate one cleanup receipt against the registered supervisor identity."""
        self._exact_fields(receipt, RECEIPT_FIELDS, "cleanup receipt")
        if receipt["receipt_version"] != RECEIPT_VERSION or receipt["supervisor_pid"] != locator.supervisor_pid or receipt["challenge_digest"] != self._public_locator(locator)["challenge_digest"] or receipt["reason"] not in TERMINAL_REASONS or not self._optional_int(receipt["exit_code"]) or not self._optional_int(receipt["signal"]) or not isinstance(receipt["caller_launched"], bool) or receipt["cleanup_complete"] is not True or receipt["process_group_alive"] is not False or not isinstance(receipt["completed_at"], str):
            raise AttemptStateError("cleanup receipt is invalid")
        return receipt

    @staticmethod
    def _public_locator(locator: SupervisorLocator) -> dict[str, Any]:
        return {
            "supervisor_pid": locator.supervisor_pid,
            "start_identity": locator.start_identity,
            "socket_path": locator.socket_path,
            "control_dir": locator.control_dir,
            "challenge_digest": hashlib.sha256(locator.challenge.encode("utf-8")).hexdigest(),
            "created_at": locator.created_at,
        }

    def _validate_journal_record(self, root: Path, result: dict[str, Any], expected_digest: str) -> None:
        """Validate the append-only journal against the published result record."""
        journal = root / "lifecycle-journal.jsonl"
        try:
            lines = [json.loads(line) for line in _read_regular_bytes(journal, "lifecycle journal").decode("utf-8").splitlines()]
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise AttemptStateError("lifecycle journal is invalid") from exc
        if len(lines) < 2:
            raise AttemptStateError("lifecycle journal is invalid")
        header, terminal = lines[0], lines[-1]
        self._exact_fields(header, {"record", "journal_version", "spec_digest", "submission_mode", "completion_mode", "started_at"}, "lifecycle journal header")
        self._exact_fields(terminal, {"record", "terminal_reason", "success", "cleanup_complete", "process_group_alive", "ended_at"}, "lifecycle journal terminal")
        if header["record"] != "header" or header["journal_version"] != 1 or header["spec_digest"] != expected_digest or header["submission_mode"] != result["submission_mode"] or header["completion_mode"] != result["completion_mode"] or header["started_at"] != result["started_at"] or not isinstance(header["started_at"], str) or terminal["record"] != "terminal" or terminal["terminal_reason"] != result["terminal_reason"] or terminal["success"] != result["success"] or terminal["cleanup_complete"] is not True or terminal["process_group_alive"] is not False or terminal["ended_at"] != result["ended_at"] or not isinstance(terminal["ended_at"], str):
            raise AttemptStateError("lifecycle journal is invalid")
        for event in lines[1:-1]:
            self._exact_fields(event, {"record", "kind", "source", "stream", "monotonic_ns", "source_monotonic_ns", "observed_at", "detail"}, "lifecycle journal event")
        if lines[1:-1] != result["events"]:
            raise AttemptStateError("lifecycle journal events are invalid")

    def _read_bound_lifecycle_terminal(self, root: Path, locator: SupervisorLocator, expected_digest: str) -> dict[str, Any] | None:
        result_path = root / "lifecycle-result.json"
        if not result_path.exists() and not result_path.is_symlink():
            return None
        result = self._read_json_file(root, "lifecycle-result.json")
        self._validate_result_record(result, locator, expected_digest)
        self._validate_journal_record(root, result, expected_digest)
        receipt = Path(locator.control_dir) / "cleanup-receipt.json"
        if receipt.parent != Path(locator.control_dir) or not _contained(receipt, root):
            raise AttemptStateError("cleanup receipt escapes attempt root")
        receipt_data = self._validate_receipt_record(self._read_json_file(receipt.parent, receipt.name), locator)
        if receipt_data["reason"] != result["terminal_reason"]:
            raise AttemptStateError("cleanup receipt is invalid")
        events = self._validate_terminal_events(result["events"])
        self._validate_terminal_coherence(result, receipt_data, events)
        return result

    def validate_invocation_terminal(self, attempt: Attempt, invocation: InvocationResult) -> dict[str, Any]:
        """Validate direct lifecycle output against the identity committed before launch."""
        if not isinstance(invocation, InvocationResult):
            raise AttemptStateError("invocation result is invalid")
        run, root = self._bound_attempt(attempt)
        record = self._attempt_record(root, run, attempt.identity)
        if record is None or record["state"] != NONTERMINAL_STATE:
            raise AttemptStateError("invocation transition is invalid")
        raw_locator = record.get("locator")
        expected_digest = record.get("spec_digest")
        if raw_locator is None or not isinstance(expected_digest, str):
            raise AttemptStateError("invocation identity was not committed")
        locator = self._locator_from_record(root, raw_locator)
        expected_result_path = root / "lifecycle-result.json"
        expected_journal_path = root / "lifecycle-journal.jsonl"
        if invocation.locator != locator or invocation.spec_digest != expected_digest or Path(invocation.result_path).resolve(strict=False) != expected_result_path.resolve() or Path(invocation.journal_path).resolve(strict=False) != expected_journal_path.resolve():
            raise AttemptStateError("invocation result identity is invalid")
        terminal = self._read_bound_lifecycle_terminal(root, locator, expected_digest)
        if terminal is None:
            raise AttemptStateError("lifecycle terminal is unavailable")
        fields = ("success", "terminal_reason", "exit_code", "signal", "submitted", "finish_then_idle_then_quiet", "cleanup_complete", "process_group_alive", "spec_digest", "started_at", "ended_at", "duration_ns")
        if any(getattr(invocation, field) != terminal[field] for field in fields):
            raise AttemptStateError("invocation result does not match durable terminal")
        return terminal

    def reconcile(self, attempt: Attempt) -> Attempt:
        """Commit only authenticated terminal evidence before recovery cleanup."""
        run, root = self._bound_attempt(attempt)
        record = self._attempt_record(root, run, attempt.identity, absent_ok=True)
        if record is None:
            return self.publish_terminal(attempt, "interrupted", result={"terminal_reason": "interrupted"})
        if record["state"] in TERMINAL_STATES:
            return Attempt(attempt.identity, attempt.root, str(record["state"]))
        raw_locator = record.get("locator")
        if raw_locator is None:
            return self.publish_terminal(attempt, "interrupted", result={"terminal_reason": "interrupted"})
        locator = self._locator_from_record(root, raw_locator)
        expected_digest = record.get("spec_digest")
        if not isinstance(expected_digest, str) or not DIGEST_RE.fullmatch(expected_digest):
            raise AttemptStateError("recovery identity is invalid")
        terminal = self._read_bound_lifecycle_terminal(root, locator, expected_digest)
        if terminal is not None:
            return self.publish_terminal(attempt, self._state_for_reason(terminal["terminal_reason"]), result=terminal)
        try:
            outcome = recover_invocation(locator, stop=True)
        except LifecycleRecoveryError as exc:
            raise AttemptStateError("recovery is unverified") from exc
        receipt = Path(outcome.receipt_path)
        if not outcome.cleanup_complete or outcome.process_group_alive or receipt.parent != Path(locator.control_dir) or not _contained(receipt, root):
            raise AttemptStateError("recovery cleanup is unverified")
        try:
            self._validate_receipt_record(self._read_json_file(receipt.parent, receipt.name), locator)
        except AttemptStateError as exc:
            raise AttemptStateError("recovery cleanup is unverified") from exc
        return self.publish_terminal(attempt, "interrupted", result={"terminal_reason": "interrupted"})

    def execute_attempt(
        self,
        attempt: Attempt,
        *,
        prepare: Callable[[Attempt], Any],
        invoke: Callable[[Attempt, Callable[[SupervisorLocator, str], None]], InvocationResult],
    ) -> Attempt:
        """Prepare once, publish running identity, then invoke lifecycle once."""
        run, root = self._bound_attempt(attempt)
        if self._attempt_record(root, run, attempt.identity, absent_ok=True) is not None:
            raise AttemptStateError("attempt has already been prepared")
        try:
            prepare(attempt)
        except Exception:
            _write_new(root / "attempt.json", _json_bytes(self._initial_record(run, attempt, "failed", reason="preparation_failed")))
            raise
        _write_new(root / "attempt.json", _json_bytes(self._initial_record(run, attempt, NONTERMINAL_STATE)))
        result = invoke(attempt, lambda locator, digest: self.record_locator(attempt, locator, digest))
        terminal = self.validate_invocation_terminal(attempt, result)
        return self.publish_terminal(attempt, self._state_for_reason(terminal["terminal_reason"]), result=terminal)

    def status(self, run: RunIdentity, manifest: Manifest) -> dict[str, Any]:
        """Read-only deterministic status; it neither creates nor reconciles."""
        bound_run = self.open(manifest, run.run_id)
        if bound_run != run:
            raise AttemptStateError("run identity is invalid")
        states = {name: 0 for name in sorted(TERMINAL_STATES | {NONTERMINAL_STATE})}
        for slot in self.slots(manifest):
            for attempt in self.attempts(bound_run, slot):
                states[attempt.state] += 1
        return {"run_id": bound_run.run_id, "manifest_digest": bound_run.manifest_digest, "attempts": states}


def run_slots(
    store: RunStore,
    run: RunIdentity,
    manifest: Manifest,
    *,
    adapters: dict[str, Callable[[Attempt, Callable[[SupervisorLocator, str], None]], InvocationResult]],
    prepare: Callable[[Attempt], Any],
    retry_failed: bool = False,
) -> tuple[Attempt, ...]:
    """Execute pending slots using injected adapters; never resolves a real CLI."""
    missing = {cell.caller for cell in manifest.matrix if cell.caller not in adapters}
    if missing:
        raise CapabilityUnavailable("capability-unavailable: caller-adapter")
    bound_run = store.open(manifest, run.run_id)
    if bound_run != run:
        raise AttemptStateError("run identity is invalid")
    completed: list[Attempt] = []
    with store.writer(bound_run):
        for slot in store.slots(manifest):
            existing = store.attempts(bound_run, slot)
            if existing and existing[-1].state == "success":
                continue
            if existing and existing[-1].state in {"failed", "timed_out", "cancelled"} and not retry_failed:
                continue
            if existing and existing[-1].state not in TERMINAL_STATES:
                store.reconcile(existing[-1])
                existing = store.attempts(bound_run, slot)
                if existing[-1].state == "success" or (existing[-1].state in {"failed", "timed_out", "cancelled"} and not retry_failed):
                    continue
            attempt = store.allocate(bound_run, slot)
            adapter = next(cell.caller for cell in manifest.matrix if cell.id == slot.cell_id)
            completed.append(store.execute_attempt(attempt, prepare=prepare, invoke=adapters[adapter]))
    return tuple(completed)