#!/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 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_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" 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}$") 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_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.""" # --------------------------------------------------------------------------- # 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 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: 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 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 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.metric_events = 0 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() for line in capture.add_chunk(str(frame.get("data", ""))): self._consume_line(stream, line, frame) 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: if parsed is None: return if not isinstance(parsed, str) or not parsed: self.reason = self.reason or REASON_MALFORMED_EVENT return if parsed.startswith(METRIC_PREFIX): 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) return if parsed not in PARSER_TERMINAL_KINDS: self.reason = self.reason or REASON_MALFORMED_EVENT return self._apply_terminal_evidence(parsed, stream, frame, redacted) 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 ) -> None: 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=self._redact(detail)[: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, ) 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"`` or a ``"metric:"`` data event. 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:]))