iop/scripts/agent_benchmark/lifecycle.py

#!/usr/bin/env python3
"""
lifecycle.py - Generic bounded caller invocation lifecycle.

Owns exactly one caller invocation with exactly one harness-owned task
submission, normalized finish/idle terminal evidence, bounded and redacted
output capture, closed completion policies, single-owner terminal
arbitration, and verified owned-process-group cleanup on every return path.

The controller never launches the caller directly. It launches an internal
supervisor (this module in supervisor mode) in a new POSIX session, receives a
durable authenticated locator, commits it through ``on_started`` and only then
authorizes the caller launch. This module encodes no caller-specific command
line or protocol; adapters inject an event parser and a redactor.
"""

from __future__ import annotations

import ctypes
import datetime
import hashlib
import hmac
import json
import os
import queue
import re
import secrets
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import threading
import time
from dataclasses import dataclass
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any, Callable, Optional

from scripts.agent_benchmark.manifest import Timeout

# ---------------------------------------------------------------------------
# Closed vocabularies
# ---------------------------------------------------------------------------

SUBMISSION_ARGV_TASK = "argv_task"
SUBMISSION_STDIN_ONCE = "stdin_once"
SUBMISSION_MODES = (SUBMISSION_ARGV_TASK, SUBMISSION_STDIN_ONCE)

COMPLETION_EXIT_AFTER_IDLE = "exit_after_idle"
COMPLETION_STOP_AFTER_IDLE = "stop_after_idle"
COMPLETION_MODES = (COMPLETION_EXIT_AFTER_IDLE, COMPLETION_STOP_AFTER_IDLE)

EVENT_SUBMITTED = "submitted"
EVENT_FIRST_OUTPUT = "first_output"
EVENT_FINISH = "finish"
EVENT_IDLE = "idle"
EVENT_QUIET = "quiet"
EVENT_EXITED = "exited"
EVENT_TERMINAL = "terminal"
PARSER_TERMINAL_KINDS = (EVENT_FINISH, EVENT_IDLE)
METRIC_PREFIX = "metric:"

SOURCE_HARNESS = "harness"
SOURCE_CALLER_OUTPUT = "caller_output"
SOURCE_WORKSPACE_POLL = "workspace_poll"
METRIC_SOURCES = (SOURCE_HARNESS, SOURCE_CALLER_OUTPUT, SOURCE_WORKSPACE_POLL)

# A metric value is meaningless without the clock that produced it.  Counts are
# not temporal at all, so they carry the explicit ``none`` clock rather than an
# implied one, and no value from one clock is ever compared with another.
CLOCK_NONE = "none"
CLOCK_HARNESS_MONOTONIC = "harness_monotonic"
CLOCK_CALLER_REPORTED = "caller_reported"
CLOCK_FILESYSTEM_MTIME = "filesystem_mtime"
METRIC_CLOCKS = (
    CLOCK_NONE, CLOCK_HARNESS_MONOTONIC, CLOCK_CALLER_REPORTED, CLOCK_FILESYSTEM_MTIME,
)
TEMPORAL_CLOCKS = (
    CLOCK_HARNESS_MONOTONIC, CLOCK_CALLER_REPORTED, CLOCK_FILESYSTEM_MTIME,
)

UNIT_NANOSECONDS = "ns"
UNIT_CALLS = "calls"
UNIT_TOKENS = "tokens"

# The closed metric vocabulary.  A name that is absent here can never become
# durable evidence, and each name owns exactly one unit.
METRIC_UNITS = {
    "queue_duration": UNIT_NANOSECONDS,
    "model_duration": UNIT_NANOSECONDS,
    "tool_duration": UNIT_NANOSECONDS,
    "total_duration": UNIT_NANOSECONDS,
    "model_calls": UNIT_CALLS,
    "tool_calls": UNIT_CALLS,
    "input_tokens": UNIT_TOKENS,
    "cached_input_tokens": UNIT_TOKENS,
    "cache_write_tokens": UNIT_TOKENS,
    "output_tokens": UNIT_TOKENS,
    "reasoning_tokens": UNIT_TOKENS,
    "total_tokens": UNIT_TOKENS,
}
METRIC_NAMES = tuple(sorted(METRIC_UNITS))
DURATION_SCALES_NS = {"s": 10 ** 9, "ms": 10 ** 6, "us": 10 ** 3, "ns": 1}

REASON_SUCCESS = "success"
REASON_START_CALLBACK_FAILED = "start_callback_failed"
REASON_LAUNCH_FAILED = "launch_failed"
REASON_NONZERO_EXIT = "nonzero_exit"
REASON_MISSING_IDLE = "missing_idle"
REASON_DUPLICATE_EVENT = "duplicate_event"
REASON_OUT_OF_ORDER_EVENT = "out_of_order_event"
REASON_MALFORMED_EVENT = "malformed_event"
REASON_PARSER_ERROR = "parser_error"
REASON_READER_ERROR = "reader_error"
REASON_TIMED_OUT = "timed_out"
REASON_CANCELLED = "cancelled"
REASON_CONTROLLER_LOST = "controller_lost"
REASON_RECOVERED_STOP = "recovered_stop"
REASON_CLEANUP_FAILED = "cleanup_failed"
REASON_SUPERVISOR_ERROR = "supervisor_error"
TERMINAL_REASONS = (
    REASON_SUCCESS,
    REASON_START_CALLBACK_FAILED,
    REASON_LAUNCH_FAILED,
    REASON_NONZERO_EXIT,
    REASON_MISSING_IDLE,
    REASON_DUPLICATE_EVENT,
    REASON_OUT_OF_ORDER_EVENT,
    REASON_MALFORMED_EVENT,
    REASON_PARSER_ERROR,
    REASON_READER_ERROR,
    REASON_TIMED_OUT,
    REASON_CANCELLED,
    REASON_CONTROLLER_LOST,
    REASON_RECOVERED_STOP,
    REASON_CLEANUP_FAILED,
    REASON_SUPERVISOR_ERROR,
)

FAULT_NONE = ""
FAULT_READER_ERROR = "reader_error"
FAULT_MODES = (FAULT_NONE, FAULT_READER_ERROR)

DEFAULT_ENV_ALLOWLIST = (
    "PATH", "HOME", "LANG", "LC_ALL", "LC_CTYPE", "TZ", "TERM", "TMPDIR",
    "USER", "LOGNAME", "SHELL", "PWD", "PYTHONPATH", "PYTHONHASHSEED",
    "NO_COLOR", "CI",
)

ENV_KEY_RE = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$")
METRIC_KIND_RE = re.compile(r"^metric:[a-z0-9][a-z0-9_.+-]{0,63}$")
SAFE_LABEL_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9_.:+-]{0,63}$")

JOURNAL_FILENAME = "lifecycle-journal.jsonl"
RESULT_FILENAME = "lifecycle-result.json"
LOCATOR_FILENAME = "locator.json"
RECEIPT_FILENAME = "cleanup-receipt.json"
SOCKET_FILENAME = "control.sock"
SUPERVISOR_ERR_FILENAME = "supervisor.err"

REDACTED = "[redacted]"
RECEIPT_VERSION = 1
JOURNAL_VERSION = 1

MAX_TASK_PAYLOAD_BYTES = 1 << 20
MAX_CAPTURE_BYTES_LIMIT = 1 << 24
MAX_CAPTURE_LINES_LIMIT = 1 << 20
MAX_EVENT_DETAIL_CHARS = 512
MAX_METRIC_EVENTS = 1000
MAX_METRIC_KIND_CHARS = len(METRIC_PREFIX) + 64
MAX_PARSED_ITEMS = 16
_MAX_CHUNK_BYTES = 1 << 16
_PROXY_CAP_FACTOR = 4
_POLL_INTERVAL_SECONDS = 0.02
_REGISTER_TIMEOUT_SECONDS = 30.0
_CONTROL_SOCKET_TIMEOUT_SECONDS = 20.0
_READER_JOIN_SECONDS = 10.0
_KILL_WAIT_SECONDS = 10.0
_TERMINAL_SLACK_SECONDS = 20.0
_SUPERVISOR_EXIT_SECONDS = 10.0

_REPO_ROOT = Path(__file__).resolve().parents[2]

_FALLBACK_SECRET_PATTERNS = (
    re.compile(r"(?i)\bauthorization\s*:?[ \t]*bearer\s+\S+"),
    re.compile(r"(?i)\bbearer\s+\S+"),
    re.compile(r"(?i)\b(?:api[_-]?key|access[_-]?token|secret|password|passwd)\b"
               r"\s*[:=]\s*\S+"),
    re.compile(r"\b(?:sk|pk|rk)-[A-Za-z0-9_\-]{8,}"),
    re.compile(r"\b(?:ghp|gho|ghs|ghu)_[A-Za-z0-9]{8,}"),
    re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{8,}"),
    re.compile(r"\biop_[A-Za-z0-9_\-]{12,}"),
    re.compile(r"\bAKIA[0-9A-Z]{12,}"),
)


# ---------------------------------------------------------------------------
# Errors
# ---------------------------------------------------------------------------

class LifecycleError(Exception):
    """Base error for invocation lifecycle failures."""


class LifecycleValidationError(LifecycleError):
    """Raised when the invocation specification or environment fails preflight."""


class LifecycleProtocolError(LifecycleError):
    """Raised when the internal supervisor protocol is violated."""


class LifecycleRecoveryError(LifecycleError):
    """Raised when an authenticated recovery request cannot be trusted."""


class LifecycleMetricError(LifecycleError):
    """Raised when an observation cannot be represented without invention."""


# ---------------------------------------------------------------------------
# Frozen contracts
# ---------------------------------------------------------------------------

