iop/scripts/agent_benchmark/lifecycle_test.py

"""Deterministic subprocess coverage for the bounded benchmark lifecycle."""

from __future__ import annotations

import json
import os
import select
import signal
import subprocess
import sys
import tempfile
import threading
import time
import unittest
from dataclasses import replace
from pathlib import Path

from scripts.agent_benchmark.lifecycle import (
    CALLER_REASON_ERROR,
    CALLER_REASON_SUCCESS,
    CALLER_STATUS_FAILED,
    CALLER_STATUS_SUCCEEDED,
    COMPLETION_EXIT_AFTER_IDLE,
    COMPLETION_STOP_AFTER_IDLE,
    REASON_CANCELLED,
    REASON_CLEANUP_FAILED,
    REASON_CONTROLLER_LOST,
    REASON_DUPLICATE_EVENT,
    REASON_MALFORMED_EVENT,
    REASON_MISSING_IDLE,
    REASON_NONZERO_EXIT,
    REASON_OUT_OF_ORDER_EVENT,
    REASON_PARSER_ERROR,
    REASON_READER_ERROR,
    REASON_RECOVERED_STOP,
    REASON_START_CALLBACK_FAILED,
    REASON_TIMED_OUT,
    MAX_METRIC_EVENTS,
    SUBMISSION_ARGV_TASK,
    SUBMISSION_STDIN_ONCE,
    CancellationToken,
    CallerEvent,
    CallerTerminal,
    HarnessOutcome,
    InvocationSpec,
    LifecycleError,
    LifecycleRecoveryError,
    LifecycleValidationError,
    ParsedMetric,
    SupervisorLocator,
    count_metric,
    duration_metric,
    env_pairs,
    exact_value_redactor,
    read_locator,
    recover_invocation,
    run_invocation,
)
from scripts.agent_benchmark.manifest import Timeout


def _events(_: str, line: str):
    return {
        "FINISH": (
            CallerTerminal(CALLER_STATUS_SUCCEEDED, CALLER_REASON_SUCCESS),
            CallerEvent("finish"),
        ),
        "IDLE": CallerEvent("idle"),
    }.get(line)


