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, Mapping, Protocol

from scripts.agent_benchmark.connectivity import (
    CallerCapability,
    ConnectivityIssue,
    ConnectivityResult,
    EffectiveBinding,
    RequestedEffectiveBinding,
    canonical_evidence_bytes,
    validate_requested_binding,
    validate_result,
)

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,
    MatrixCell,
    Timeout,
    validate_manifest_bytes,
)
from scripts.agent_benchmark.workspace import AttemptIdentity, PreparedWorkspace

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})$")
PREFLIGHT_RE = re.compile(r"^preflight-([0-9]{6})\.json$")
DIGEST_RE = re.compile(r"^sha256:[0-9a-f]{64}$")
PREFLIGHT_SCHEMA_VERSION = "1"
PREFLIGHT_STATUSES = ("ready", "registration_required", "implementation_gap")
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


@dataclass(frozen=True)
class PreflightObservation:
    """One adapter result plus opaque identities safe for durable evidence."""

    result: ConnectivityResult
    endpoint_identity: str
    config_identity: str


class PreflightAdapter(Protocol):
    """Closed adapter boundary consumed by the public preflight controller."""

    capability: CallerCapability

    def preflight(self, cell: "MatrixCell") -> PreflightObservation:
        """Return one typed observation without exposing raw caller output."""


class ExecutionAdapter(PreflightAdapter, Protocol):
    """Typed caller boundary for one preflight-approved scored attempt."""

    def invoke(
        self,
        cell: "MatrixCell",
        prepared: PreparedWorkspace,
        attempt: Attempt,
        task_payload: bytes,
        timeout: Timeout,
        on_started: Callable[[SupervisorLocator, str], None],
    ) -> InvocationResult:
        """Submit the exact task once for the bound cell and prepared identity."""


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


def _requested_binding(cell: MatrixCell) -> RequestedEffectiveBinding:
    return RequestedEffectiveBinding(
        cell.id,
        cell.caller,
        cell.iop.route_kind,
        cell.iop.route_id,
        cell.iop.request_model,
        cell.iop.requested_effort,
    )


def _overall_preflight_status(statuses: Iterator[str]) -> str:
    found = tuple(statuses)
    if not found:
        raise AttemptStateError("preflight result set is empty")
    if any(status == "implementation_gap" for status in found):
        return "implementation_gap"
    if any(status == "registration_required" for status in found):
        return "registration_required"
    if all(status == "ready" for status in found):
        return "ready"
    raise AttemptStateError("preflight status is invalid")


def _preflight_counts(results: list[dict[str, Any]]) -> dict[str, int]:
    counts = {status: 0 for status in PREFLIGHT_STATUSES}
    for result in results:
        status = result.get("status")
        if status not in counts:
            raise AttemptStateError("preflight result status is invalid")
        counts[str(status)] += 1
    return counts


def _connectivity_result_from_payload(
    payload: dict[str, Any], cell: MatrixCell
) -> tuple[ConnectivityResult, str, str]:
    """Rebuild one closed result so durable reads re-run semantic validation."""
    if not isinstance(payload, dict) or set(payload) != {
        "schema_version", "cell", "status", "binding", "issues",
        "endpoint_identity", "config_identity",
    }:
        raise AttemptStateError("preflight result schema is invalid")
    binding_raw = payload.get("binding")
    if not isinstance(binding_raw, dict) or set(binding_raw) != {
        "cell_id", "caller", "requested_route_kind", "requested_route_id",
        "requested_model", "requested_effort", "effective_route_kind",
        "effective_route_id", "effective_model", "effective_effort",
        "effective_bindings",
    }:
        raise AttemptStateError("preflight result binding is invalid")
    stages_raw = binding_raw.get("effective_bindings")
    if not isinstance(stages_raw, list):
        raise AttemptStateError("preflight result binding is invalid")
    stages: list[EffectiveBinding] = []
    for stage in stages_raw:
        if not isinstance(stage, dict) or set(stage) != {"stage", "model", "effort"}:
            raise AttemptStateError("preflight result binding is invalid")
        stages.append(EffectiveBinding(stage["stage"], stage["model"], stage["effort"]))
    issues_raw = payload.get("issues")
    if not isinstance(issues_raw, list):
        raise AttemptStateError("preflight result issues are invalid")
    issues: list[ConnectivityIssue] = []
    for issue in issues_raw:
        if not isinstance(issue, dict) or set(issue) != {"code", "resume_code"}:
            raise AttemptStateError("preflight result issues are invalid")
        issues.append(ConnectivityIssue(issue["code"], issue["resume_code"]))
    try:
        binding = RequestedEffectiveBinding(
            binding_raw["cell_id"],
            binding_raw["caller"],
            binding_raw["requested_route_kind"],
            binding_raw["requested_route_id"],
            binding_raw["requested_model"],
            binding_raw["requested_effort"],
            binding_raw["effective_route_kind"],
            binding_raw["effective_route_id"],
            binding_raw["effective_model"],
            binding_raw["effective_effort"],
            tuple(stages),
        )
        # Capability is deliberately not serialized.  The persisted proof is
        # revalidated against the exact immutable cell and its one direct route.
        capability = CallerCapability(
            cell.caller, (cell.iop.route_kind,), (cell.iop.requested_effort,)
        )
        result = ConnectivityResult(capability, binding, tuple(issues), payload["status"])
        validate_result(cell, result)
        endpoint_identity = payload["endpoint_identity"]
        config_identity = payload["config_identity"]
        canonical = json.loads(
            canonical_evidence_bytes(
                cell, result, endpoint_identity, config_identity
            ).decode("ascii")
        )
    except Exception as exc:
        raise AttemptStateError("preflight result is invalid") from exc
    if payload != canonical or payload.get("cell") != {"id": cell.id, "caller": cell.caller}:
        raise AttemptStateError("preflight result is non-canonical")
    return result, endpoint_identity, config_identity