@dataclass(frozen=True)
class InvocationSpec:
    """Immutable specification of exactly one bounded caller invocation."""

    argv: tuple[str, ...]
    cwd: str
    env: tuple[tuple[str, str], ...]
    submission_mode: str
    completion_mode: str
    timeout: Timeout
    evidence_dir: str
    task_payload: bytes = b""
    env_allowlist: tuple[str, ...] = ()
    max_capture_bytes: int = 1 << 20
    max_capture_lines: int = 10000
    control_dir: Optional[str] = None
    caller_detaches: bool = False
    fault_injection: str = FAULT_NONE


@dataclass(frozen=True)
class LifecycleEvent:
    kind: str
    source: str
    stream: str
    monotonic_ns: int
    source_monotonic_ns: int
    observed_at: str
    detail: str


@dataclass(frozen=True)
class ParsedMetric:
    """One validated numeric observation with its clock, source and binding.

    ``stage``, ``model`` and ``call_id`` are optional closed labels; an empty
    label means the observation is an unqualified caller total.  ``overlap``
    marks an interval that may be contained in another reported interval, so a
    consumer can never treat the set as a partition to subtract.
    """

    name: str
    value: int
    unit: str
    clock: str
    source: str
    stage: str = ""
    model: str = ""
    call_id: str = ""
    overlap: bool = False


@dataclass(frozen=True)
class CaptureStream:
    stream: str
    text: str
    line_count: int
    byte_count: int
    truncated: bool


@dataclass(frozen=True)
class SupervisorLocator:
    supervisor_pid: int
    start_identity: str
    socket_path: str
    challenge: str
    control_dir: str
    created_at: str


@dataclass(frozen=True)
class TerminalOutcome:
    reason: str
    exit_code: Optional[int]
    signal: Optional[int]
    caller_launched: bool
    cleanup_complete: bool
    process_group_alive: bool
    receipt_path: str


@dataclass(frozen=True)
class InvocationResult:
    """Terminal projection of one invocation, including typed observations."""

    success: bool
    terminal_reason: str
    exit_code: Optional[int]
    signal: Optional[int]
    submitted: bool
    finish_then_idle_then_quiet: bool
    cleanup_complete: bool
    process_group_alive: bool
    events: tuple[LifecycleEvent, ...]
    stdout: CaptureStream
    stderr: CaptureStream
    journal_path: str
    result_path: str
    locator: Optional[SupervisorLocator]
    spec_digest: str
    started_at: str
    ended_at: str
    duration_ns: int
    metrics: tuple[ParsedMetric, ...] = ()


class CancellationToken:
    """Thread-safe cancellation flag accepted by :func:`run_invocation`."""

    def __init__(self) -> None:
        self._event = threading.Event()

    def cancel(self) -> None:
        self._event.set()

    def is_cancelled(self) -> bool:
        return self._event.is_set()


# ---------------------------------------------------------------------------
# Small helpers
# ---------------------------------------------------------------------------

def _utc_now() -> str:
    return datetime.datetime.now(datetime.timezone.utc).isoformat()


def _safe_detail(exc: BaseException) -> str:
    """Return a bounded, class-anchored error detail without raw inputs."""
    return f"{type(exc).__name__}"[:MAX_EVENT_DETAIL_CHARS]


@dataclass(frozen=True)
class _FileIdentity:
    device: int
    inode: int


def _fsync_directory(directory: Path) -> None:
    dir_fd = os.open(str(directory), os.O_RDONLY)
    try:
        os.fsync(dir_fd)
    finally:
        os.close(dir_fd)


def _stage_bytes(directory: Path, data: bytes, mode: int) -> Path:
    """Write and fsync private staging bytes in the target directory."""
    fd, tmp_name = tempfile.mkstemp(prefix=".tmp-", dir=str(directory))
    try:
        with os.fdopen(fd, "wb") as handle:
            handle.write(data)
            handle.flush()
            os.fsync(handle.fileno())
        os.chmod(tmp_name, mode)
    except BaseException:
        try:
            os.unlink(tmp_name)
        except OSError:
            pass
        raise
    return Path(tmp_name)


def _identity(path: Path) -> _FileIdentity:
    stat_result = path.stat(follow_symlinks=False)
    return _FileIdentity(stat_result.st_dev, stat_result.st_ino)


def _rollback_owned(path: Path, identity: _FileIdentity) -> None:
    """Remove path only while it still names the inode published by this call."""
    try:
        if _identity(path) != identity:
            return
        path.unlink()
        _fsync_directory(path.parent)
    except FileNotFoundError:
        return


def _publish_staged_no_replace(staged: Path, path: Path) -> _FileIdentity:
    """Atomically link staged bytes into an absent target without replacement."""
    staged_identity = _identity(staged)
    linked = False
    try:
        os.link(staged, path, follow_symlinks=False)
        linked = True
        if _identity(path) != staged_identity:
            raise LifecycleError("published target identity changed concurrently")
        _fsync_directory(path.parent)
        return staged_identity
    except BaseException:
        if linked:
            _rollback_owned(path, staged_identity)
        raise
    finally:
        try:
            staged.unlink()
        except FileNotFoundError:
            pass


def _write_bytes_no_replace(path: Path, data: bytes, mode: int = 0o600) -> _FileIdentity:
    """Stage, fsync and atomically publish bytes only when target is absent."""
    staged = _stage_bytes(path.parent, data, mode)
    return _publish_staged_no_replace(staged, path)


def _enable_child_subreaper() -> bool:
    """Adopt owned orphan descendants on Linux so they can be reaped."""
    if not sys.platform.startswith("linux"):
        return False
    try:
        libc = ctypes.CDLL(None, use_errno=True)
        if libc.prctl(36, 1, 0, 0, 0) != 0:  # PR_SET_CHILD_SUBREAPER
            raise OSError(ctypes.get_errno(), "prctl(PR_SET_CHILD_SUBREAPER)")
    except (AttributeError, OSError):
        return False
    return True


def _process_start_identity(pid: int) -> str:
    """Return an OS start identity for pid, or '' when unavailable."""
    try:
        raw = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8", errors="replace")
    except OSError:
        return ""
    try:
        return raw.rsplit(") ", 1)[1].split()[19]
    except (IndexError, ValueError):
        return ""


def _proc_group_has_live_member(pgid: int) -> bool:
    """Return True when /proc shows a non-zombie member of pgid."""
    proc = Path("/proc")
    for entry in proc.iterdir():
        if not entry.name.isdigit():
            continue
        try:
            raw = (entry / "stat").read_text(encoding="utf-8", errors="replace")
            fields = raw.rsplit(") ", 1)[1].split()
            state, group = fields[0], int(fields[2])
        except (OSError, IndexError, ValueError):
            continue
        if group == pgid and state != "Z":
            return True
    return False


def _proc_group_has_member(pgid: int) -> bool:
    """Return True when /proc still contains any member, including a zombie."""
    proc = Path("/proc")
    if not proc.is_dir():
        return False
    for entry in proc.iterdir():
        if not entry.name.isdigit():
            continue
        try:
            raw = (entry / "stat").read_text(encoding="utf-8", errors="replace")
            group = int(raw.rsplit(") ", 1)[1].split()[2])
        except (OSError, IndexError, ValueError):
            continue
        if group == pgid:
            return True
    return False