class LifecycleTest(unittest.TestCase):
    """Each test owns a temporary evidence directory and real process group."""

    def setUp(self) -> None:
        self.tmp = tempfile.TemporaryDirectory()
        self.root = Path(self.tmp.name)

    def tearDown(self) -> None:
        self.tmp.cleanup()

    def _spec(
        self,
        source: str,
        *,
        submission_mode: str = SUBMISSION_ARGV_TASK,
        completion_mode: str = COMPLETION_EXIT_AFTER_IDLE,
        payload: bytes = b"",
        run_seconds: int = 5,
        max_capture_bytes: int = 4096,
        max_capture_lines: int = 100,
        fault_injection: str = "",
    ) -> InvocationSpec:
        return InvocationSpec(
            argv=(sys.executable, "-u", "-c", source),
            cwd=str(self.root),
            env=env_pairs({"PATH": os.environ.get("PATH", "/usr/bin:/bin")}),
            submission_mode=submission_mode,
            completion_mode=completion_mode,
            timeout=Timeout(run_seconds, 2, 1, 1),
            evidence_dir=str(self.root),
            task_payload=payload,
            max_capture_bytes=max_capture_bytes,
            max_capture_lines=max_capture_lines,
            fault_injection=fault_injection,
        )

    def _run(self, spec: InvocationSpec, **kwargs: object):
        return run_invocation(spec, parse_event=_events, on_started=lambda _: None, **kwargs)

    def _start_supervisor(self, control_dir: Path):
        control_dir.mkdir()
        controller_read, supervisor_write = os.pipe()
        supervisor_read, controller_write = os.pipe()
        process = subprocess.Popen(
            (
                sys.executable,
                "-m",
                "scripts.agent_benchmark.lifecycle",
                f"--read-fd={supervisor_read}",
                f"--write-fd={supervisor_write}",
                f"--control-dir={control_dir}",
            ),
            cwd=str(Path(__file__).resolve().parents[2]),
            env={
                "PATH": os.environ.get("PATH", "/usr/bin:/bin"),
                "PYTHONPATH": str(Path(__file__).resolve().parents[2]),
            },
            pass_fds=(supervisor_read, supervisor_write),
            start_new_session=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        os.close(supervisor_read)
        os.close(supervisor_write)
        return (
            process,
            os.fdopen(controller_write, "wb", buffering=0),
            os.fdopen(controller_read, "rb", buffering=0),
        )

    def _send_supervisor_spec(self, writer: object, source: str) -> None:
        self._write_frame(writer, {
            "op": "spec",
            "argv": [sys.executable, "-u", "-c", source],
            "cwd": str(self.root),
            "env": [["PATH", os.environ.get("PATH", "/usr/bin:/bin")]],
            "submission_mode": SUBMISSION_ARGV_TASK,
            "task_payload_hex": "",
            "max_capture_bytes": 1024,
            "cleanup_grace_seconds": 1,
            "fault_injection": "",
        })

    def _close_supervisor(
        self, process: subprocess.Popen, writer: object, reader: object
    ) -> None:
        if process.poll() is None:
            for candidate in self.root.glob("*-control/locator.json"):
                if candidate.is_file():
                    try:
                        recover_invocation(read_locator(candidate.parent))
                    except LifecycleRecoveryError:
                        pass
                    if process.poll() is not None:
                        break
        for handle in (writer, reader):
            try:
                handle.close()  # type: ignore[attr-defined]
            except OSError:
                pass
        if process.poll() is None:
            process.kill()
            process.wait(timeout=5)

    def test_exit_after_idle_publishes_ordered_atomic_evidence(self) -> None:
        result = self._run(self._spec("print('FINISH'); print('IDLE')"))

        self.assertTrue(result.product.status == "succeeded")
        self.assertTrue(result.harness.cleanup_complete)
        self.assertFalse(result.process_group_alive)
        self.assertTrue(result.harness.ordered_terminal)
        self.assertEqual([event.kind for event in result.events], [
            "submitted", "first_output", "caller_terminal", "finish", "idle",
            "exited", "quiet",
        ])
        self.assertTrue(Path(result.journal_path).is_file())
        published = json.loads(Path(result.result_path).read_text(encoding="utf-8"))
        self.assertNotIn("argv", published)
        self.assertNotIn("env", published)

    def test_stdin_once_submits_exactly_once_and_closes_input(self) -> None:
        source = "import sys; print(sys.stdin.read()); print('FINISH'); print('IDLE')"
        result = self._run(self._spec(
            source,
            submission_mode=SUBMISSION_STDIN_ONCE,
            payload=b"single task payload",
        ))

        self.assertTrue(result.product.status == "succeeded")
        self.assertIn("single task payload", result.stdout.text)
        self.assertEqual(sum(event.kind == "submitted" for event in result.events), 1)

    def test_stdin_once_non_reader_times_out_and_cleans_group(self) -> None:
        started = time.monotonic()
        result = self._run(self._spec(
            "import time; time.sleep(30)",
            submission_mode=SUBMISSION_STDIN_ONCE,
            payload=b"x" * (1 << 20),
            run_seconds=1,
        ))

        self.assertLess(time.monotonic() - started, 6)
        self.assertEqual(result.harness.reason, REASON_TIMED_OUT)
        self.assertFalse(result.submitted)
        self.assertEqual(sum(event.kind == "submitted" for event in result.events), 0)
        self.assertTrue(result.harness.cleanup_complete)
        self.assertFalse(result.process_group_alive)

    def test_unterminated_final_idle_is_consumed_before_terminal(self) -> None:
        source = "import sys; sys.stdout.write('FINISH\\nIDLE'); sys.stdout.flush()"
        for index, mode in enumerate(
            (COMPLETION_EXIT_AFTER_IDLE, COMPLETION_STOP_AFTER_IDLE)
        ):
            with self.subTest(completion_mode=mode):
                evidence = self.root / f"unterminated-{index}"
                evidence.mkdir()
                result = self._run(replace(
                    self._spec(source, completion_mode=mode),
                    evidence_dir=str(evidence),
                ))

                self.assertTrue(result.product.status == "succeeded")
                kinds = [event.kind for event in result.events]
                self.assertLess(kinds.index("finish"), kinds.index("idle"))
                self.assertLess(kinds.index("idle"), kinds.index("quiet"))
                records = [
                    json.loads(line)
                    for line in Path(result.journal_path).read_text(encoding="utf-8").splitlines()
                ]
                self.assertEqual(sum(record.get("record") == "terminal" for record in records), 1)
                self.assertEqual(records[-1]["record"], "terminal")

    def test_stop_after_idle_gracefully_stops_live_caller(self) -> None:
        started = time.monotonic()
        result = self._run(self._spec(
            "import time; print('FINISH'); print('IDLE'); time.sleep(30)",
            completion_mode=COMPLETION_STOP_AFTER_IDLE,
        ))

        self.assertTrue(result.product.status == "succeeded")
        self.assertLess(time.monotonic() - started, 8)
        self.assertTrue(result.harness.cleanup_complete)
        self.assertFalse(result.process_group_alive)

    def test_caller_output_cannot_synthesize_submission(self) -> None:
        result = self._run(self._spec("print('submitted'); print('FINISH'); print('IDLE')"))

        self.assertTrue(result.product.status == "succeeded")
        self.assertEqual(sum(event.kind == "submitted" for event in result.events), 1)
        self.assertEqual(result.events[0].source, "harness")

    def test_invalid_terminal_sequences_fail_closed(self) -> None:
        cases = (
            ("print('IDLE')", REASON_OUT_OF_ORDER_EVENT),
            ("print('FINISH'); print('FINISH')", REASON_DUPLICATE_EVENT),
            ("print('FINISH')", REASON_MISSING_IDLE),
        )
        for source, reason in cases:
            with self.subTest(reason=reason):
                evidence = self.root / reason
                evidence.mkdir()
                spec = replace(self._spec(source), evidence_dir=str(evidence))
                result = self._run(spec)
                self.assertFalse(result.product.status == "succeeded")
                self.assertEqual(result.harness.reason, reason)
                self.assertTrue(result.harness.cleanup_complete)

    def test_malformed_parser_and_nonzero_exit_fail_closed(self) -> None:
        malformed = run_invocation(
            self._spec("print('UNKNOWN')"),
            parse_event=lambda _stream, _line: "submitted",
            on_started=lambda _: None,
        )
        self.assertEqual(malformed.harness.reason, REASON_MALFORMED_EVENT)
        self.assertTrue(malformed.harness.cleanup_complete)

        evidence = self.root / "nonzero"
        evidence.mkdir()
        failed = self._run(replace(
            self._spec("print('FINISH'); print('IDLE'); raise SystemExit(7)"),
            evidence_dir=str(evidence),
        ))
        self.assertEqual(failed.product.status, "succeeded")
        self.assertEqual(failed.harness.status, "passed")
        self.assertEqual(failed.process.exit_code, 7)
        self.assertTrue(failed.harness.cleanup_complete)

    def test_product_error_can_have_clean_harness_and_process(self) -> None:
        def parse_error(_stream: str, line: str):
            return {
                "ERROR": (
                    CallerTerminal(CALLER_STATUS_FAILED, CALLER_REASON_ERROR),
                    CallerEvent("finish"),
                ),
                "IDLE": CallerEvent("idle"),
            }.get(line)

        result = run_invocation(
            self._spec("print('ERROR'); print('IDLE')"),
            parse_event=parse_error,
            on_started=lambda _: None,
        )

        self.assertEqual((result.product.status, result.product.reason), (
            "failed", CALLER_REASON_ERROR,
        ))
        self.assertEqual((result.harness.status, result.harness.reason), (
            "passed", "success",
        ))
        self.assertEqual((result.process.status, result.process.exit_code), (
            "exited", 0,
        ))

        with self.assertRaises(LifecycleValidationError):
            HarnessOutcome("failed", "success", True, True)
        with self.assertRaises(LifecycleValidationError):
            HarnessOutcome("passed", "success", False, True)

    def test_parser_failure_leaves_product_unknown(self) -> None:
        def broken_parser(_stream: str, _line: str):
            raise ValueError("synthetic parser failure")

        result = run_invocation(
            self._spec("print('BROKEN')"),
            parse_event=broken_parser,
            on_started=lambda _: None,
        )

        self.assertEqual((result.product.status, result.product.reason), (
            "unknown", "unavailable",
        ))
        self.assertEqual((result.harness.status, result.harness.reason), (
            "failed", REASON_PARSER_ERROR,
        ))

    def test_timeout_cancel_and_cleanup_do_not_fabricate_product(self) -> None:
        timeout = self._run(self._spec("import time; time.sleep(30)", run_seconds=1))

        token = CancellationToken()
        timer = threading.Timer(0.2, token.cancel)
        timer.start()
        try:
            cancel_root = self.root / "independent-cancel"
            cancel_root.mkdir()
            cancelled = self._run(
                replace(
                    self._spec("import time; time.sleep(30)"),
                    evidence_dir=str(cancel_root),
                ),
                cancellation=token,
            )
        finally:
            timer.cancel()

        cleanup_root = self.root / "independent-cleanup"
        cleanup_root.mkdir()
        cleanup_control = self.root / "independent-cleanup-control"

        def collide_receipt(locator: SupervisorLocator) -> None:
            (Path(locator.control_dir) / "cleanup-receipt.json").write_bytes(
                b"collision"
            )

        cleanup = run_invocation(
            replace(
                self._spec("print('FINISH'); print('IDLE')"),
                evidence_dir=str(cleanup_root),
                control_dir=str(cleanup_control),
            ),
            parse_event=_events,
            on_started=collide_receipt,
        )

        for result, product_status, process_status, reason in (
            (timeout, "unknown", "timed_out", REASON_TIMED_OUT),
            (cancelled, "unknown", "cancelled", REASON_CANCELLED),
            (cleanup, "succeeded", "exited", REASON_CLEANUP_FAILED),
        ):
            with self.subTest(reason=reason):
                self.assertEqual(result.product.status, product_status)
                self.assertEqual(result.harness.reason, reason)
                self.assertEqual(result.process.status, process_status)

    def test_timeout_cancel_and_reader_error_all_cleanup(self) -> None:
        timeout = self._run(self._spec("import time; time.sleep(30)", run_seconds=1))
        self.assertEqual(timeout.harness.reason, REASON_TIMED_OUT)
        self.assertTrue(timeout.harness.cleanup_complete)

        token = CancellationToken()
        timer = threading.Timer(0.2, token.cancel)
        timer.start()
        try:
            evidence = self.root / "cancel"
            evidence.mkdir()
            cancelled = self._run(replace(
                self._spec("import time; time.sleep(30)"), evidence_dir=str(evidence)),
                cancellation=token,
            )
        finally:
            timer.cancel()
        self.assertEqual(cancelled.harness.reason, REASON_CANCELLED)
        self.assertTrue(cancelled.harness.cleanup_complete)

        evidence = self.root / "reader"
        evidence.mkdir()
        reader_error = self._run(replace(
            self._spec("print('FINISH'); print('IDLE')", fault_injection="reader_error"),
            evidence_dir=str(evidence),
        ))
        self.assertEqual(reader_error.harness.reason, REASON_READER_ERROR)
        self.assertTrue(reader_error.harness.cleanup_complete)

    def test_redaction_and_capture_bounds_apply_before_publication(self) -> None:
        secret = "EXACT_SECRET_123456789"
        source = (
            f"print('{secret}'); print('Authorization Bearer fallback-token-123456789'); "
            "print('FINISH'); print('IDLE')"
        )
        result = self._run(
            self._spec(source, max_capture_bytes=4096, max_capture_lines=3),
            redact=exact_value_redactor((secret,)),
        )

        evidence = Path(result.result_path).read_text(encoding="utf-8")
        self.assertNotIn(secret, evidence)
        self.assertNotIn("fallback-token-123456789", evidence)
        self.assertIn("[redacted]", evidence)
        self.assertTrue(result.stdout.truncated)

    def test_concurrent_evidence_collision_preserves_existing_files(self) -> None:
        occupied_control = self.root / "occupied-control"
        occupied_control.mkdir()
        locator_sentinel = occupied_control / "locator.json"
        locator_sentinel.write_bytes(b"locator-sentinel")
        with self.assertRaises(LifecycleValidationError):
            self._run(replace(
                self._spec("print('FINISH'); print('IDLE')"),
                control_dir=str(occupied_control),
            ))
        self.assertEqual(locator_sentinel.read_bytes(), b"locator-sentinel")

        racing_control = self.root / "racing-locator-control"
        process, writer, reader = self._start_supervisor(racing_control)
        racing_locator = racing_control / "locator.json"
        racing_locator.write_bytes(b"concurrent-locator")
        try:
            self._send_supervisor_spec(writer, "print('FINISH'); print('IDLE')")
            frames = []
            while True:
                frame = self._read_frame(reader, 5)
                frames.append(frame)
                if frame.get("op") == "terminal":
                    break
            self.assertEqual(process.wait(timeout=8), 1)
            self.assertTrue(any(frame.get("op") == "error" for frame in frames))
        finally:
            self._close_supervisor(process, writer, reader)
        self.assertEqual(racing_locator.read_bytes(), b"concurrent-locator")

        for index, collision_name in enumerate(
            ("lifecycle-journal.jsonl", "lifecycle-result.json")
        ):
            with self.subTest(collision=collision_name):
                evidence = self.root / f"collision-{index}"
                evidence.mkdir()
                control = self.root / f"collision-control-{index}"
                sentinel = evidence / collision_name
                unrelated = evidence / "unrelated.txt"
                sentinel_bytes = f"sentinel-{index}".encode()
                unrelated.write_bytes(b"unrelated")

                def collide(_: SupervisorLocator) -> None:
                    sentinel.write_bytes(sentinel_bytes)

                with self.assertRaises(LifecycleError):
                    run_invocation(
                        replace(
                            self._spec("print('FINISH'); print('IDLE')"),
                            evidence_dir=str(evidence),
                            control_dir=str(control),
                        ),
                        parse_event=_events,
                        on_started=collide,
                    )

                self.assertEqual(sentinel.read_bytes(), sentinel_bytes)
                self.assertEqual(unrelated.read_bytes(), b"unrelated")
                other_name = (
                    "lifecycle-result.json"
                    if collision_name.endswith("jsonl")
                    else "lifecycle-journal.jsonl"
                )
                self.assertFalse((evidence / other_name).exists())

        receipt_evidence = self.root / "receipt-collision-evidence"
        receipt_evidence.mkdir()
        receipt_control = self.root / "receipt-collision-control"
        receipt_sentinel = b"receipt-sentinel"

        def collide_receipt(locator: SupervisorLocator) -> None:
            (Path(locator.control_dir) / "cleanup-receipt.json").write_bytes(
                receipt_sentinel
            )

        receipt_result = run_invocation(
            replace(
                self._spec("print('FINISH'); print('IDLE')"),
                evidence_dir=str(receipt_evidence),
                control_dir=str(receipt_control),
            ),
            parse_event=_events,
            on_started=collide_receipt,
        )
        self.assertEqual(receipt_result.harness.reason, REASON_CLEANUP_FAILED)
        self.assertEqual(receipt_result.product.status, "succeeded")
        self.assertFalse(receipt_result.harness.cleanup_complete)
        self.assertFalse(receipt_result.process_group_alive)
        self.assertEqual(
            (receipt_control / "cleanup-receipt.json").read_bytes(), receipt_sentinel
        )
        published = json.loads(Path(receipt_result.result_path).read_text(encoding="utf-8"))
        self.assertEqual(published["harness"]["reason"], REASON_CLEANUP_FAILED)
        journal = [
            json.loads(line)
            for line in Path(receipt_result.journal_path).read_text(encoding="utf-8").splitlines()
        ]
        terminals = [record for record in journal if record.get("record") == "terminal"]
        self.assertEqual(len(terminals), 1)
        self.assertEqual(terminals[0]["harness"]["reason"], REASON_CLEANUP_FAILED)

    def test_metric_kind_cannot_leak_secret(self) -> None:
        cases = (
            ("exact_secret_123456789", exact_value_redactor(("exact_secret_123456789",))),
            ("iop_abcdefghijklmnop", None),
            ("a" * 65, None),
        )
        for index, (metric_name, redactor) in enumerate(cases):
            with self.subTest(metric=metric_name[:24]):
                evidence = self.root / f"metric-invalid-{index}"
                evidence.mkdir()

                def parse_metric(_stream: str, line: str) -> str | None:
                    terminal = _events(_stream, line)
                    return terminal if terminal is not None else f"metric:{line}"

                result = run_invocation(
                    replace(
                        self._spec(
                            f"print({metric_name!r}); print('FINISH'); print('IDLE')"
                        ),
                        evidence_dir=str(evidence),
                    ),
                    parse_event=parse_metric,
                    on_started=lambda _: None,
                    redact=redactor,
                )
                self.assertEqual(result.harness.reason, REASON_MALFORMED_EVENT)
                self.assertFalse(any(event.kind == f"metric:{metric_name}" for event in result.events))
                persisted = "\n".join(
                    path.read_text(encoding="utf-8")
                    for path in (Path(result.journal_path), Path(result.result_path))
                )
                if index < 2:
                    self.assertNotIn(metric_name, persisted)

        evidence = self.root / "metric-valid"
        evidence.mkdir()

        def parse_valid(_stream: str, line: str) -> str | None:
            terminal = _events(_stream, line)
            return terminal if terminal is not None else f"metric:{line}"

        valid = run_invocation(
            replace(
                self._spec("print('duration_ms'); print('FINISH'); print('IDLE')"),
                evidence_dir=str(evidence),
            ),
            parse_event=parse_valid,
            on_started=lambda _: None,
        )
        self.assertEqual(valid.product.status, "unknown")
        self.assertEqual(valid.harness.reason, REASON_MALFORMED_EVENT)
        self.assertNotIn("metric:duration_ms", [event.kind for event in valid.events])

    def test_first_output_is_recorded_once_before_terminal_evidence(self) -> None:
        source = (
            "import sys; sys.stdout.write('   \\n'); sys.stdout.flush(); "
            "print('chatter'); print('more chatter'); print('FINISH'); print('IDLE')"
        )
        result = self._run(self._spec(source))

        self.assertTrue(result.product.status == "succeeded")
        kinds = [event.kind for event in result.events]
        self.assertEqual(kinds.count("first_output"), 1)
        self.assertLess(kinds.index("submitted"), kinds.index("first_output"))
        self.assertLess(kinds.index("first_output"), kinds.index("finish"))
        first_output = result.events[kinds.index("first_output")]
        # The instant is the harness observation of a caller frame, so its
        # source is the harness and its stream is the observed caller stream.
        self.assertEqual((first_output.source, first_output.stream), ("harness", "stdout"))
        self.assertGreater(first_output.monotonic_ns, 0)

    def test_silent_caller_records_no_first_output(self) -> None:
        result = self._run(self._spec("import time; time.sleep(30)", run_seconds=1))
        self.assertEqual(result.harness.reason, REASON_TIMED_OUT)
        self.assertNotIn("first_output", [event.kind for event in result.events])

    def test_typed_observations_are_published_with_terminal_evidence(self) -> None:
        def parse_metric(_stream: str, line: str) -> object:
            if line.strip() != "REPORT":
                return _events(_stream, line)
            return (
                duration_metric("total_duration", "12.5", model="claude-sonnet"),
                count_metric("input_tokens", 11, model="claude-sonnet"),
            )

        result = run_invocation(
            self._spec("print('REPORT'); print('FINISH'); print('IDLE')"),
            parse_event=parse_metric,
            on_started=lambda _: None,
        )
        self.assertTrue(result.product.status == "succeeded")
        self.assertEqual(
            [(metric.name, metric.value) for metric in result.metrics],
            [("total_duration", 12_500_000), ("input_tokens", 11)],
        )
        kinds = [event.kind for event in result.events]
        self.assertEqual(kinds.count("metric:total_duration"), 1)
        journal = Path(result.journal_path).read_text(encoding="utf-8")
        self.assertIn('\\"clock\\":\\"caller_reported\\"', journal)
        self.assertIn('\\"model\\":\\"claude-sonnet\\"', journal)
        self.assertIn('\\"unit\\":\\"tokens\\"', journal)

    def test_invalid_or_oversized_observation_sets_fail_closed(self) -> None:
        forged = ParsedMetric(
            "total_duration", 5, "tokens", "caller_reported", "caller_output"
        )
        cases = {
            "wrong-unit": forged,
            "empty-set": (),
            "oversized-set": tuple(
                count_metric("input_tokens", index) for index in range(17)
            ),
            "nested-set": ((count_metric("input_tokens", 1),),),
        }
        for name, parsed in cases.items():
            with self.subTest(name=name):
                evidence = self.root / f"observation-{name}"
                evidence.mkdir()
                result = run_invocation(
                    replace(
                        self._spec("print('REPORT'); print('FINISH'); print('IDLE')"),
                        evidence_dir=str(evidence),
                    ),
                    parse_event=lambda _stream, line, parsed=parsed: (
                        parsed if line.strip() == "REPORT" else _events(_stream, line)
                    ),
                    on_started=lambda _: None,
                )
                self.assertEqual(result.harness.reason, REASON_MALFORMED_EVENT)
                self.assertEqual(result.metrics, ())

    def test_metric_event_overflow_is_malformed_after_retaining_the_bound(self) -> None:
        def parse_metric(_stream: str, line: str) -> object:
            if line == "REPORT":
                return count_metric("input_tokens", 1)
            return _events(_stream, line)

        reports = "\n".join(["print('REPORT')"] * (MAX_METRIC_EVENTS + 1))
        result = run_invocation(
            self._spec(reports + "; print('FINISH'); print('IDLE')"),
            parse_event=parse_metric,
            on_started=lambda _: None,
        )
        self.assertEqual(result.harness.reason, REASON_MALFORMED_EVENT)
        self.assertEqual(len(result.metrics), MAX_METRIC_EVENTS)

    def test_adapter_redactor_cannot_corrupt_a_validated_observation(self) -> None:
        secret = "EXACT_SECRET_123456789"

        def parse_metric(_stream: str, line: str) -> object:
            if line.strip().startswith("REPORT"):
                return (count_metric("output_tokens", 22, model="claude-sonnet"), )
            return _events(_stream, line)

        result = run_invocation(
            self._spec(f"print('REPORT {secret}'); print('FINISH'); print('IDLE')"),
            parse_event=parse_metric,
            on_started=lambda _: None,
            # A structural redactor that rewrites every raw caller line must not
            # rewrite a detail built from already validated closed fields.
            redact=lambda _line: "[structural]",
        )
        self.assertTrue(result.product.status == "succeeded")
        detail = next(
            event.detail for event in result.events if event.kind == "metric:output_tokens"
        )
        self.assertEqual(json.loads(detail)["value"], 22)
        self.assertNotIn(secret, Path(result.result_path).read_text(encoding="utf-8"))

    def test_callback_failure_launches_no_caller_and_persists_failure(self) -> None:
        marker = self.root / "caller-ran"
        spec = self._spec(f"from pathlib import Path; Path({str(marker)!r}).write_text('ran')")

        result = run_invocation(
            spec,
            parse_event=_events,
            on_started=lambda _locator: (_ for _ in ()).throw(RuntimeError("durable write failed")),
        )

        self.assertEqual(result.harness.reason, REASON_START_CALLBACK_FAILED)
        self.assertFalse(result.submitted)
        self.assertFalse(marker.exists())
        self.assertTrue(result.harness.cleanup_complete)

    def test_forged_live_locator_refuses_recovery(self) -> None:
        checked: list[SupervisorLocator] = []

        def verify(locator: SupervisorLocator) -> None:
            self.assertEqual(read_locator(locator.control_dir), locator)
            checked.append(locator)
            with self.assertRaises(LifecycleRecoveryError):
                recover_invocation(replace(locator, challenge="forged-marker"))

        result = run_invocation(
            self._spec("print('FINISH'); print('IDLE')"),
            parse_event=_events,
            on_started=verify,
        )
        self.assertTrue(checked)
        self.assertTrue(result.product.status == "succeeded")

    def test_control_socket_bind_supports_short_symlink_alias(self) -> None:
        with tempfile.TemporaryDirectory(
            dir=Path.cwd(), prefix=".lifecycle-symlink-target-"
        ) as target_name, tempfile.TemporaryDirectory(
            dir=tempfile.gettempdir(), prefix="iop-life-alias-parent-"
        ) as alias_parent_name:
            target = Path(target_name)
            alias = Path(alias_parent_name) / "attempt"
            alias.symlink_to(target.resolve(), target_is_directory=True)
            evidence = target / "evidence"
            evidence.mkdir()
            result = self._run(replace(
                self._spec("print('FINISH'); print('IDLE')"),
                evidence_dir=str(evidence),
                control_dir=str(alias / "control"),
            ))
            self.assertTrue(result.product.status == "succeeded")
            self.assertTrue((target / "control" / "locator.json").is_file())
            self.assertFalse((target / "control" / "control.sock").exists())

    def test_owned_descendant_ignoring_term_is_killed_and_reaped(self) -> None:
        descendant_pid_path = self.root / "descendant.pid"
        descendant_source = (
            "import signal,time; from pathlib import Path; "
            "signal.signal(signal.SIGTERM, signal.SIG_IGN); "
            f"Path({str(descendant_pid_path)!r}).write_text(str(__import__('os').getpid())); "
            "time.sleep(30)"
        )
        caller_source = (
            "import subprocess,sys,time; from pathlib import Path; "
            f"p=subprocess.Popen([sys.executable,'-c',{descendant_source!r}]); "
            f"deadline=time.monotonic()+5; marker=Path({str(descendant_pid_path)!r}); "
            "\nwhile not marker.exists() and time.monotonic()<deadline: time.sleep(.01)\n"
            "print('FINISH'); print('IDLE'); sys.stdout.flush(); time.sleep(30)"
        )
        evidence = self.root / "descendant-evidence"
        evidence.mkdir()
        control = self.root / "descendant-control"
        descendant_pid = 0
        try:
            result = self._run(replace(
                self._spec(
                    caller_source,
                    completion_mode=COMPLETION_STOP_AFTER_IDLE,
                ),
                evidence_dir=str(evidence),
                control_dir=str(control),
            ))

            descendant_pid = int(descendant_pid_path.read_text(encoding="utf-8"))
            proc_path = Path(f"/proc/{descendant_pid}")
            deadline = time.monotonic() + 2
            while proc_path.exists() and time.monotonic() < deadline:
                time.sleep(.02)
            self.assertTrue(result.product.status == "succeeded")
            self.assertTrue(result.harness.cleanup_complete)
            self.assertFalse(result.process_group_alive)
            self.assertFalse(proc_path.exists())
            receipt = json.loads(
                (control / "cleanup-receipt.json").read_text(encoding="utf-8")
            )
            self.assertTrue(receipt["cleanup_complete"])
            self.assertFalse(receipt["process_group_alive"])
        finally:
            if not descendant_pid and descendant_pid_path.is_file():
                descendant_pid = int(descendant_pid_path.read_text(encoding="utf-8"))
            if descendant_pid and Path(f"/proc/{descendant_pid}").exists():
                try:
                    os.kill(descendant_pid, signal.SIGKILL)
                except ProcessLookupError:
                    pass

    def test_authenticated_recovery_status_and_stop(self) -> None:
        control = self.root / "recovery-control"
        process, writer, reader = self._start_supervisor(control)
        try:
            self._send_supervisor_spec(writer, "import time; time.sleep(30)")
            self.assertEqual(self._read_frame(reader, 5).get("op"), "registered")
            locator = read_locator(control)
            self._write_frame(writer, {"op": "start"})
            self.assertEqual(self._read_frame(reader, 5).get("op"), "started")

            status = recover_invocation(locator, stop=False)
            self.assertTrue(status.caller_launched)
            self.assertTrue(status.process_group_alive)
            stopped = recover_invocation(locator, stop=True)
            self.assertEqual(stopped.reason, REASON_RECOVERED_STOP)
            self.assertTrue(stopped.cleanup_complete)
            self.assertFalse(stopped.process_group_alive)
            self.assertEqual(process.wait(timeout=8), 0)
        finally:
            self._close_supervisor(process, writer, reader)

        receipt = json.loads((control / "cleanup-receipt.json").read_text(encoding="utf-8"))
        self.assertEqual(receipt["reason"], REASON_RECOVERED_STOP)
        self.assertTrue(receipt["cleanup_complete"])

    def test_locator_identity_mismatches_refuse_recovery(self) -> None:
        control = self.root / "mismatch-control"
        process, writer, reader = self._start_supervisor(control)
        independent = subprocess.Popen(
            (sys.executable, "-c", "import time; time.sleep(30)"),
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        try:
            self._send_supervisor_spec(writer, "import time; time.sleep(30)")
            self.assertEqual(self._read_frame(reader, 5).get("op"), "registered")
            locator = read_locator(control)
            self._write_frame(writer, {"op": "start"})
            self.assertEqual(self._read_frame(reader, 5).get("op"), "started")

            with self.assertRaises(LifecycleRecoveryError):
                recover_invocation(replace(locator, start_identity="mismatched-start"))
            with self.assertRaises(LifecycleRecoveryError):
                recover_invocation(replace(locator, supervisor_pid=independent.pid))
            status = recover_invocation(locator, stop=False)
            self.assertTrue(status.process_group_alive)
            stopped = recover_invocation(locator)
            self.assertTrue(stopped.cleanup_complete)
            self.assertFalse(stopped.process_group_alive)
            self.assertEqual(process.wait(timeout=8), 0)
        finally:
            independent.terminate()
            independent.wait(timeout=5)
            self._close_supervisor(process, writer, reader)

    def test_controller_eof_before_start_launches_no_caller(self) -> None:
        control = self.root / "pre-start-control"
        marker = self.root / "pre-start-caller-ran"
        process, writer, reader = self._start_supervisor(control)
        try:
            self._send_supervisor_spec(
                writer,
                f"from pathlib import Path; Path({str(marker)!r}).write_text('ran')",
            )
            self.assertEqual(self._read_frame(reader, 5).get("op"), "registered")
            writer.close()
            self.assertEqual(process.wait(timeout=8), 0)
        finally:
            self._close_supervisor(process, writer, reader)

        receipt = json.loads((control / "cleanup-receipt.json").read_text(encoding="utf-8"))
        self.assertEqual(receipt["reason"], REASON_CONTROLLER_LOST)
        self.assertFalse(receipt["caller_launched"])
        self.assertTrue(receipt["cleanup_complete"])
        self.assertFalse(marker.exists())

    def test_near_deadline_terminal_reason_and_receipt_are_consistent(self) -> None:
        cases = (
            (
                "timeout",
                "import time; time.sleep(.85); print('FINISH'); print('IDLE')",
                REASON_TIMED_OUT,
                None,
            ),
            (
                "cancel",
                "import time; time.sleep(.15); print('FINISH'); print('IDLE'); time.sleep(30)",
                REASON_CANCELLED,
                .9,
            ),
        )
        for index, (name, source, expected, cancel_after) in enumerate(cases):
            with self.subTest(race=name):
                evidence = self.root / f"race-evidence-{index}"
                evidence.mkdir()
                control = self.root / f"race-control-{index}"
                token = CancellationToken() if cancel_after is not None else None
                timer = (
                    threading.Timer(cancel_after, token.cancel)
                    if cancel_after is not None and token is not None
                    else None
                )
                if timer is not None:
                    timer.start()
                try:
                    result = self._run(
                        replace(
                            self._spec(
                                source,
                                completion_mode=COMPLETION_STOP_AFTER_IDLE,
                                run_seconds=1 if name == "timeout" else 5,
                            ),
                            evidence_dir=str(evidence),
                            control_dir=str(control),
                        ),
                        cancellation=token,
                    )
                finally:
                    if timer is not None:
                        timer.cancel()

                receipt = json.loads(
                    (control / "cleanup-receipt.json").read_text(encoding="utf-8")
                )
                published = json.loads(Path(result.result_path).read_text(encoding="utf-8"))
                journal = [
                    json.loads(line)
                    for line in Path(result.journal_path).read_text(encoding="utf-8").splitlines()
                ]
                terminals = [record for record in journal if record.get("record") == "terminal"]
                self.assertEqual(result.harness.reason, expected)
                self.assertEqual(receipt["reason"], expected)
                self.assertEqual(published["harness"]["reason"], expected)
                self.assertEqual(receipt["exit_code"], published["process"]["exit_code"])
                self.assertEqual(receipt["signal"], published["process"]["signal"])
                self.assertEqual(len(terminals), 1)
                self.assertEqual(terminals[0]["harness"]["reason"], expected)
                self.assertEqual(journal[-1]["record"], "terminal")
                self.assertTrue(result.harness.cleanup_complete)
                self.assertFalse(result.process_group_alive)

    def test_controller_eof_routes_supervisor_through_cleanup(self) -> None:
        """Exercise the crash window directly: EOF after START must clean the group."""
        control_dir = self.root / "control"
        control_dir.mkdir()
        controller_read, supervisor_write = os.pipe()
        supervisor_read, controller_write = os.pipe()
        process = subprocess.Popen(
            (
                sys.executable,
                "-m",
                "scripts.agent_benchmark.lifecycle",
                f"--read-fd={supervisor_read}",
                f"--write-fd={supervisor_write}",
                f"--control-dir={control_dir}",
            ),
            cwd=str(Path(__file__).resolve().parents[2]),
            env={"PATH": os.environ.get("PATH", "/usr/bin:/bin"), "PYTHONPATH": str(Path(__file__).resolve().parents[2])},
            pass_fds=(supervisor_read, supervisor_write),
            start_new_session=True,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
        )
        os.close(supervisor_read)
        os.close(supervisor_write)
        writer = os.fdopen(controller_write, "wb", buffering=0)
        reader = os.fdopen(controller_read, "rb", buffering=0)
        try:
            self._write_frame(writer, {
                "op": "spec",
                "argv": [sys.executable, "-u", "-c", "import time; time.sleep(30)"],
                "cwd": str(self.root),
                "env": [["PATH", os.environ.get("PATH", "/usr/bin:/bin")]],
                "submission_mode": SUBMISSION_ARGV_TASK,
                "task_payload_hex": "",
                "max_capture_bytes": 1024,
                "cleanup_grace_seconds": 1,
                "fault_injection": "",
            })
            self.assertEqual(self._read_frame(reader, 5).get("op"), "registered")
            self._write_frame(writer, {"op": "start"})
            self.assertEqual(self._read_frame(reader, 5).get("op"), "started")
            writer.close()  # Simulated controller loss.
            self.assertEqual(process.wait(timeout=8), 0)
        finally:
            reader.close()
            if process.poll() is None:
                process.kill()
                process.wait(timeout=5)
        receipt = json.loads((control_dir / "cleanup-receipt.json").read_text(encoding="utf-8"))
        self.assertEqual(receipt["reason"], REASON_CONTROLLER_LOST)
        self.assertTrue(receipt["caller_launched"])
        self.assertTrue(receipt["cleanup_complete"])
        self.assertFalse(receipt["process_group_alive"])

    @staticmethod
    def _write_frame(handle: object, value: dict[str, object]) -> None:
        handle.write((json.dumps(value) + "\n").encode("utf-8"))  # type: ignore[attr-defined]
        handle.flush()  # type: ignore[attr-defined]

    @staticmethod
    def _read_frame(handle: object, timeout: float) -> dict[str, object]:
        ready, _, _ = select.select([handle], [], [], timeout)
        if not ready:
            raise AssertionError("timed out waiting for supervisor frame")
        raw = handle.readline()  # type: ignore[attr-defined]
        if not raw:
            raise AssertionError("supervisor closed its frame stream")
        return json.loads(raw)


if __name__ == "__main__":
    unittest.main()