def collect_preflight_observations(
    manifest: Manifest,
    adapters: Mapping[str, PreflightAdapter],
) -> dict[str, PreflightObservation]:
    """Validate the full registry, then probe direct cells in manifest order.

    Execution-preset cells exercise only the local adapter capability contract in
    this milestone.  They never become a synthetic live-ready observation.
    """
    if not isinstance(adapters, Mapping):
        raise CapabilityUnavailable("capability-unavailable: caller-adapter")
    required_callers = {cell.caller for cell in manifest.matrix}
    if any(caller not in adapters for caller in required_callers):
        raise CapabilityUnavailable("capability-unavailable: caller-adapter")

    for cell in manifest.matrix:
        adapter = adapters[cell.caller]
        capability = getattr(adapter, "capability", None)
        validate_requested_binding(cell, capability, _requested_binding(cell))

    observations: dict[str, PreflightObservation] = {}
    for cell in manifest.matrix:
        if cell.iop.route_kind != "direct":
            continue
        observation = adapters[cell.caller].preflight(cell)
        if not isinstance(observation, PreflightObservation):
            raise AttemptStateError("preflight observation is invalid")
        validate_result(cell, observation.result)
        # Materializing the canonical bytes validates both opaque identities and
        # proves that no adapter-specific/raw value can enter the durable record.
        canonical_evidence_bytes(
            cell,
            observation.result,
            observation.endpoint_identity,
            observation.config_identity,
        )
        observations[cell.id] = observation
    return observations


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 _direct_cells(manifest: Manifest) -> tuple[MatrixCell, ...]:
        return tuple(cell for cell in manifest.matrix if cell.iop.route_kind == "direct")

    def _preflight_root(self, run: RunIdentity, *, create: bool) -> Path:
        root = Path(run.root) / "preflight"
        if root.exists() or root.is_symlink():
            try:
                mode = os.lstat(root).st_mode
            except OSError as exc:
                raise AttemptStateError("preflight state is unavailable") from exc
            if not stat.S_ISDIR(mode) or root.is_symlink():
                raise AttemptStateError("preflight state is invalid")
            return root
        if not create:
            return root
        try:
            root.mkdir(mode=0o700)
            _fsync_dir(root.parent)
        except OSError as exc:
            raise AttemptStateError("preflight state is unavailable") from exc
        return root

    def _validate_preflight_record(
        self,
        raw: bytes,
        path: Path,
        run: RunIdentity,
        manifest: Manifest,
        expected_sequence: int,
    ) -> dict[str, Any]:
        try:
            record = json.loads(raw.decode("ascii"))
        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
            raise AttemptStateError("preflight record is invalid") from exc
        if not isinstance(record, dict) or set(record) != {
            "schema_version", "run_id", "manifest_digest", "sequence",
            "status", "results",
        }:
            raise AttemptStateError("preflight record schema is invalid")
        if (
            record["schema_version"] != PREFLIGHT_SCHEMA_VERSION
            or record["run_id"] != run.run_id
            or record["manifest_digest"] != run.manifest_digest
            or record["sequence"] != expected_sequence
            or record["status"] not in PREFLIGHT_STATUSES
            or not isinstance(record["results"], list)
        ):
            raise AttemptStateError("preflight record identity is invalid")
        direct_cells = self._direct_cells(manifest)
        if len(record["results"]) != len(direct_cells):
            raise AttemptStateError("preflight result set is invalid")
        statuses: list[str] = []
        for cell, result_payload in zip(direct_cells, record["results"]):
            result, _, _ = _connectivity_result_from_payload(result_payload, cell)
            statuses.append(result.status)
        if record["status"] != _overall_preflight_status(iter(statuses)):
            raise AttemptStateError("preflight aggregate status is invalid")
        if raw != _json_bytes(record):
            raise AttemptStateError("preflight record is non-canonical")
        expected_name = f"preflight-{expected_sequence:06d}.json"
        if path.name != expected_name:
            raise AttemptStateError("preflight sequence is invalid")
        return record

    def _preflight_records(
        self, run: RunIdentity, manifest: Manifest
    ) -> tuple[dict[str, Any], ...]:
        root = self._preflight_root(run, create=False)
        if not root.exists() and not root.is_symlink():
            return ()
        records: list[dict[str, Any]] = []
        for expected_sequence, child in enumerate(
            sorted(root.iterdir(), key=lambda item: item.name), start=1
        ):
            match = PREFLIGHT_RE.fullmatch(child.name)
            if match is None or int(match.group(1)) != expected_sequence:
                raise AttemptStateError("preflight sequence is invalid")
            raw = _read_regular_bytes(child, "preflight record")
            records.append(
                self._validate_preflight_record(
                    raw, child, run, manifest, expected_sequence
                )
            )
        return tuple(records)

    def preflights(
        self, run: RunIdentity, manifest: Manifest
    ) -> tuple[dict[str, Any], ...]:
        """Read append-only preflight records without creating or reconciling state."""
        bound_run = self.open(manifest, run.run_id)
        if bound_run != run:
            raise AttemptStateError("run identity is invalid")
        return self._preflight_records(bound_run, manifest)

    def record_preflight(
        self,
        run: RunIdentity,
        manifest: Manifest,
        observations: Mapping[str, PreflightObservation],
    ) -> dict[str, Any]:
        """Append one canonical result set while exclusively owning the run writer."""
        bound_run = self.open(manifest, run.run_id)
        if bound_run != run:
            raise AttemptStateError("run identity is invalid")
        with self.writer(bound_run):
            return self._record_preflight_locked(bound_run, manifest, observations)

    def _record_preflight_locked(
        self,
        run: RunIdentity,
        manifest: Manifest,
        observations: Mapping[str, PreflightObservation],
    ) -> dict[str, Any]:
        """Append one preflight while the caller owns the run writer."""
        bound_run = self.open(manifest, run.run_id)
        if bound_run != run:
            raise AttemptStateError("run identity is invalid")
        direct_cells = self._direct_cells(manifest)
        if set(observations) != {cell.id for cell in direct_cells}:
            raise AttemptStateError("preflight observation set is invalid")
        results: list[dict[str, Any]] = []
        for cell in direct_cells:
            observation = observations[cell.id]
            if not isinstance(observation, PreflightObservation):
                raise AttemptStateError("preflight observation is invalid")
            try:
                encoded = canonical_evidence_bytes(
                    cell,
                    observation.result,
                    observation.endpoint_identity,
                    observation.config_identity,
                )
                payload = json.loads(encoded.decode("ascii"))
            except Exception as exc:
                raise AttemptStateError("preflight observation is invalid") from exc
            _connectivity_result_from_payload(payload, cell)
            results.append(payload)
        status = _overall_preflight_status(
            iter(str(result["status"]) for result in results)
        )

        previous = self._preflight_records(bound_run, manifest)
        sequence = len(previous) + 1
        root = self._preflight_root(bound_run, create=True)
        path = root / f"preflight-{sequence:06d}.json"
        record = {
            "schema_version": PREFLIGHT_SCHEMA_VERSION,
            "run_id": bound_run.run_id,
            "manifest_digest": bound_run.manifest_digest,
            "sequence": sequence,
            "status": status,
            "results": results,
        }
        raw = _json_bytes(record)
        _write_new(path, raw)
        self._validate_preflight_record(
            _read_regular_bytes(path, "preflight record"),
            path,
            bound_run,
            manifest,
            sequence,
        )
        return record

    @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
        preflights = self._preflight_records(bound_run, manifest)
        latest = preflights[-1] if preflights else None
        projection = {
            "records": len(preflights),
            "latest_sequence": 0 if latest is None else latest["sequence"],
            "latest_status": "unavailable" if latest is None else latest["status"],
            **(
                {status: 0 for status in PREFLIGHT_STATUSES}
                if latest is None
                else _preflight_counts(latest["results"])
            ),
        }
        return {
            "run_id": bound_run.run_id,
            "manifest_digest": bound_run.manifest_digest,
            "preflight": projection,
            "attempts": states,
        }