def _group_alive(pgid: Optional[int]) -> bool:
    """Return True when the owned process group still has a live member."""
    if not pgid:
        return False
    try:
        os.killpg(pgid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    except OSError:
        return True
    if Path("/proc/self/stat").exists():
        return _proc_group_has_live_member(pgid)
    return True


def _killpg_quiet(pgid: int, sig: int) -> None:
    try:
        os.killpg(pgid, sig)
    except (ProcessLookupError, PermissionError, OSError):
        pass


def exact_value_redactor(values: tuple[str, ...]) -> Callable[[str], str]:
    """Build a redactor replacing every non-empty exact secret value."""
    ordered = tuple(sorted({v for v in values if v}, key=len, reverse=True))

    def _redact(text: str) -> str:
        for value in ordered:
            text = text.replace(value, REDACTED)
        return text

    return _redact


def fallback_redact(text: str) -> str:
    """Redact secret-shaped substrings that no adapter redactor removed."""
    for pattern in _FALLBACK_SECRET_PATTERNS:
        text = pattern.sub(REDACTED, text)
    return text


def normalize_duration_ns(value: Any, reported_unit: str = "ms") -> int:
    """Convert one reported non-negative duration into exact integer nanoseconds.

    Callers report durations as integers or decimals.  The decimal text is the
    authority, so the value is rebuilt with ``Decimal(str(value))`` and refused
    whenever it cannot be represented in whole nanoseconds.
    """
    scale = DURATION_SCALES_NS.get(reported_unit)
    if scale is None:
        raise LifecycleMetricError("duration unit is not a supported scale")
    if isinstance(value, bool) or not isinstance(value, (int, float, str, Decimal)):
        raise LifecycleMetricError("duration value is not a reported number")
    try:
        reported = Decimal(value) if isinstance(value, int) else Decimal(str(value))
    except (InvalidOperation, ValueError) as exc:
        raise LifecycleMetricError("duration value is not a reported number") from exc
    if not reported.is_finite() or reported < 0:
        raise LifecycleMetricError("duration value is not finite and non-negative")
    exact = reported * scale
    if exact != exact.to_integral_value():
        raise LifecycleMetricError("duration precision is finer than one nanosecond")
    return int(exact)


def is_reported_number(value: Any) -> bool:
    """True only for a plain JSON number, so no wire string is coerced."""
    return not isinstance(value, bool) and isinstance(value, (int, float))


def normalize_count(value: Any) -> int:
    """Admit only a non-negative integer call or token count."""
    if isinstance(value, bool) or not isinstance(value, int) or value < 0:
        raise LifecycleMetricError("count value must be a non-negative integer")
    return value


def _safe_label(label: Any) -> bool:
    if not isinstance(label, str):
        return False
    if not label:
        return True
    return SAFE_LABEL_RE.fullmatch(label) is not None and fallback_redact(label) == label


def validate_metric(metric: Any) -> ParsedMetric:
    """Validate one observation against the closed metric contract."""
    if not isinstance(metric, ParsedMetric):
        raise LifecycleMetricError("metric must be a ParsedMetric instance")
    unit = METRIC_UNITS.get(metric.name)
    if unit is None or metric.unit != unit:
        raise LifecycleMetricError("metric name and unit are not a closed pair")
    if metric.clock not in METRIC_CLOCKS or metric.source not in METRIC_SOURCES:
        raise LifecycleMetricError("metric clock and source must be closed values")
    if not isinstance(metric.overlap, bool):
        raise LifecycleMetricError("metric overlap must be a boolean")
    if unit == UNIT_NANOSECONDS:
        if metric.clock not in TEMPORAL_CLOCKS:
            raise LifecycleMetricError("a duration requires a temporal clock")
    elif metric.clock != CLOCK_NONE or metric.overlap:
        raise LifecycleMetricError("a count has no clock and cannot overlap")
    if isinstance(metric.value, bool) or not isinstance(metric.value, int) or metric.value < 0:
        raise LifecycleMetricError("metric value must be a non-negative integer")
    if not all(_safe_label(label) for label in (metric.stage, metric.model, metric.call_id)):
        raise LifecycleMetricError("metric labels must be safe closed identifiers")
    return metric


def duration_metric(
    name: str,
    value: Any,
    *,
    reported_unit: str = "ms",
    clock: str = CLOCK_CALLER_REPORTED,
    source: str = SOURCE_CALLER_OUTPUT,
    stage: str = "",
    model: str = "",
    call_id: str = "",
    overlap: bool = False,
) -> ParsedMetric:
    """Build one validated duration observation in integer nanoseconds."""
    return validate_metric(ParsedMetric(
        name, normalize_duration_ns(value, reported_unit), UNIT_NANOSECONDS,
        clock, source, stage, model, call_id, overlap,
    ))


def count_metric(
    name: str,
    value: Any,
    *,
    source: str = SOURCE_CALLER_OUTPUT,
    stage: str = "",
    model: str = "",
    call_id: str = "",
) -> ParsedMetric:
    """Build one validated call or token count observation."""
    unit = METRIC_UNITS.get(name)
    if unit not in (UNIT_CALLS, UNIT_TOKENS):
        raise LifecycleMetricError("metric name is not a closed count")
    return validate_metric(ParsedMetric(
        name, normalize_count(value), unit, CLOCK_NONE, source, stage, model, call_id, False,
    ))


def metric_record(metric: ParsedMetric) -> dict[str, Any]:
    """Return the canonical durable projection of one validated observation."""
    validated = validate_metric(metric)
    return {
        "name": validated.name,
        "value": validated.value,
        "unit": validated.unit,
        "clock": validated.clock,
        "source": validated.source,
        "stage": validated.stage,
        "model": validated.model,
        "call_id": validated.call_id,
        "overlap": validated.overlap,
    }


def publish_bytes_no_replace(path: Path, data: bytes, mode: int = 0o600) -> None:
    """Publish bytes atomically and only into an absent target."""
    _write_bytes_no_replace(path, data, mode)


def spec_digest(spec: InvocationSpec) -> str:
    """Compute a stable digest binding argv/env/payload without revealing them."""
    hasher = hashlib.sha256()
    hasher.update(b"IOP-BENCH-INVOCATION\x00")
    for item in spec.argv:
        hasher.update(item.encode("utf-8") + b"\x00")
    for key, value in spec.env:
        hasher.update(key.encode("utf-8") + b"=" + value.encode("utf-8") + b"\x00")
    hasher.update(spec.cwd.encode("utf-8") + b"\x00")
    hasher.update(spec.submission_mode.encode("utf-8") + b"\x00")
    hasher.update(spec.completion_mode.encode("utf-8") + b"\x00")
    hasher.update(hashlib.sha256(spec.task_payload).digest())
    return "sha256:" + hasher.hexdigest()


def env_pairs(mapping: dict[str, str]) -> tuple[tuple[str, str], ...]:
    """Freeze an environment mapping into canonical immutable pairs."""
    return tuple(sorted((str(k), str(v)) for k, v in mapping.items()))


# ---------------------------------------------------------------------------
# Frame transport
# ---------------------------------------------------------------------------

class _FrameWriter:
    """Serialized newline-delimited JSON writer shared by supervisor threads."""

    def __init__(self, handle: Any) -> None:
        self._handle = handle
        self._lock = threading.Lock()
        self.alive = True

    def send(self, frame: dict[str, Any]) -> None:
        payload = (json.dumps(frame, ensure_ascii=False) + "\n").encode("utf-8")
        with self._lock:
            if not self.alive:
                return
            try:
                self._handle.write(payload)
                self._handle.flush()
            except (BrokenPipeError, ValueError, OSError):
                self.alive = False


def _read_frame(handle: Any) -> Optional[dict[str, Any]]:
    """Read one JSON frame; return None on EOF or a closed handle."""
    try:
        line = handle.readline()
    except (ValueError, OSError):
        return None
    if not line:
        return None
    try:
        frame = json.loads(line)
    except (json.JSONDecodeError, UnicodeDecodeError):
        return {"op": "error", "detail": "malformed_frame"}
    return frame if isinstance(frame, dict) else {"op": "error", "detail": "malformed_frame"}


def _send_json(handle: Any, payload: dict[str, Any]) -> None:
    handle.write((json.dumps(payload, ensure_ascii=False) + "\n").encode("utf-8"))
    handle.flush()


# ---------------------------------------------------------------------------
# Supervisor
# ---------------------------------------------------------------------------

class _Supervisor:
    """Registered owner of one caller process group and its terminal arbitration."""

    def __init__(self, reader: Any, writer: Any, control_dir: Path) -> None:
        self.reader = reader
        self.writer = _FrameWriter(writer)
        self.control_dir = control_dir
        self.spec: dict[str, Any] = {}
        self.challenge = secrets.token_hex(32)
        self.start_identity = _process_start_identity(os.getpid())
        self.child: Optional[subprocess.Popen] = None
        self.pgid: Optional[int] = None
        self.readers: list[threading.Thread] = []
        self.submission_writer: Optional[threading.Thread] = None
        self.exit_watcher: Optional[threading.Thread] = None
        self.subreaper_enabled = False
        self.forwarded: dict[str, int] = {"stdout": 0, "stderr": 0}
        self.truncated: dict[str, bool] = {"stdout": False, "stderr": False}
        self.sock: Optional[socket.socket] = None
        self.terminal: Optional[dict[str, Any]] = None
        self._arbiter_lock = threading.Lock()
        self._terminal_sent = False
        self._send_lock = threading.Lock()

    # -- registration ------------------------------------------------------

    def _register(self) -> None:
        socket_path = self.control_dir / SOCKET_FILENAME
        self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
        previous_umask = os.umask(0o177)
        try:
            self.sock.bind(str(socket_path))
        finally:
            os.umask(previous_umask)
        os.chmod(socket_path, 0o600)
        self.sock.listen(4)
        locator = {
            "supervisor_pid": os.getpid(),
            "start_identity": self.start_identity,
            "socket_path": str(socket_path),
            "challenge": self.challenge,
            "control_dir": str(self.control_dir),
            "created_at": _utc_now(),
        }
        _write_bytes_no_replace(
            self.control_dir / LOCATOR_FILENAME,
            json.dumps(locator, ensure_ascii=False).encode("utf-8"),
        )
        threading.Thread(target=self._serve_control, daemon=True).start()
        self.writer.send({"op": "registered", "locator": locator})

    # -- caller launch -----------------------------------------------------

    def _launch(self) -> None:
        spec = self.spec
        mode = spec["submission_mode"]
        stdin = subprocess.PIPE if mode == SUBMISSION_STDIN_ONCE else subprocess.DEVNULL
        self.subreaper_enabled = _enable_child_subreaper()
        self.child = subprocess.Popen(
            list(spec["argv"]),
            cwd=spec["cwd"],
            env={key: value for key, value in spec["env"]},
            stdin=stdin,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            process_group=0,
            close_fds=True,
        )
        self.pgid = self.child.pid
        for name in ("stdout", "stderr"):
            thread = threading.Thread(
                target=self._pump, args=(name, getattr(self.child, name)), daemon=True
            )
            thread.start()
            self.readers.append(thread)
        self.exit_watcher = threading.Thread(target=self._await_exit, daemon=True)
        self.exit_watcher.start()
        if mode == SUBMISSION_STDIN_ONCE:
            payload = bytes.fromhex(spec["task_payload_hex"])
            self.submission_writer = threading.Thread(
                target=self._submit_stdin_once, args=(payload,), daemon=True
            )
            self.submission_writer.start()
        else:
            self._send_started()

    def _send_started(self) -> None:
        child = self.child
        if child is None:
            return
        self.writer.send({
            "op": "started",
            "pid": child.pid,
            "pgid": self.pgid,
            "ns": time.monotonic_ns(),
            "submission_mode": self.spec["submission_mode"],
        })

    def _submit_stdin_once(self, payload: bytes) -> None:
        child = self.child
        if child is None or child.stdin is None:
            return
        complete = False
        try:
            written = child.stdin.write(payload)
            child.stdin.flush()
            complete = written == len(payload)
        except (BrokenPipeError, ValueError, OSError):
            complete = False
        finally:
            try:
                child.stdin.close()
            except (BrokenPipeError, ValueError, OSError):
                complete = False
        if complete:
            self._send_started()
        else:
            self.writer.send({"op": "submission_error", "detail": "stdin_write_failed"})

    def _pump(self, stream: str, pipe: Any) -> None:
        cap = int(self.spec["max_capture_bytes"]) * _PROXY_CAP_FACTOR + _MAX_CHUNK_BYTES
        injected = self.spec.get("fault_injection", FAULT_NONE) == FAULT_READER_ERROR
        try:
            while True:
                chunk = pipe.readline(_MAX_CHUNK_BYTES)
                if not chunk:
                    break
                if injected and stream == "stdout":
                    raise OSError("injected reader failure")
                now = time.monotonic_ns()
                if self.forwarded[stream] < cap:
                    self.forwarded[stream] += len(chunk)
                    self.writer.send({
                        "op": "output",
                        "stream": stream,
                        "ns": now,
                        "data": chunk.decode("utf-8", "replace"),
                    })
                elif not self.truncated[stream]:
                    self.truncated[stream] = True
                    self.writer.send({"op": "proxy_truncated", "stream": stream, "ns": now})
        except Exception as exc:  # reader failure is terminal evidence, never silent
            self.writer.send(
                {"op": "reader_error", "stream": stream, "detail": _safe_detail(exc)}
            )
        finally:
            try:
                pipe.close()
            except OSError:
                pass
            self.writer.send({
                "op": "stream_eof",
                "stream": stream,
                "ns": time.monotonic_ns(),
            })

    def _await_exit(self) -> None:
        child = self.child
        if child is None:
            return
        try:
            code = child.wait()
        except OSError as exc:
            self.writer.send({"op": "error", "detail": _safe_detail(exc)})
            return
        self.writer.send({
            "op": "exited",
            "exit_code": code if code >= 0 else None,
            "signal": -code if code < 0 else None,
            "ns": time.monotonic_ns(),
        })

    # -- terminal arbitration ---------------------------------------------

    def finish(self, reason: str) -> dict[str, Any]:
        """Single-owner terminal arbiter: first reason wins, cleanup always runs."""
        with self._arbiter_lock:
            if self.terminal is not None:
                return self.terminal
            exit_code: Optional[int] = None
            signal_num: Optional[int] = None
            group_alive = False
            if self.child is not None:
                grace = int(self.spec.get("cleanup_grace_seconds", 5))
                self._terminate_group(grace)
                code = self.child.returncode
                if code is not None:
                    exit_code = code if code >= 0 else None
                    signal_num = -code if code < 0 else None
                group_alive = _group_alive(self.pgid)
            io_complete = self._join_io_threads()
            descendants_reaped = self._reap_owned_descendants()
            outcome = {
                "reason": reason,
                "exit_code": exit_code,
                "signal": signal_num,
                "caller_launched": self.child is not None,
                "cleanup_complete": not group_alive and io_complete and descendants_reaped,
                "process_group_alive": group_alive,
                "receipt_path": str(self.control_dir / RECEIPT_FILENAME),
            }
            if not self._write_receipt(outcome):
                outcome["reason"] = REASON_CLEANUP_FAILED
                outcome["cleanup_complete"] = False
            self.terminal = outcome
            return outcome

    def _terminate_group(self, grace_seconds: int) -> None:
        child = self.child
        if child is None or self.pgid is None:
            return
        if not _group_alive(self.pgid):
            child.poll()
            return
        _killpg_quiet(self.pgid, signal.SIGTERM)
        if self._wait_group_gone(grace_seconds):
            return
        _killpg_quiet(self.pgid, signal.SIGKILL)
        self._wait_group_gone(_KILL_WAIT_SECONDS)

    def _wait_group_gone(self, seconds: float) -> bool:
        deadline = time.monotonic() + max(0.0, float(seconds))
        while True:
            if self.child is not None:
                self.child.poll()
            if not _group_alive(self.pgid):
                return True
            if time.monotonic() >= deadline:
                return False
            time.sleep(_POLL_INTERVAL_SECONDS)

    def _join_io_threads(self) -> bool:
        deadline = time.monotonic() + _READER_JOIN_SECONDS
        threads = [self.submission_writer, *self.readers, self.exit_watcher]
        for thread in threads:
            if thread is None:
                continue
            thread.join(max(0.0, deadline - time.monotonic()))
        return all(thread is None or not thread.is_alive() for thread in threads)

    def _reap_owned_descendants(self) -> bool:
        if self.pgid is None or not self.subreaper_enabled:
            return True
        deadline = time.monotonic() + _KILL_WAIT_SECONDS
        while True:
            reaped = False
            try:
                while True:
                    pid, _ = os.waitpid(-self.pgid, os.WNOHANG)
                    if pid <= 0:
                        break
                    reaped = True
            except ChildProcessError:
                pass
            if not _proc_group_has_member(self.pgid):
                return True
            if time.monotonic() >= deadline:
                return False
            if not reaped:
                time.sleep(_POLL_INTERVAL_SECONDS)

    def _write_receipt(self, outcome: dict[str, Any]) -> bool:
        receipt = {
            "receipt_version": RECEIPT_VERSION,
            "supervisor_pid": os.getpid(),
            "challenge_digest": hashlib.sha256(self.challenge.encode("utf-8")).hexdigest(),
            "reason": outcome["reason"],
            "exit_code": outcome["exit_code"],
            "signal": outcome["signal"],
            "caller_launched": outcome["caller_launched"],
            "cleanup_complete": outcome["cleanup_complete"],
            "process_group_alive": outcome["process_group_alive"],
            "completed_at": _utc_now(),
        }
        try:
            _write_bytes_no_replace(
                self.control_dir / RECEIPT_FILENAME,
                json.dumps(receipt, ensure_ascii=False).encode("utf-8"),
            )
        except (OSError, LifecycleError):
            return False
        return True

    def _send_terminal(self, outcome: dict[str, Any]) -> None:
        with self._send_lock:
            if self._terminal_sent:
                return
            self._terminal_sent = True
        self.writer.send({"op": "terminal", "outcome": outcome})

    # -- authenticated control endpoint -----------------------------------

    def _serve_control(self) -> None:
        while True:
            try:
                conn, _ = self.sock.accept()  # type: ignore[union-attr]
            except OSError:
                return
            threading.Thread(
                target=self._handle_control, args=(conn,), daemon=True
            ).start()

    def _handle_control(self, conn: socket.socket) -> None:
        with conn:
            conn.settimeout(_CONTROL_SOCKET_TIMEOUT_SECONDS)
            stream = conn.makefile("rwb")
            try:
                self._control_exchange(stream)
            except (OSError, ValueError, json.JSONDecodeError):
                return
            finally:
                try:
                    stream.close()
                except OSError:
                    pass

    def _control_exchange(self, stream: Any) -> None:
        auth = _read_frame(stream) or {}
        presented = str(auth.get("challenge", ""))
        if auth.get("op") != "auth" or not hmac.compare_digest(presented, self.challenge):
            _send_json(stream, {"ok": False, "error": "authentication_failed"})
            return
        _send_json(stream, {
            "ok": True,
            "supervisor_pid": os.getpid(),
            "start_identity": self.start_identity,
        })
        request = _read_frame(stream) or {}
        operation = request.get("op")
        if operation == "status":
            _send_json(stream, {"ok": True, "status": self._status()})
            return
        if operation != "stop":
            _send_json(stream, {"ok": False, "error": "unsupported_op"})
            return
        outcome = self.finish(REASON_RECOVERED_STOP)
        _send_json(stream, {"ok": True, "outcome": outcome})
        self._send_terminal(outcome)
        self._exit_after_recovered_stop(stream)

    def _status(self) -> dict[str, Any]:
        return {
            "caller_launched": self.child is not None,
            "process_group_alive": _group_alive(self.pgid),
            "terminal_reason": None if self.terminal is None else self.terminal["reason"],
        }

    def _exit_after_recovered_stop(self, stream: Any) -> None:
        """Leave immediately after a recovered stop; cleanup and receipt are done."""
        for handle in (stream, self.sock):
            try:
                if handle is not None:
                    handle.close()
            except OSError:
                pass
        os._exit(0)

    # -- main loop ---------------------------------------------------------

    def run(self) -> int:
        try:
            spec = _read_frame(self.reader)
            if spec is None or spec.get("op") != "spec":
                self._send_terminal(self.finish(REASON_CONTROLLER_LOST))
                return 0
            self.spec = spec
            self._register()
        except Exception as exc:
            self.writer.send({"op": "error", "detail": _safe_detail(exc)})
            self._send_terminal(self.finish(REASON_SUPERVISOR_ERROR))
            return 1

        gate = _read_frame(self.reader)
        if gate is None:
            self._send_terminal(self.finish(REASON_CONTROLLER_LOST))
            return 0
        if gate.get("op") != "start":
            self._send_terminal(self.finish(REASON_START_CALLBACK_FAILED))
            return 0

        try:
            self._launch()
        except Exception as exc:
            self.writer.send({"op": "error", "detail": _safe_detail(exc)})
            self._send_terminal(self.finish(REASON_LAUNCH_FAILED))
            return 0

        outcome = self._control_loop()
        self._send_terminal(outcome)
        self._close_socket()
        return 0

    def _control_loop(self) -> dict[str, Any]:
        while True:
            frame = _read_frame(self.reader)
            if frame is None:
                return self.finish(REASON_CONTROLLER_LOST)
            if frame.get("op") == "stop":
                reason = str(frame.get("reason") or REASON_SUPERVISOR_ERROR)
                if reason not in TERMINAL_REASONS:
                    reason = REASON_SUPERVISOR_ERROR
                return self.finish(reason)

    def _close_socket(self) -> None:
        if self.sock is None:
            return
        try:
            self.sock.close()
        except OSError:
            pass
        try:
            os.unlink(self.control_dir / SOCKET_FILENAME)
        except OSError:
            pass


def _supervisor_main(argv: list[str]) -> int:
    options: dict[str, str] = {}
    for item in argv:
        key, _, value = item.partition("=")
        options[key] = value
    read_fd = int(options["--read-fd"])
    write_fd = int(options["--write-fd"])
    control_dir = Path(options["--control-dir"])
    reader = os.fdopen(read_fd, "rb")
    writer = os.fdopen(write_fd, "wb")
    return _Supervisor(reader, writer, control_dir).run()


# ---------------------------------------------------------------------------
# Controller-side capture
# ---------------------------------------------------------------------------

class _StreamCapture:
    """Line assembler applying redaction and hard byte/line capture bounds."""

    def __init__(self, stream: str, max_bytes: int, max_lines: int) -> None:
        self.stream = stream
        self.max_bytes = max_bytes
        self.max_lines = max_lines
        self.parts: list[str] = []
        self.byte_count = 0
        self.line_count = 0
        self.truncated = False
        self._pending = ""

    def add_chunk(self, text: str) -> list[str]:
        """Buffer a proxied chunk and return every newly completed raw line."""
        self._pending += text
        lines: list[str] = []
        while "\n" in self._pending:
            line, _, self._pending = self._pending.partition("\n")
            lines.append(line)
        if len(self._pending) > _MAX_CHUNK_BYTES:
            lines.append(self._pending)
            self._pending = ""
        return lines

    def flush(self) -> list[str]:
        if not self._pending:
            return []
        line, self._pending = self._pending, ""
        return [line]

    def record(self, redacted_line: str) -> None:
        """Append one redacted line while enforcing the capture bounds."""
        if self.line_count >= self.max_lines or self.byte_count >= self.max_bytes:
            self.truncated = True
            return
        encoded = len(redacted_line.encode("utf-8")) + 1
        remaining = self.max_bytes - self.byte_count
        if encoded > remaining:
            self.parts.append(redacted_line.encode("utf-8")[:remaining].decode("utf-8", "ignore"))
            self.byte_count = self.max_bytes
            self.line_count += 1
            self.truncated = True
            return
        self.parts.append(redacted_line)
        self.byte_count += encoded
        self.line_count += 1

    def freeze(self) -> CaptureStream:
        text = "\n".join(self.parts)
        if self.truncated:
            text = text + "\n[truncated]"
        return CaptureStream(
            stream=self.stream,
            text=text,
            line_count=self.line_count,
            byte_count=self.byte_count,
            truncated=self.truncated,
        )


# ---------------------------------------------------------------------------
# Controller
# ---------------------------------------------------------------------------

class _Invocation:
    """Controller state machine for one registered, bounded caller invocation."""

    def __init__(
        self,
        spec: InvocationSpec,
        parse_event: Callable[[str, str], Any],
        redact: Optional[Callable[[str], str]],
        cancellation: Any,
        on_started: Callable[[SupervisorLocator], None],
    ) -> None:
        self.spec = spec
        self.parse_event = parse_event
        self.redact = redact
        self.cancellation = cancellation
        self.on_started = on_started
        self.queue: "queue.Queue[Optional[dict[str, Any]]]" = queue.Queue()
        self.captures = {
            name: _StreamCapture(name, spec.max_capture_bytes, spec.max_capture_lines)
            for name in ("stdout", "stderr")
        }
        self.events: list[LifecycleEvent] = []
        self.metrics: list[ParsedMetric] = []
        self.metric_events = 0
        self.first_output_at: Optional[float] = None
        self.stream_eof: set[str] = set()
        self.locator: Optional[SupervisorLocator] = None
        self.reason: Optional[str] = None
        self.external_outcome: Optional[dict[str, Any]] = None
        self.submitted = False
        self.finish_at: Optional[float] = None
        self.idle_at: Optional[float] = None
        self.last_output_at: Optional[float] = None
        self.quiet = False
        self.exited = False
        self.exit_code: Optional[int] = None
        self.signal: Optional[int] = None
        self.run_deadline = 0.0
        self.control_dir: Optional[Path] = None
        self.owns_control_dir = False
        self.supervisor: Optional[subprocess.Popen] = None
        self.err_handle: Optional[Any] = None
        self.to_supervisor: Optional[Any] = None
        self.from_supervisor: Optional[Any] = None
        self.started_at = ""
        self.start_ns = 0

    # -- public entry ------------------------------------------------------

    def run(self) -> InvocationResult:
        _preflight(self.spec)
        self.started_at = _utc_now()
        self.start_ns = time.monotonic_ns()
        self._spawn_supervisor()
        try:
            if self._register_and_start():
                self._pump_until_terminal()
            outcome = self._request_terminal()
        finally:
            self._shutdown_supervisor()
        return self._publish(outcome)

    # -- supervisor process ------------------------------------------------

    def _spawn_supervisor(self) -> None:
        if self.spec.control_dir:
            # Keep a caller-supplied short pathname for the AF_UNIX endpoint.
            # Resolving a containment-preserving alias here can exceed the
            # platform socket-path limit before the supervisor is registered.
            self.control_dir = Path(self.spec.control_dir)
            try:
                self.control_dir.mkdir(mode=0o700)
            except FileExistsError as exc:
                raise LifecycleValidationError(
                    "control_dir must be absent so the invocation can own it exclusively"
                ) from exc
        else:
            self.control_dir = Path(tempfile.mkdtemp(prefix="iop-bench-lifecycle-"))
            self.owns_control_dir = True
        os.chmod(self.control_dir, 0o700)
        controller_read, supervisor_write = os.pipe()
        supervisor_read, controller_write = os.pipe()
        self.err_handle = open(self.control_dir / SUPERVISOR_ERR_FILENAME, "xb")
        argv = [
            sys.executable, "-m", "scripts.agent_benchmark.lifecycle",
            f"--read-fd={supervisor_read}",
            f"--write-fd={supervisor_write}",
            f"--control-dir={self.control_dir}",
        ]
        env = {
            "PATH": os.environ.get("PATH", ""),
            "PYTHONPATH": str(_REPO_ROOT),
            "LANG": os.environ.get("LANG", "C"),
            "TMPDIR": os.environ.get("TMPDIR", tempfile.gettempdir()),
        }
        try:
            self.supervisor = subprocess.Popen(
                argv,
                cwd=str(_REPO_ROOT),
                env=env,
                stdin=subprocess.DEVNULL,
                stdout=subprocess.DEVNULL,
                stderr=self.err_handle,
                pass_fds=(supervisor_read, supervisor_write),
                start_new_session=True,
                close_fds=True,
            )
        finally:
            os.close(supervisor_read)
            os.close(supervisor_write)
        self.to_supervisor = os.fdopen(controller_write, "wb")
        self.from_supervisor = os.fdopen(controller_read, "rb")
        threading.Thread(target=self._frame_reader, daemon=True).start()

    def _frame_reader(self) -> None:
        handle = self.from_supervisor
        try:
            while True:
                frame = _read_frame(handle)
                if frame is None:
                    break
                self.queue.put(frame)
        finally:
            self.queue.put(None)

    def _send(self, frame: dict[str, Any]) -> None:
        handle = self.to_supervisor
        if handle is None:
            return
        try:
            handle.write((json.dumps(frame, ensure_ascii=False) + "\n").encode("utf-8"))
            handle.flush()
        except (BrokenPipeError, ValueError, OSError):
            self.reason = self.reason or REASON_SUPERVISOR_ERROR

    # -- registration gate -------------------------------------------------

    def _register_and_start(self) -> bool:
        self._send({
            "op": "spec",
            "argv": list(self.spec.argv),
            "cwd": self.spec.cwd,
            "env": [list(pair) for pair in self.spec.env],
            "submission_mode": self.spec.submission_mode,
            "task_payload_hex": self.spec.task_payload.hex(),
            "max_capture_bytes": self.spec.max_capture_bytes,
            "cleanup_grace_seconds": self.spec.timeout.cleanup_grace_seconds,
            "fault_injection": self.spec.fault_injection,
        })
        frame = self._await_frame("registered", _REGISTER_TIMEOUT_SECONDS)
        if frame is None:
            self.reason = REASON_SUPERVISOR_ERROR
            return False
        raw = frame.get("locator") or {}
        self.locator = SupervisorLocator(
            supervisor_pid=int(raw.get("supervisor_pid", 0)),
            start_identity=str(raw.get("start_identity", "")),
            socket_path=str(raw.get("socket_path", "")),
            challenge=str(raw.get("challenge", "")),
            control_dir=str(raw.get("control_dir", "")),
            created_at=str(raw.get("created_at", "")),
        )
        try:
            self.on_started(self.locator)
        except Exception:
            self._send({"op": "abort"})
            self.reason = REASON_START_CALLBACK_FAILED
            return False
        self.run_deadline = time.monotonic() + self.spec.timeout.run_seconds
        self._send({"op": "start"})
        return True

    def _await_frame(self, expected: str, timeout: float) -> Optional[dict[str, Any]]:
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            try:
                frame = self.queue.get(timeout=_POLL_INTERVAL_SECONDS)
            except queue.Empty:
                continue
            if frame is None:
                return None
            if frame.get("op") == expected:
                return frame
            if frame.get("op") in ("error", "terminal"):
                return None
        return None

    # -- main pump ---------------------------------------------------------

    def _pump_until_terminal(self) -> None:
        while self.reason is None and self.external_outcome is None:
            self._check_deadlines()
            if self.reason is not None:
                break
            try:
                frame = self.queue.get(timeout=_POLL_INTERVAL_SECONDS)
            except queue.Empty:
                continue
            if frame is None:
                self.reason = self.reason or REASON_SUPERVISOR_ERROR
                break
            self._handle_frame(frame)

    def _handle_frame(self, frame: dict[str, Any]) -> None:
        operation = frame.get("op")
        if operation == "started":
            self._record_submitted(frame)
        elif operation == "output":
            self._handle_output(frame)
        elif operation == "stream_eof":
            self._handle_stream_eof(frame)
        elif operation == "submission_error":
            self.reason = self.reason or REASON_LAUNCH_FAILED
        elif operation == "proxy_truncated":
            self.captures[str(frame.get("stream", "stdout"))].truncated = True
        elif operation == "exited":
            self._handle_exit(frame)
        elif operation == "reader_error":
            self.reason = self.reason or REASON_READER_ERROR
        elif operation == "terminal":
            self.external_outcome = frame.get("outcome") or {}
        elif operation == "error":
            self.reason = self.reason or REASON_SUPERVISOR_ERROR

    def _record_submitted(self, frame: dict[str, Any]) -> None:
        if self.submitted:
            self.reason = self.reason or REASON_DUPLICATE_EVENT
            return
        self.submitted = True
        self._add_event(EVENT_SUBMITTED, SOURCE_HARNESS, "", frame, self.spec.submission_mode)

    def _handle_output(self, frame: dict[str, Any]) -> None:
        stream = str(frame.get("stream", "stdout"))
        capture = self.captures.get(stream)
        if capture is None:
            return
        self.last_output_at = time.monotonic()
        data = str(frame.get("data", ""))
        self._record_first_output(stream, frame, data)
        for line in capture.add_chunk(data):
            self._consume_line(stream, line, frame)

    def _record_first_output(
        self, stream: str, frame: dict[str, Any], data: str
    ) -> None:
        """Record the first non-empty caller frame exactly once, before parsing."""
        if self.first_output_at is not None or not data.strip():
            return
        self.first_output_at = time.monotonic()
        self._add_event(EVENT_FIRST_OUTPUT, SOURCE_HARNESS, stream, frame, "", safe=True)

    def _handle_stream_eof(self, frame: dict[str, Any]) -> None:
        stream = str(frame.get("stream", ""))
        capture = self.captures.get(stream)
        if capture is None or stream in self.stream_eof:
            self.reason = self.reason or REASON_READER_ERROR
            return
        for line in capture.flush():
            self._consume_line(stream, line, frame)
        self.stream_eof.add(stream)

    def _consume_line(self, stream: str, line: str, frame: dict[str, Any]) -> None:
        capture = self.captures[stream]
        redacted = self._redact(line)
        capture.record(redacted)
        try:
            parsed = self.parse_event(stream, line)
        except Exception:
            self.reason = self.reason or REASON_PARSER_ERROR
            return
        self._apply_parsed(parsed, stream, frame, redacted)

    def _apply_parsed(
        self, parsed: Any, stream: str, frame: dict[str, Any], redacted: str
    ) -> None:
        """Apply one parser result: a terminal string, observations, or both."""
        if parsed is None:
            return
        items = parsed if isinstance(parsed, tuple) else (parsed,)
        if not items or len(items) > MAX_PARSED_ITEMS:
            self.reason = self.reason or REASON_MALFORMED_EVENT
            return
        # A failure inside this line stops the rest of the line; a reason
        # latched by an earlier line keeps the pre-existing consume behaviour.
        entry_reason = self.reason
        for item in items:
            if self.reason is not entry_reason:
                return
            self._apply_item(item, stream, frame, redacted)

    def _apply_item(
        self, item: Any, stream: str, frame: dict[str, Any], redacted: str
    ) -> None:
        if isinstance(item, ParsedMetric):
            self._record_metric(item, stream, frame)
            return
        if not isinstance(item, str) or not item:
            self.reason = self.reason or REASON_MALFORMED_EVENT
            return
        if item.startswith(METRIC_PREFIX):
            self._record_metric_label(item, stream, frame, redacted)
            return
        if item not in PARSER_TERMINAL_KINDS:
            self.reason = self.reason or REASON_MALFORMED_EVENT
            return
        self._apply_terminal_evidence(item, stream, frame, redacted)

    def _record_metric_label(
        self, parsed: str, stream: str, frame: dict[str, Any], redacted: str
    ) -> None:
        """Record one untyped caller metric label with no numeric payload."""
        metric_kind = self._validate_metric_kind(parsed)
        if metric_kind is None:
            self.reason = self.reason or REASON_MALFORMED_EVENT
        elif self.metric_events < MAX_METRIC_EVENTS:
            self.metric_events += 1
            self._add_event(metric_kind, SOURCE_CALLER_OUTPUT, stream, frame, redacted)

    def _record_metric(
        self, metric: ParsedMetric, stream: str, frame: dict[str, Any]
    ) -> None:
        """Record one typed observation whose fields are closed and safe."""
        try:
            validated = validate_metric(metric)
            detail = json.dumps(
                metric_record(validated), sort_keys=True, separators=(",", ":")
            )
        except LifecycleMetricError:
            self.reason = self.reason or REASON_MALFORMED_EVENT
            return
        if self.metric_events >= MAX_METRIC_EVENTS:
            # Truncating a typed observation stream could leave a successful
            # terminal that claims a complete measurement.  Retain the first
            # bounded diagnostics, but fail this invocation closed.
            self.reason = self.reason or REASON_MALFORMED_EVENT
            return
        self.metric_events += 1
        self.metrics.append(validated)
        self._add_event(
            METRIC_PREFIX + validated.name, validated.source, stream, frame,
            detail, safe=True,
        )

    def _validate_metric_kind(self, parsed: str) -> Optional[str]:
        if len(parsed) > MAX_METRIC_KIND_CHARS or METRIC_KIND_RE.fullmatch(parsed) is None:
            return None
        return parsed if self._redact(parsed) == parsed else None

    def _apply_terminal_evidence(
        self, kind: str, stream: str, frame: dict[str, Any], redacted: str
    ) -> None:
        now = time.monotonic()
        if kind == EVENT_FINISH:
            if self.finish_at is not None:
                self.reason = self.reason or REASON_DUPLICATE_EVENT
                return
            if self.idle_at is not None:
                self.reason = self.reason or REASON_OUT_OF_ORDER_EVENT
                return
            self.finish_at = now
        else:
            if self.idle_at is not None:
                self.reason = self.reason or REASON_DUPLICATE_EVENT
                return
            if self.finish_at is None:
                self.reason = self.reason or REASON_OUT_OF_ORDER_EVENT
                return
            self.idle_at = now
        self._add_event(kind, SOURCE_CALLER_OUTPUT, stream, frame, redacted)

    def _handle_exit(self, frame: dict[str, Any]) -> None:
        self.exited = True
        raw_code = frame.get("exit_code")
        raw_signal = frame.get("signal")
        self.exit_code = None if raw_code is None else int(raw_code)
        self.signal = None if raw_signal is None else int(raw_signal)
        self._add_event(
            EVENT_EXITED, SOURCE_HARNESS, "", frame,
            f"exit_code={self.exit_code} signal={self.signal}",
        )

    # -- completion policy -------------------------------------------------

    def _check_deadlines(self) -> None:
        now = time.monotonic()
        if _is_cancelled(self.cancellation):
            self.reason = REASON_CANCELLED
            return
        if now >= self.run_deadline:
            self.reason = REASON_TIMED_OUT
            return
        if (
            self.finish_at is not None
            and self.idle_at is None
            and now - self.finish_at >= self.spec.timeout.idle_seconds
        ):
            self.reason = REASON_MISSING_IDLE
            return
        self._check_quiescence(now)
        if (
            self.reason is None
            and self.exited
            and len(self.stream_eof) == len(self.captures)
            and not self.quiet
        ):
            if self.exit_code != 0:
                self.reason = REASON_NONZERO_EXIT
            elif self.idle_at is None:
                self.reason = REASON_MISSING_IDLE

    def _check_quiescence(self, now: float) -> None:
        if self.idle_at is None or self.quiet:
            return
        last_output = self.last_output_at if self.last_output_at is not None else self.idle_at
        if now - last_output < self.spec.timeout.quiet_seconds:
            return
        self.quiet = True
        self._add_event(EVENT_QUIET, SOURCE_HARNESS, "", {"ns": time.monotonic_ns()}, "")
        if self.spec.completion_mode == COMPLETION_STOP_AFTER_IDLE:
            self.reason = REASON_SUCCESS
            return
        if self.exited:
            self.reason = REASON_SUCCESS if self.exit_code == 0 else REASON_NONZERO_EXIT

    # -- terminal handshake ------------------------------------------------

    def _request_terminal(self) -> dict[str, Any]:
        if self.external_outcome is not None:
            self.reason = str(self.external_outcome.get("reason") or REASON_SUPERVISOR_ERROR)
            return self.external_outcome
        reason = self.reason or REASON_SUPERVISOR_ERROR
        self.reason = reason
        self._send({"op": "stop", "reason": reason})
        wait_seconds = (
            self.spec.timeout.cleanup_grace_seconds
            + _KILL_WAIT_SECONDS
            + _TERMINAL_SLACK_SECONDS
        )
        frame = self._await_terminal(wait_seconds, reason)
        if frame is None:
            self.reason = REASON_CLEANUP_FAILED
            return {
                "reason": REASON_CLEANUP_FAILED,
                "exit_code": self.exit_code,
                "signal": self.signal,
                "caller_launched": self.submitted,
                "cleanup_complete": False,
                "process_group_alive": True,
                "receipt_path": "",
            }
        outcome = frame.get("outcome") or {}
        self.reason = str(outcome.get("reason") or reason)
        return outcome

    def _await_terminal(
        self, timeout: float, frozen_reason: str
    ) -> Optional[dict[str, Any]]:
        """Drain all frames preceding terminal while preserving the first reason."""
        deadline = time.monotonic() + timeout
        while time.monotonic() < deadline:
            try:
                frame = self.queue.get(timeout=_POLL_INTERVAL_SECONDS)
            except queue.Empty:
                continue
            if frame is None:
                return None
            if frame.get("op") == "terminal":
                self.external_outcome = frame.get("outcome") or {}
                return frame
            self._handle_frame(frame)
            self.reason = frozen_reason
        return None

    def _shutdown_supervisor(self) -> None:
        for handle in (self.to_supervisor, self.from_supervisor):
            try:
                if handle is not None:
                    handle.close()
            except OSError:
                pass
        supervisor = self.supervisor
        if supervisor is not None:
            try:
                supervisor.wait(timeout=_SUPERVISOR_EXIT_SECONDS)
            except subprocess.TimeoutExpired:
                supervisor.kill()
                try:
                    supervisor.wait(timeout=_SUPERVISOR_EXIT_SECONDS)
                except subprocess.TimeoutExpired:
                    pass
                self.reason = REASON_CLEANUP_FAILED
        if self.err_handle is not None:
            try:
                self.err_handle.close()
            except OSError:
                pass

    # -- evidence ----------------------------------------------------------

    def _redact(self, text: str) -> str:
        if self.redact is not None:
            try:
                text = self.redact(text)
            except Exception:
                text = REDACTED
        return fallback_redact(text)

    def _add_event(
        self, kind: str, source: str, stream: str, frame: dict[str, Any], detail: str,
        *, safe: bool = False,
    ) -> None:
        # ``safe`` details are built from already validated closed fields, so the
        # adapter redactor - which only understands raw caller lines - must not
        # rewrite them.  The fallback secret sweep still applies.
        text = fallback_redact(detail) if safe else self._redact(detail)
        self.events.append(LifecycleEvent(
            kind=kind,
            source=source,
            stream=stream,
            monotonic_ns=time.monotonic_ns(),
            source_monotonic_ns=int(frame.get("ns") or 0),
            observed_at=_utc_now(),
            detail=text[:MAX_EVENT_DETAIL_CHARS],
        ))

    def _publish(self, outcome: dict[str, Any]) -> InvocationResult:
        reason = str(outcome.get("reason") or self.reason or REASON_SUPERVISOR_ERROR)
        self.reason = reason
        cleanup_complete = bool(outcome.get("cleanup_complete"))
        group_alive = bool(outcome.get("process_group_alive"))
        ordered = bool(self.finish_at is not None and self.idle_at is not None and self.quiet)
        success = reason == REASON_SUCCESS and cleanup_complete and not group_alive and ordered
        evidence_dir = Path(self.spec.evidence_dir)
        result = InvocationResult(
            success=success,
            terminal_reason=reason,
            exit_code=self.exit_code if self.exit_code is not None else outcome.get("exit_code"),
            signal=self.signal if self.signal is not None else outcome.get("signal"),
            submitted=self.submitted,
            finish_then_idle_then_quiet=ordered,
            cleanup_complete=cleanup_complete,
            process_group_alive=group_alive,
            events=tuple(self.events),
            stdout=self.captures["stdout"].freeze(),
            stderr=self.captures["stderr"].freeze(),
            journal_path=str(evidence_dir / JOURNAL_FILENAME),
            result_path=str(evidence_dir / RESULT_FILENAME),
            locator=self.locator,
            spec_digest=spec_digest(self.spec),
            started_at=self.started_at,
            ended_at=_utc_now(),
            duration_ns=time.monotonic_ns() - self.start_ns,
            metrics=tuple(self.metrics),
        )
        try:
            _publish_evidence(result, self.spec)
        finally:
            self._discard_owned_control_dir()
        return result

    def _discard_owned_control_dir(self) -> None:
        if self.owns_control_dir and self.control_dir is not None:
            allowed = {
                LOCATOR_FILENAME,
                RECEIPT_FILENAME,
                SUPERVISOR_ERR_FILENAME,
                SOCKET_FILENAME,
            }
            try:
                entries = tuple(self.control_dir.iterdir())
            except OSError:
                return
            if any(entry.name not in allowed for entry in entries):
                return
            shutil.rmtree(self.control_dir, ignore_errors=True)


def _is_cancelled(token: Any) -> bool:
    if token is None:
        return False
    for attribute in ("is_cancelled", "is_set"):
        probe = getattr(token, attribute, None)
        if callable(probe):
            return bool(probe())
    return bool(token() if callable(token) else token)


def _locator_public(locator: Optional[SupervisorLocator]) -> Optional[dict[str, Any]]:
    """Return locator evidence with the challenge marker reduced to a digest."""
    if locator is None:
        return None
    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 _event_record(event: LifecycleEvent) -> dict[str, Any]:
    return {
        "record": "event",
        "kind": event.kind,
        "source": event.source,
        "stream": event.stream,
        "monotonic_ns": event.monotonic_ns,
        "source_monotonic_ns": event.source_monotonic_ns,
        "observed_at": event.observed_at,
        "detail": event.detail,
    }


def _result_record(result: InvocationResult, spec: InvocationSpec) -> dict[str, Any]:
    return {
        "record": "result",
        "success": result.success,
        "terminal_reason": result.terminal_reason,
        "exit_code": result.exit_code,
        "signal": result.signal,
        "submitted": result.submitted,
        "finish_then_idle_then_quiet": result.finish_then_idle_then_quiet,
        "cleanup_complete": result.cleanup_complete,
        "process_group_alive": result.process_group_alive,
        "submission_mode": spec.submission_mode,
        "completion_mode": spec.completion_mode,
        "spec_digest": result.spec_digest,
        "locator": _locator_public(result.locator),
        "started_at": result.started_at,
        "ended_at": result.ended_at,
        "duration_ns": result.duration_ns,
        "stdout": _capture_record(result.stdout),
        "stderr": _capture_record(result.stderr),
        "events": [_event_record(event) for event in result.events],
    }


def _capture_record(capture: CaptureStream) -> dict[str, Any]:
    return {
        "stream": capture.stream,
        "text": capture.text,
        "line_count": capture.line_count,
        "byte_count": capture.byte_count,
        "truncated": capture.truncated,
    }


def _publish_evidence(result: InvocationResult, spec: InvocationSpec) -> None:
    """Publish a no-clobber journal/result pair after terminal cleanup."""
    header = {
        "record": "header",
        "journal_version": JOURNAL_VERSION,
        "spec_digest": result.spec_digest,
        "submission_mode": spec.submission_mode,
        "completion_mode": spec.completion_mode,
        "started_at": result.started_at,
    }
    terminal = {
        "record": "terminal",
        "terminal_reason": result.terminal_reason,
        "success": result.success,
        "cleanup_complete": result.cleanup_complete,
        "process_group_alive": result.process_group_alive,
        "ended_at": result.ended_at,
    }
    lines = [header] + [_event_record(event) for event in result.events] + [terminal]
    journal = "".join(json.dumps(line, ensure_ascii=False) + "\n" for line in lines)
    journal_path = Path(result.journal_path)
    result_path = Path(result.result_path)
    staged: dict[Path, Path] = {}
    published: list[tuple[Path, _FileIdentity]] = []
    try:
        staged[journal_path] = _stage_bytes(
            journal_path.parent, journal.encode("utf-8"), 0o600
        )
        staged[result_path] = _stage_bytes(
            result_path.parent,
            json.dumps(_result_record(result, spec), ensure_ascii=False, indent=2).encode(
                "utf-8"
            ),
            0o600,
        )
        for target in (journal_path, result_path):
            published.append((target, _publish_staged_no_replace(staged[target], target)))
    except OSError as exc:
        for target, identity in reversed(published):
            _rollback_owned(target, identity)
        raise LifecycleError("evidence publication refused an existing target") from exc
    except BaseException:
        for target, identity in reversed(published):
            _rollback_owned(target, identity)
        raise
    finally:
        for stage in staged.values():
            try:
                stage.unlink()
            except FileNotFoundError:
                pass


# ---------------------------------------------------------------------------
# Preflight
# ---------------------------------------------------------------------------

def _preflight(spec: InvocationSpec) -> None:
    """Validate platform and specification before any process is created."""
    if os.name != "posix" or not hasattr(os, "killpg") or not hasattr(socket, "AF_UNIX"):
        raise LifecycleValidationError("bounded lifecycle requires a POSIX platform")
    if sys.version_info < (3, 11):
        raise LifecycleValidationError("bounded lifecycle requires Python 3.11 or newer")
    if not isinstance(spec, InvocationSpec):
        raise LifecycleValidationError("spec must be an InvocationSpec instance")
    if spec.caller_detaches:
        raise LifecycleValidationError(
            "callers that detach from the owned process group are unsupported"
        )
    if spec.fault_injection not in FAULT_MODES:
        raise LifecycleValidationError("fault_injection must be a closed fault mode")
    _preflight_invocation(spec)
    _preflight_bounds(spec)
    _preflight_evidence(spec)


def _preflight_invocation(spec: InvocationSpec) -> None:
    if not isinstance(spec.argv, tuple) or not spec.argv:
        raise LifecycleValidationError("argv must be a non-empty tuple")
    if not all(isinstance(item, str) and item for item in spec.argv):
        raise LifecycleValidationError("argv entries must be non-empty strings")
    if spec.submission_mode not in SUBMISSION_MODES:
        raise LifecycleValidationError(f"submission_mode must be one of {SUBMISSION_MODES}")
    if spec.completion_mode not in COMPLETION_MODES:
        raise LifecycleValidationError(f"completion_mode must be one of {COMPLETION_MODES}")
    if spec.submission_mode == SUBMISSION_ARGV_TASK and spec.task_payload:
        raise LifecycleValidationError("argv_task carries the task in argv and takes no payload")
    if spec.submission_mode == SUBMISSION_STDIN_ONCE and not spec.task_payload:
        raise LifecycleValidationError("stdin_once requires exactly one non-empty task payload")
    if len(spec.task_payload) > MAX_TASK_PAYLOAD_BYTES:
        raise LifecycleValidationError("task payload exceeds the bounded submission size")
    cwd = Path(spec.cwd)
    if not spec.cwd or not cwd.is_dir():
        raise LifecycleValidationError("cwd must be an existing directory")
    if not isinstance(spec.env, tuple):
        raise LifecycleValidationError("env must be a tuple of key/value pairs")
    allowed = set(DEFAULT_ENV_ALLOWLIST) | set(spec.env_allowlist)
    seen_keys: set[str] = set()
    for pair in spec.env:
        if not isinstance(pair, tuple) or len(pair) != 2:
            raise LifecycleValidationError("environment entries must be key/value pairs")
        key, value = pair
        if not isinstance(key, str) or not isinstance(value, str):
            raise LifecycleValidationError("environment keys and values must be strings")
        if not ENV_KEY_RE.match(key):
            raise LifecycleValidationError("environment keys must be POSIX identifiers")
        if key not in allowed:
            raise LifecycleValidationError(f"environment key '{key}' is not allowlisted")
        if key in seen_keys:
            raise LifecycleValidationError("environment keys must be unique")
        seen_keys.add(key)


def _preflight_bounds(spec: InvocationSpec) -> None:
    timeout = spec.timeout
    if not isinstance(timeout, Timeout):
        raise LifecycleValidationError("timeout must be a manifest Timeout instance")
    values = (
        timeout.run_seconds, timeout.idle_seconds,
        timeout.quiet_seconds, timeout.cleanup_grace_seconds,
    )
    if any(not isinstance(v, int) or isinstance(v, bool) or v <= 0 for v in values):
        raise LifecycleValidationError("every timeout bound must be a positive integer")
    if not 0 < spec.max_capture_bytes <= MAX_CAPTURE_BYTES_LIMIT:
        raise LifecycleValidationError("max_capture_bytes is out of bounds")
    if not 0 < spec.max_capture_lines <= MAX_CAPTURE_LINES_LIMIT:
        raise LifecycleValidationError("max_capture_lines is out of bounds")


def _preflight_evidence(spec: InvocationSpec) -> None:
    evidence_dir = Path(spec.evidence_dir)
    if not spec.evidence_dir or not evidence_dir.is_dir():
        raise LifecycleValidationError("evidence_dir must be an existing directory")
    for name in (JOURNAL_FILENAME, RESULT_FILENAME):
        target = evidence_dir / name
        if target.exists() or target.is_symlink():
            raise LifecycleValidationError(
                f"evidence '{name}' already exists and must never be overwritten"
            )
    if spec.control_dir:
        control_dir = Path(spec.control_dir)
        if control_dir.exists() or control_dir.is_symlink():
            raise LifecycleValidationError(
                "control_dir must be absent so the invocation can own it exclusively"
            )
        if not control_dir.parent.is_dir():
            raise LifecycleValidationError("control_dir parent must be an existing directory")


# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------

def run_invocation(
    spec: InvocationSpec,
    *,
    parse_event: Callable[[str, str], Any],
    on_started: Callable[[SupervisorLocator], None],
    redact: Optional[Callable[[str], str]] = None,
    cancellation: Any = None,
) -> InvocationResult:
    """Execute exactly one bounded caller invocation and publish its evidence.

    The caller is launched only after ``on_started`` durably commits the
    supervisor locator. Terminal outcome, owned-process-group cleanup and
    atomic evidence publication happen on every return path.

    Args:
        spec: Frozen invocation specification.
        parse_event: Adapter parser mapping ``(stream, line)`` to ``None``,
            ``"finish"``, ``"idle"``, a ``"metric:<name>"`` label, a typed
            ``ParsedMetric``, or a bounded tuple of those items when one caller
            line carries observations and terminal evidence together.
        on_started: Required durable locator commit callback.
        redact: Optional adapter redactor for exact secret values.
        cancellation: Optional cancellation token, event or predicate.

    Returns:
        Frozen InvocationResult.

    Raises:
        LifecycleValidationError: If platform or specification preflight fails.
    """
    if not callable(parse_event):
        raise LifecycleValidationError("parse_event must be callable")
    if not callable(on_started):
        raise LifecycleValidationError("on_started must be callable")
    return _Invocation(spec, parse_event, redact, cancellation, on_started).run()


def recover_invocation(
    locator: SupervisorLocator, stop: bool = True
) -> TerminalOutcome:
    """Authenticate a recorded supervisor and request status or bounded cleanup.

    Authentication is the marker challenge over the recorded control endpoint.
    Process identity is corroboration only and never authorizes a signal.

    Raises:
        LifecycleRecoveryError: If the locator is stale, forged or mismatched.
    """
    _validate_locator_endpoint(locator)
    connection = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
    connection.settimeout(_CONTROL_SOCKET_TIMEOUT_SECONDS)
    try:
        try:
            connection.connect(locator.socket_path)
        except OSError as exc:
            raise LifecycleRecoveryError("supervisor control endpoint is stale") from exc
        stream = connection.makefile("rwb")
        _authenticate(stream, locator)
        if not stop:
            _send_json(stream, {"op": "status"})
            reply = _read_frame(stream) or {}
            if not reply.get("ok"):
                raise LifecycleRecoveryError("supervisor refused the status request")
            status = reply.get("status") or {}
            return TerminalOutcome(
                reason=str(status.get("terminal_reason") or ""),
                exit_code=None,
                signal=None,
                caller_launched=bool(status.get("caller_launched")),
                cleanup_complete=not bool(status.get("process_group_alive")),
                process_group_alive=bool(status.get("process_group_alive")),
                receipt_path="",
            )
        _send_json(stream, {"op": "stop", "reason": REASON_RECOVERED_STOP})
        reply = _read_frame(stream) or {}
        if not reply.get("ok"):
            raise LifecycleRecoveryError("supervisor refused the cleanup request")
        outcome = reply.get("outcome") or {}
    finally:
        connection.close()
    _verify_receipt(locator, outcome)
    return TerminalOutcome(
        reason=str(outcome.get("reason") or REASON_RECOVERED_STOP),
        exit_code=outcome.get("exit_code"),
        signal=outcome.get("signal"),
        caller_launched=bool(outcome.get("caller_launched")),
        cleanup_complete=bool(outcome.get("cleanup_complete")),
        process_group_alive=bool(outcome.get("process_group_alive")),
        receipt_path=str(outcome.get("receipt_path") or ""),
    )


def _validate_locator_endpoint(locator: SupervisorLocator) -> None:
    if not isinstance(locator, SupervisorLocator):
        raise LifecycleRecoveryError("locator must be a SupervisorLocator instance")
    if not locator.challenge or not locator.socket_path:
        raise LifecycleRecoveryError("locator is missing its authenticated endpoint")
    path = Path(locator.socket_path)
    if not path.is_socket():
        raise LifecycleRecoveryError("locator socket is missing or not a socket")
    live_identity = _process_start_identity(locator.supervisor_pid)
    if live_identity and locator.start_identity and live_identity != locator.start_identity:
        raise LifecycleRecoveryError("supervisor start identity does not match the locator")


def _authenticate(stream: Any, locator: SupervisorLocator) -> None:
    _send_json(stream, {"op": "auth", "challenge": locator.challenge})
    reply = _read_frame(stream) or {}
    if not reply.get("ok"):
        raise LifecycleRecoveryError("supervisor challenge authentication failed")
    if int(reply.get("supervisor_pid", -1)) != locator.supervisor_pid:
        raise LifecycleRecoveryError("supervisor pid does not match the locator")
    if str(reply.get("start_identity", "")) != locator.start_identity:
        raise LifecycleRecoveryError("supervisor start identity does not match the locator")


def _verify_receipt(locator: SupervisorLocator, outcome: dict[str, Any]) -> None:
    receipt_path = Path(str(outcome.get("receipt_path") or ""))
    if not receipt_path.is_file():
        raise LifecycleRecoveryError("cleanup receipt is missing")
    try:
        receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise LifecycleRecoveryError("cleanup receipt is unreadable") from exc
    expected = hashlib.sha256(locator.challenge.encode("utf-8")).hexdigest()
    if receipt.get("challenge_digest") != expected:
        raise LifecycleRecoveryError("cleanup receipt does not match the locator")
    if not receipt.get("cleanup_complete") or receipt.get("process_group_alive"):
        raise LifecycleRecoveryError("cleanup receipt does not prove owned-group cleanup")


def read_locator(control_dir: str | Path) -> SupervisorLocator:
    """Load a durably registered locator from a supervisor control directory."""
    raw = json.loads((Path(control_dir) / LOCATOR_FILENAME).read_text(encoding="utf-8"))
    return SupervisorLocator(
        supervisor_pid=int(raw["supervisor_pid"]),
        start_identity=str(raw["start_identity"]),
        socket_path=str(raw["socket_path"]),
        challenge=str(raw["challenge"]),
        control_dir=str(raw["control_dir"]),
        created_at=str(raw["created_at"]),
    )


if __name__ == "__main__":
    sys.exit(_supervisor_main(sys.argv[1:]))