def preflight_manifest(
    store: RunStore,
    manifest: Manifest,
    manifest_bytes: bytes,
    *,
    adapters: Mapping[str, PreflightAdapter],
) -> tuple[RunIdentity, dict[str, Any]]:
    """Collect direct observations, then create one run and append one record."""
    observations = collect_preflight_observations(manifest, adapters)
    if not observations:
        raise AttemptStateError("preflight requires a direct cell")
    run = store.create(manifest, manifest_bytes)
    return run, store.record_preflight(run, manifest, observations)


def _validate_prepared_binding(
    store: RunStore,
    manifest: Manifest,
    cell: MatrixCell,
    attempt: Attempt,
    prepared: PreparedWorkspace,
) -> None:
    """Reject any caller/cell/workspace identity drift before invocation."""
    if not isinstance(prepared, PreparedWorkspace):
        raise AttemptStateError("prepared workspace is invalid")
    if cell.id != attempt.identity.cell_id or prepared.identity != attempt.identity:
        raise AttemptStateError("prepared workspace identity mismatch")

    attempt_root = Path(attempt.root).resolve()
    workspace = Path(prepared.workspace_dir)
    session = Path(prepared.session_dir)
    if Path(prepared.attempt_root).resolve() != attempt_root:
        raise AttemptStateError("prepared attempt root mismatch")
    if (
        workspace.is_symlink()
        or session.is_symlink()
        or not workspace.is_dir()
        or not session.is_dir()
        or workspace.resolve() != attempt_root / "workspace"
        or session.resolve() != attempt_root / "session"
    ):
        raise AttemptStateError("prepared workspace path mismatch")
    if (
        not prepared.session_is_fresh
        or not isinstance(prepared.session_id, str)
        or not prepared.session_id
        or prepared.workspace_checksum != manifest.fixture.checksum
        or prepared.setup_cache_policy != manifest.setup_cache_policy
    ):
        raise AttemptStateError("prepared workspace policy mismatch")
    expected_testbed = (store.repo_root / manifest.testbed).resolve()
    if (
        not prepared.testbed_provenance.clean
        or Path(prepared.testbed_provenance.path).resolve() != expected_testbed
    ):
        raise AttemptStateError("prepared testbed provenance mismatch")


def run_slots(
    store: RunStore,
    run: RunIdentity,
    manifest: Manifest,
    *,
    adapters: Mapping[str, ExecutionAdapter],
    prepare: Callable[[Manifest, Attempt], PreparedWorkspace],
    retry_failed: bool = False,
) -> tuple[Attempt, ...]:
    """Append fresh preflight, then execute eligible slots under one writer."""
    if not isinstance(adapters, Mapping):
        raise CapabilityUnavailable("capability-unavailable: caller-adapter")
    required_callers = {cell.caller for cell in manifest.matrix}
    if any(
        caller not in adapters or not callable(getattr(adapters[caller], "invoke", None))
        for caller in required_callers
    ):
        raise CapabilityUnavailable("capability-unavailable: caller-adapter")
    if not callable(prepare):
        raise AttemptStateError("workspace preparer is invalid")

    observations = collect_preflight_observations(manifest, adapters)
    if not observations:
        raise AttemptStateError("preflight requires a direct cell")
    bound_run = store.open(manifest, run.run_id)
    if bound_run != run:
        raise AttemptStateError("run identity is invalid")
    cells = {cell.id: cell for cell in manifest.matrix}
    if len(cells) != len(manifest.matrix):
        raise AttemptStateError("manifest cell identity is invalid")
    completed: list[Attempt] = []
    with store.writer(bound_run):
        preflight = store._record_preflight_locked(
            bound_run, manifest, observations
        )
        if preflight["status"] != "ready":
            return ()
        if frozenset(observations) != frozenset(cells):
            return ()
        for slot in store.slots(manifest):
            cell = cells.get(slot.cell_id)
            if cell is None:
                raise AttemptStateError("slot cell identity is invalid")
            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)
            prepared: PreparedWorkspace | None = None

            def prepare_bound(current: Attempt) -> PreparedWorkspace:
                nonlocal prepared
                candidate = prepare(manifest, current)
                _validate_prepared_binding(store, manifest, cell, current, candidate)
                prepared = candidate
                return candidate

            def invoke_bound(
                current: Attempt,
                on_started: Callable[[SupervisorLocator, str], None],
            ) -> InvocationResult:
                if prepared is None:
                    raise AttemptStateError("prepared workspace is unavailable")
                return adapters[cell.caller].invoke(
                    cell,
                    prepared,
                    current,
                    manifest.fixture.prompt_content,
                    manifest.timeout,
                    on_started,
                )

            completed.append(
                store.execute_attempt(
                    attempt,
                    prepare=prepare_bound,
                    invoke=invoke_bound,
                )
            )
    return tuple(completed)