tests/test_supervisor.py

from __future__ import annotations

import json
import socket
import struct
import tempfile
import threading
import time
import unittest
from pathlib import Path
from unittest.mock import MagicMock, patch

from lumi_eggcracker import incidents
from lumi_eggcracker.artifacts import ArtifactEvidence
from lumi_eggcracker.client import OPERATOR_SOCKET as CLIENT_OPERATOR_SOCKET
from lumi_eggcracker.client import QUERY_SOCKET as CLIENT_QUERY_SOCKET
from lumi_eggcracker.client import SOCKETS as CLIENT_SOCKETS
from lumi_eggcracker.containment import EmptyProof
from lumi_eggcracker.detectors import DetectionMatch, load_bundled, match
from lumi_eggcracker.discovery import ProcessIdentity, ProcessSnapshot
from lumi_eggcracker.elfmarkers import (
    PYTORCH_ATEN_EVIDENCE_ID,
    PYTORCH_BRIDGE_EVIDENCE_ID,
    PYTORCH_PAIR_EVIDENCE_ID,
    VLLM_EXTENSION_EVIDENCE_ID,
    VLLM_PAIR_EVIDENCE_ID,
    VLLM_PYTHON_EVIDENCE_ID,
    RuntimeEvidence,
    with_pytorch_pair,
)
from lumi_eggcracker.jsonio import JsonInputError
from lumi_eggcracker.records import RUN_SCHEMA, command_summary, load_run
from lumi_eggcracker.supervisor import (
    MAX_FRAME,
    OPERATOR_SOCKET,
    PROTECTED_SYSTEMD_ADDRESS_FAMILIES,
    PROTECTED_SYSTEMD_ENVIRONMENT,
    PROTECTED_SYSTEMD_INTERFACE_DISCOVERY_ADDRESS_FAMILIES,
    QUERY_SOCKET,
    SOCKET_ACTIONS,
    Supervisor,
    _bounded_query_items,
    _EvidenceCandidate,
    _receive,
    protected_systemd_address_families,
)


def record() -> dict[str, object]:
    run_id = "a" * 24
    return {**command_summary(["/bin/true"]), "boot_id": "b" * 36, "boundary": None, "cgroup": f"/system.slice/lumi-eggcracker-workload-{run_id}.service", "cgroup_device": 1, "cgroup_inode": 2, "cpu_quota_percent": 400, "created_monotonic_ns": 3, "max_memory_mib": 2048, "max_pids": 8, "name": "demo", "network_mode": "none", "operator_uid": 1001, "run_id": run_id, "schema_version": RUN_SCHEMA, "state": "RUNNING", "unit": f"lumi-eggcracker-workload-{run_id}.service", "workload_gid": 2001, "workload_uid": 2001}


def retained_record(run_id: str, state: str, created_monotonic_ns: int) -> dict[str, object]:
    return {
        **record(),
        "cgroup": f"/system.slice/lumi-eggcracker-workload-{run_id}.service",
        "created_monotonic_ns": created_monotonic_ns,
        "name": f"workload-{run_id[:8]}",
        "run_id": run_id,
        "state": state,
        "unit": f"lumi-eggcracker-workload-{run_id}.service",
    }


class SupervisorTests(unittest.TestCase):
    def _instance(self) -> Supervisor:
        value = object.__new__(Supervisor)
        value.policy = {
            "network_mode": "offline",
            "operator_gid": 1001,
            "operator_uid": 1001,
            "source_commit": "c" * 40,
            "workload_gid": 2001,
            "workload_uid": 2001,
        }
        value.installation_epoch = "e" * 64
        value.runs = Path(".")
        value.names = Path(".")
        value.receipts = Path(".")
        value.locks = {}
        value.lock_guard = threading.Lock()
        value.start_lock = threading.Lock()
        value.completed = {}
        value.operations = []
        value.discovery_active = set()
        value.discovery_done = {}
        value.discovery_containing_cgroups = set()
        value.owned_containment_cgroups = set()
        value.discovery_lock = threading.Lock()
        value.content_scan_tick = 0
        value.receipt_persistence_healthy = True
        return value

    def test_protected_systemd_environment_has_fixed_minimal_path(self) -> None:
        path_entries = tuple(
            entry for entry in PROTECTED_SYSTEMD_ENVIRONMENT if entry.startswith("--setenv=PATH=")
        )
        self.assertEqual(path_entries, ("--setenv=PATH=/usr/bin",))
        self.assertNotIn("--setenv=PATH=.", PROTECTED_SYSTEMD_ENVIRONMENT)
        self.assertNotIn("--setenv=PATH=/bin", PROTECTED_SYSTEMD_ENVIRONMENT)
        self.assertFalse(any("venv" in entry for entry in path_entries))
        self.assertEqual(len(PROTECTED_SYSTEMD_ENVIRONMENT), len(set(PROTECTED_SYSTEMD_ENVIRONMENT)))

    def test_protected_workload_allows_only_required_address_families(self) -> None:
        self.assertEqual(
            PROTECTED_SYSTEMD_ADDRESS_FAMILIES,
            "--property=RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6",
        )
        self.assertEqual(
            PROTECTED_SYSTEMD_INTERFACE_DISCOVERY_ADDRESS_FAMILIES,
            "--property=RestrictAddressFamilies=AF_UNIX AF_INET AF_INET6 AF_NETLINK",
        )
        self.assertEqual(
            protected_systemd_address_families(None),
            PROTECTED_SYSTEMD_ADDRESS_FAMILIES,
        )
        self.assertEqual(
            protected_systemd_address_families(
                {"allow_interface_discovery": False}
            ),
            PROTECTED_SYSTEMD_ADDRESS_FAMILIES,
        )
        self.assertEqual(
            protected_systemd_address_families(
                {"allow_interface_discovery": True}
            ),
            PROTECTED_SYSTEMD_INTERFACE_DISCOVERY_ADDRESS_FAMILIES,
        )

    def test_live_workload_identity_requires_distinct_exact_uid_gid(self) -> None:
        supervisor = self._instance()
        supervisor.policy.update(
            {
                "operator_gid": 1000,
                "operator_uid": 1000,
                "workload_gid": 988,
                "workload_uid": 999,
            }
        )
        account = MagicMock(
            pw_dir="/nonexistent",
            pw_gid=988,
            pw_name="lumi-eggcracker-workload",
            pw_shell="/usr/sbin/nologin",
            pw_uid=999,
        )
        group = MagicMock(gr_name="lumi-eggcracker-workload")
        with (
            patch("lumi_eggcracker.supervisor.pwd.getpwnam", return_value=account),
            patch("lumi_eggcracker.supervisor.grp.getgrgid", return_value=group),
            patch("lumi_eggcracker.supervisor.os.getgrouplist", return_value=[988]),
        ):
            self.assertTrue(supervisor._workload_identity_status()["healthy"])
            supervisor.policy["workload_gid"] = 999
            self.assertFalse(supervisor._workload_identity_status()["healthy"])

    def test_incomplete_vllm_delay_survives_witness_and_helper_churn(self) -> None:
        supervisor = self._instance()
        supervisor.runtime_pytorch_delay_spent = set()
        runtime_identity = ProcessIdentity(41, 401)
        first_content_identity = ProcessIdentity(42, 402)
        second_content_identity = ProcessIdentity(43, 403)
        first_helper_identity = ProcessIdentity(44, 404)
        second_helper_identity = ProcessIdentity(45, 405)

        def snapshot(identity: ProcessIdentity) -> ProcessSnapshot:
            return ProcessSnapshot(
                identity,
                2001,
                "/usr/bin/python3",
                "python3",
                ("python3",),
                (),
                (),
                (),
            )

        content = ArtifactEvidence(
            "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
        )
        runtime = RuntimeEvidence(
            PYTORCH_PAIR_EVIDENCE_ID, "PyTorch/ATen", "SHA256_PAIR", ()
        )
        runtime_candidate = _EvidenceCandidate(
            snapshot(runtime_identity), (), (runtime,), 1
        )
        first_content = _EvidenceCandidate(
            snapshot(first_content_identity), (content,), (), 1
        )
        first_helper = _EvidenceCandidate(snapshot(first_helper_identity), (), (), 1)
        detected = DetectionMatch("content.safetensors-pytorch", "CONTENT", ())
        delayed_this_scan: set[ProcessIdentity] = set()

        delay, contributors = supervisor._delay_incomplete_vllm_decision(
            detected,
            (runtime_candidate, first_content),
            (runtime_candidate, first_content, first_helper),
            {first_helper_identity},
            delayed_this_scan,
        )
        self.assertTrue(delay)
        self.assertEqual(
            {runtime_identity, first_content_identity}, contributors
        )

        # Witness order and duplicate groups cannot spend a second delay in
        # the same scan.
        duplicate_delay, _ = supervisor._delay_incomplete_vllm_decision(
            detected,
            (first_content, runtime_candidate),
            (first_helper, first_content, runtime_candidate),
            {first_helper_identity},
            delayed_this_scan,
        )
        self.assertTrue(duplicate_delay)

        second_content = _EvidenceCandidate(
            snapshot(second_content_identity), (content,), (), 2
        )
        second_helper = _EvidenceCandidate(snapshot(second_helper_identity), (), (), 2)
        supervisor._prune_runtime_pytorch_delay_spent(
            {runtime_identity, second_content_identity, second_helper_identity}
        )
        self.assertEqual(
            {runtime_identity}, supervisor.runtime_pytorch_delay_spent
        )

        # The stable runtime contributor propagates spent state to the new
        # content witness, so a rotated helper cannot suppress enforcement.
        retry_delay, retry_contributors = supervisor._delay_incomplete_vllm_decision(
            detected,
            (second_content, runtime_candidate),
            (second_helper, second_content, runtime_candidate),
            {second_helper_identity},
            set(),
        )
        self.assertFalse(retry_delay)
        self.assertEqual(
            {runtime_identity, second_content_identity}, retry_contributors
        )
        self.assertEqual(
            {runtime_identity, second_content_identity},
            supervisor.runtime_pytorch_delay_spent,
        )

        vllm = DetectionMatch("content.safetensors-vllm", "CONTENT", ())
        vllm_delay, _ = supervisor._delay_incomplete_vllm_decision(
            vllm,
            (runtime_candidate, second_content),
            (runtime_candidate, second_content, second_helper),
            set(),
            set(),
        )
        self.assertFalse(vllm_delay)
        self.assertEqual(set(), supervisor.runtime_pytorch_delay_spent)

    def test_incomplete_vllm_does_not_delay_unrelated_complete_profile(self) -> None:
        supervisor = self._instance()
        supervisor.runtime_pytorch_delay_spent = set()
        identity = ProcessIdentity(51, 501)
        candidate = _EvidenceCandidate(
            ProcessSnapshot(
                identity,
                2001,
                "/usr/bin/runtime",
                "runtime",
                ("runtime",),
                (),
                (),
                (),
            ),
            (),
            (),
            1,
        )
        delay, contributors = supervisor._delay_incomplete_vllm_decision(
            DetectionMatch("content.gguf-llama", "CONTENT", ()),
            (candidate,),
            (candidate,),
            {identity},
            set(),
        )
        self.assertFalse(delay)
        self.assertEqual(set(), contributors)
        self.assertEqual(set(), supervisor.runtime_pytorch_delay_spent)

    def test_authenticated_partial_vllm_topology_gets_one_bounded_delay(self) -> None:
        for evidence_id in (
            VLLM_PYTHON_EVIDENCE_ID,
            VLLM_EXTENSION_EVIDENCE_ID,
        ):
            with self.subTest(evidence_id=evidence_id):
                supervisor = self._instance()
                supervisor.runtime_pytorch_delay_spent = set()
                identity = ProcessIdentity(61, 601)
                snapshot = ProcessSnapshot(
                    identity,
                    2001,
                    "/usr/bin/python3",
                    "python3",
                    ("python3",),
                    (),
                    (),
                    (),
                )
                content = ArtifactEvidence(
                    "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
                )
                pytorch = RuntimeEvidence(
                    PYTORCH_PAIR_EVIDENCE_ID, "PyTorch/ATen", "SHA256_PAIR", ()
                )
                partial = RuntimeEvidence(evidence_id, "vLLM", "SHA256", ())
                candidate = _EvidenceCandidate(
                    snapshot, (content,), (pytorch, partial), 1
                )
                pending = (
                    {identity}
                    if supervisor._vllm_topology_pending((partial,), frozenset())
                    else set()
                )
                detected = DetectionMatch(
                    "content.safetensors-pytorch", "CONTENT", ()
                )
                first, _ = supervisor._delay_incomplete_vllm_decision(
                    detected, (candidate,), (candidate,), pending, set()
                )
                second, _ = supervisor._delay_incomplete_vllm_decision(
                    detected, (candidate,), (candidate,), pending, set()
                )
                self.assertTrue(first)
                self.assertFalse(second)

    def test_complete_vllm_topology_overrides_pending_and_clears_delay(self) -> None:
        supervisor = self._instance()
        identity = ProcessIdentity(71, 701)
        supervisor.runtime_pytorch_delay_spent = {identity}
        python = RuntimeEvidence(
            VLLM_PYTHON_EVIDENCE_ID, "vLLM/CPython", "SHA256", ()
        )
        extension = RuntimeEvidence(
            VLLM_EXTENSION_EVIDENCE_ID, "vLLM", "SHA256", ()
        )
        pair = RuntimeEvidence(VLLM_PAIR_EVIDENCE_ID, "vLLM", "SHA256_PAIR", ())
        self.assertFalse(
            supervisor._vllm_topology_pending(
                (python, extension, pair), {VLLM_EXTENSION_EVIDENCE_ID}
            )
        )
        snapshot = ProcessSnapshot(
            identity, 2001, "/usr/bin/python3", "python3", (), (), (), ()
        )
        content = ArtifactEvidence(
            "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
        )
        pytorch = RuntimeEvidence(
            PYTORCH_PAIR_EVIDENCE_ID, "PyTorch/ATen", "SHA256_PAIR", ()
        )
        candidate = _EvidenceCandidate(
            snapshot, (content,), (pytorch, python, extension, pair), 1
        )
        delay, _ = supervisor._delay_incomplete_vllm_decision(
            DetectionMatch("content.safetensors-vllm", "CONTENT", ()),
            (candidate,),
            (candidate,),
            {identity},
            set(),
        )
        self.assertFalse(delay)
        self.assertEqual(set(), supervisor.runtime_pytorch_delay_spent)

    def test_lookalike_runtime_ids_do_not_create_vllm_pending_state(self) -> None:
        supervisor = self._instance()
        lookalikes = (
            RuntimeEvidence("vllm-python-lookalike", "vLLM/CPython", "SHA256", ()),
            RuntimeEvidence("vllm-extension-lookalike", "vLLM", "SHA256", ()),
        )
        self.assertFalse(supervisor._vllm_topology_pending(lookalikes, frozenset()))

    def test_gated_process_credentials_reject_foreign_supplementary_group(self) -> None:
        supervisor = self._instance()
        supervisor.policy.update({"workload_gid": 988, "workload_uid": 999})
        process = ProcessIdentity(42, 100)
        exact = "Uid:\t999\t999\t999\t999\nGid:\t988\t988\t988\t988\nGroups:\t988\n"
        with patch("lumi_eggcracker.supervisor.Path.read_text", return_value=exact):
            supervisor._validate_gated_credentials(process)
        contaminated = exact.replace("Groups:\t988", "Groups:\t988 999")
        with patch(
            "lumi_eggcracker.supervisor.Path.read_text",
            return_value=contaminated,
        ), self.assertRaisesRegex(JsonInputError, "do not match policy"):
            supervisor._validate_gated_credentials(process)

    def test_required_approval_rejects_before_launch_side_effects(self) -> None:
        supervisor = self._instance()
        supervisor.approval_lock = threading.Lock()
        supervisor.active_cgroups = set()
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            supervisor.approvals = root / "approvals"
            supervisor.approvals.mkdir()
            supervisor.names = root / "names"
            supervisor.names.mkdir()
            with (
                patch.object(supervisor, "_require_workload_identity"),
                patch.object(
                    supervisor,
                    "_incident_status",
                    return_value={"healthy": True},
                ),
                patch.object(
                    supervisor,
                    "_incident_block_for_start",
                    return_value=None,
                ),
                patch(
                    "lumi_eggcracker.supervisor.match_launch",
                    side_effect=JsonInputError("approval not found"),
                ),
                patch("lumi_eggcracker.supervisor.OfflineBoundary.create") as boundary,
                self.assertRaisesRegex(
                    JsonInputError,
                    "^protected launch is not approved$",
                ),
            ):
                supervisor._start(
                    {
                        "argv": ["/bin/true"],
                        "cpu_quota_percent": 400,
                        "max_memory_mib": 2048,
                        "max_pids": 8,
                        "name": "demo",
                        "require_approval": True,
                    }
                )
            boundary.assert_not_called()

    def test_check_start_matches_exact_tuple_and_is_side_effect_free(self) -> None:
        supervisor = self._instance()
        supervisor.approval_lock = threading.Lock()
        supervisor.active_cgroups = set()
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            supervisor.approvals = root / "approvals"
            supervisor.approvals.mkdir()
            supervisor.names = root / "names"
            supervisor.names.mkdir()
            supervisor.incidents = root / "incidents"
            supervisor.exec_policies = root / "exec-policies"
            args = {
                "argv": ["/bin/true", "--exact"],
                "cpu_quota_percent": 700,
                "max_memory_mib": 4096,
                "max_pids": 12,
                "name": "demo",
            }
            approval = {"allow_interface_discovery": False}
            with (
                patch.object(supervisor, "_require_workload_identity"),
                patch(
                    "lumi_eggcracker.supervisor.match_launch",
                    return_value=approval,
                ) as match_launch,
                patch(
                    "lumi_eggcracker.supervisor.inspect_executable",
                    return_value={"sha256": "a" * 64},
                ) as inspect,
                patch("lumi_eggcracker.supervisor.OfflineBoundary.create") as boundary,
                patch("lumi_eggcracker.supervisor.os.urandom") as random,
                patch("lumi_eggcracker.supervisor.write_atomic") as write,
                patch.object(supervisor, "_make_gate") as make_gate,
                patch.object(supervisor, "_make_exec_channel") as make_exec,
                patch.object(supervisor, "_run") as run,
            ):
                value = supervisor.handle({"action": "check_start", "args": args})
            self.assertEqual(
                {
                    "next_action": (
                        "run start --require-approval immediately; admission is revalidated"
                    ),
                    "reason_code": "EXACT_APPROVAL_MATCHED",
                    "result": "READY_TO_ATTEMPT",
                    "schema": "lumi-eggcracker.check-start.v1",
                },
                value,
            )
            match_launch.assert_called_once_with(
                uid=2001,
                argv=["/bin/true", "--exact"],
                approvals=[],
                max_pids=12,
                max_memory_mib=4096,
                cpu_quota_percent=700,
            )
            inspect.assert_called_once_with("/bin/true")
            boundary.assert_not_called()
            random.assert_not_called()
            write.assert_not_called()
            make_gate.assert_not_called()
            make_exec.assert_not_called()
            run.assert_not_called()

    def test_check_start_rejects_missing_approval_without_mutation(self) -> None:
        supervisor = self._instance()
        supervisor.approval_lock = threading.Lock()
        supervisor.active_cgroups = set()
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            supervisor.approvals = root / "approvals"
            supervisor.approvals.mkdir()
            supervisor.names = root / "names"
            supervisor.names.mkdir()
            supervisor.incidents = root / "incidents"
            supervisor.exec_policies = root / "exec-policies"
            args = {
                "argv": ["/bin/true"],
                "cpu_quota_percent": 400,
                "max_memory_mib": 2048,
                "max_pids": 8,
                "name": "demo",
            }
            with (
                patch.object(supervisor, "_require_workload_identity"),
                patch("lumi_eggcracker.supervisor.match_launch", return_value=None),
                patch("lumi_eggcracker.supervisor.OfflineBoundary.create") as boundary,
                patch("lumi_eggcracker.supervisor.os.urandom") as random,
                patch("lumi_eggcracker.supervisor.write_atomic") as write,
            ):
                value = supervisor.handle({"action": "check_start", "args": args})
            self.assertEqual("NOT_READY", value["result"])
            self.assertEqual("APPROVAL_MISSING", value["reason_code"])
            self.assertNotIn("/bin/true", json.dumps(value))
            boundary.assert_not_called()
            random.assert_not_called()
            write.assert_not_called()

    def test_check_start_reports_name_lockdown_and_unavailable_state(self) -> None:
        supervisor = self._instance()
        supervisor.approval_lock = threading.Lock()
        supervisor.active_cgroups = set()
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            supervisor.approvals = root / "approvals"
            supervisor.approvals.mkdir()
            supervisor.names = root / "names"
            supervisor.names.mkdir()
            supervisor.incidents = root / "incidents"
            supervisor.exec_policies = root / "exec-policies"
            args = {
                "argv": ["/bin/true"],
                "cpu_quota_percent": 400,
                "max_memory_mib": 2048,
                "max_pids": 8,
                "name": "demo",
            }
            with patch.object(supervisor, "_require_workload_identity"):
                supervisor.active_cgroups.add("owned")
                self.assertEqual(
                    "NAME_UNAVAILABLE",
                    supervisor.handle({"action": "check_start", "args": args})[
                        "reason_code"
                    ],
                )
                supervisor.active_cgroups.clear()
                with patch.object(
                    supervisor,
                    "_incident_block_for_start_read_only",
                    return_value={"incident_id": "private-incident"},
                ):
                    value = supervisor.handle({"action": "check_start", "args": args})
                self.assertEqual("LOCKDOWN", value["reason_code"])
                self.assertNotIn("private-incident", json.dumps(value))
                with patch.object(
                    supervisor,
                    "_read_only_incident_status",
                    return_value={"healthy": False},
                ):
                    self.assertEqual(
                        "STATE_UNAVAILABLE",
                        supervisor.handle({"action": "check_start", "args": args})[
                            "reason_code"
                        ],
                    )

    def test_check_start_rejects_invalid_arguments_and_execution_policy(self) -> None:
        supervisor = self._instance()
        supervisor.approval_lock = threading.Lock()
        supervisor.active_cgroups = set()
        with tempfile.TemporaryDirectory() as temporary:
            root = Path(temporary)
            supervisor.approvals = root / "approvals"
            supervisor.approvals.mkdir()
            supervisor.names = root / "names"
            supervisor.names.mkdir()
            supervisor.incidents = root / "incidents"
            supervisor.exec_policies = root / "exec-policies"
            invalid = {
                "argv": ["/bin/true"],
                "cpu_quota_percent": 400,
                "max_memory_mib": 2048,
                "max_pids": 3,
                "name": "demo",
            }
            with patch.object(supervisor, "_require_workload_identity"):
                self.assertEqual(
                    "INVALID_ARGUMENTS",
                    supervisor.handle({"action": "check_start", "args": invalid})[
                        "reason_code"
                    ],
                )
                valid = {**invalid, "max_pids": 8}
                with (
                    patch("lumi_eggcracker.supervisor.match_launch", return_value={}),
                    patch(
                        "lumi_eggcracker.supervisor.inspect_executable",
                        side_effect=JsonInputError("private policy path"),
                    ),
                ):
                    self.assertEqual(
                        "EXECUTION_POLICY_INVALID",
                        supervisor.handle({"action": "check_start", "args": valid})[
                            "reason_code"
                        ],
                    )

    def test_check_start_is_only_authorized_on_operator_socket(self) -> None:
        self.assertIn("check_start", SOCKET_ACTIONS[OPERATOR_SOCKET])
        self.assertNotIn("check_start", SOCKET_ACTIONS[QUERY_SOCKET])
        self.assertNotIn(
            "check_start",
            SOCKET_ACTIONS[Path("/run/lumi-eggcracker/admin.sock")],
        )
        self.assertEqual(CLIENT_OPERATOR_SOCKET, CLIENT_SOCKETS["check_start"])

    def test_history_is_allowlisted_on_the_query_socket_only(self) -> None:
        self.assertIn("history", SOCKET_ACTIONS[QUERY_SOCKET])
        self.assertNotIn("history", SOCKET_ACTIONS[OPERATOR_SOCKET])
        self.assertNotIn("history", SOCKET_ACTIONS[Path("/run/lumi-eggcracker/admin.sock")])
        self.assertEqual(CLIENT_QUERY_SOCKET, CLIENT_SOCKETS["history"])

    def test_history_query_uses_existing_query_authorization(self) -> None:
        supervisor = self._instance()
        supervisor.policy["operator_uid"] = 1001
        connection = MagicMock()
        connection.getsockopt.return_value = struct.pack("3i", 42, 1001, 1001)
        with (
            patch(
                "lumi_eggcracker.supervisor._receive",
                return_value={"action": "history", "args": {"limit": 1}},
            ),
            patch.object(supervisor, "handle", return_value={"runs": []}) as handle,
            patch("lumi_eggcracker.supervisor._send") as send,
            patch("lumi_eggcracker.supervisor.socket.SO_PEERCRED", 17, create=True),
        ):
            supervisor._serve_connection(connection, QUERY_SOCKET)
        handle.assert_called_once_with({"action": "history", "args": {"limit": 1}})
        send.assert_called_once_with(connection, {"ok": True, "value": {"runs": []}})

        connection.getsockopt.return_value = struct.pack("3i", 42, 2002, 1001)
        with (
            patch("lumi_eggcracker.supervisor._send") as send,
            patch("lumi_eggcracker.supervisor.socket.SO_PEERCRED", 17, create=True),
        ):
            supervisor._serve_connection(connection, QUERY_SOCKET)
        send.assert_called_once_with(
            connection,
            {"ok": False, "value": "peer uid is not authorized"},
        )

    def test_oversized_valid_json_integer_is_rejected_without_escaping(self) -> None:
        server, client = socket.socketpair()
        try:
            payload = (
                b'{"action":"status","args":{"name":'
                + b"9" * 5000
                + b"}}"
            )
            self.assertLess(len(payload), 32 * 1024)
            client.sendall(struct.pack("!I", len(payload)) + payload)
            with self.assertRaisesRegex(JsonInputError, "integer is too large"):
                _receive(server)
        finally:
            client.close()
            server.close()

    def test_connection_boundary_contains_validation_value_error(self) -> None:
        supervisor = self._instance()
        supervisor.policy["operator_uid"] = 1000
        connection = MagicMock()
        connection.getsockopt.return_value = struct.pack("3i", 42, 0, 0)
        with patch(
            "lumi_eggcracker.supervisor._receive",
            side_effect=ValueError("bounded validation failure"),
        ), patch("lumi_eggcracker.supervisor._send") as send:
            supervisor._serve_connection(connection, QUERY_SOCKET)

        send.assert_called_once_with(
            connection, {"ok": False, "value": "bounded validation failure"}
        )

    def test_query_history_is_truncated_before_wire_frame_limit(self) -> None:
        values = [
            {
                "detector": {
                    "profile": "content.safetensors-pytorch",
                    "evidence": ["e" * 128, "f" * 128, "g" * 128],
                },
                "event_id": f"{index:024x}",
                "result": "TERMINATED",
                "trigger": {
                    "kind": "UNAPPROVED_AI_MATCH",
                    "observed_monotonic_ns": index,
                },
                "version": "1.0.0",
            }
            for index in range(130)
        ]

        response = _bounded_query_items("detections", values, maximum=100)
        payload = json.dumps(
            {"ok": True, "value": response},
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")

        self.assertLessEqual(len(payload), MAX_FRAME)
        self.assertGreater(len(response["detections"]), 0)
        self.assertLess(len(response["detections"]), 100)
        self.assertEqual(
            values[: len(response["detections"])],
            response["detections"],
        )

    def test_recently_contained_identity_is_not_reenforced(self) -> None:
        supervisor = self._instance()
        identity = object()
        with supervisor.discovery_lock:
            supervisor.discovery_done[identity] = 1
            self.assertIn(identity, supervisor.discovery_done)

    def test_managed_processes_use_reconciled_in_memory_cgroups(self) -> None:
        supervisor = self._instance()
        cgroup = "/system.slice/lumi-eggcracker-workload-" + "a" * 24 + ".service"
        supervisor.active_cgroups = {cgroup}
        snapshot = ProcessSnapshot(
            ProcessIdentity(10, 100),
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            ("0::" + cgroup,),
            (),
            (),
        )
        self.assertTrue(supervisor._managed(snapshot))
        self.assertFalse(supervisor._discovery_excluded(snapshot))

    def test_topology_closure_requires_live_exact_anchor_in_detection_scope(self) -> None:
        supervisor = self._instance()
        cgroup = "/system.slice/lumi-eggcracker-workload-" + "a" * 24 + ".service"
        supervisor.active_cgroups = {cgroup}
        anchor = ProcessSnapshot(
            ProcessIdentity(10, 100),
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            ("0::" + cgroup,),
            (),
            (),
        )
        child = ProcessSnapshot(
            ProcessIdentity(11, 101),
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            ("0::" + cgroup,),
            (),
            (),
        )
        provenance = {"cgroup": cgroup, "pid": 10, "start_time": 100}
        anchor_candidate = _EvidenceCandidate(anchor, (), (), 1)
        child_candidate = _EvidenceCandidate(child, (), (), 1)

        with (
            patch.object(
                supervisor, "_cached_executable_digest", return_value="a" * 64
            ),
            patch("lumi_eggcracker.supervisor.launch_authorizes", return_value=True),
        ):
            self.assertFalse(
                supervisor._authorizes_protected_scope(
                    child,
                    provenance,
                    profile="content.gguf-llama",
                    scope=(anchor_candidate, child_candidate),
                    independently_complete=False,
                )
            )
            self.assertTrue(
                supervisor._authorizes_protected_scope(
                    child,
                    provenance,
                    profile="content.gguf-ollama",
                    scope=(anchor_candidate, child_candidate),
                    independently_complete=False,
                )
            )
            self.assertFalse(
                supervisor._authorizes_protected_scope(
                    child,
                    provenance,
                    profile="content.gguf-ollama",
                    scope=(child_candidate,),
                    independently_complete=False,
                )
            )
            self.assertFalse(
                supervisor._authorizes_protected_scope(
                    child,
                    provenance,
                    profile="content.gguf-ollama",
                    scope=(anchor_candidate, child_candidate),
                    independently_complete=True,
                )
            )

    def test_incident_sweep_runs_after_response_lock_release(self) -> None:
        supervisor = self._instance()
        supervisor.incidents = Path(".")
        supervisor.incident_response_lock = threading.Lock()
        supervisor.catalogue = load_bundled()
        receipt = {
            "event_id": "e" * 24,
            "result": "TERMINATED",
            "trigger": {"kind": "UNAPPROVED_AI_MATCH"},
        }
        incident = {"incident_id": "i" * 24, "response": {}}

        def sweep(_incident_id: str) -> None:
            acquired = supervisor.incident_response_lock.acquire(timeout=0.1)
            self.assertTrue(acquired)
            if acquired:
                supervisor.incident_response_lock.release()

        with (
            patch.object(
                supervisor,
                "_incident_match",
                return_value={
                    "argv_sha256": "a" * 64,
                    "uid": 2001,
                    "executable_sha256": "b" * 64,
                    "profile": "content.gguf-llama",
                },
            ),
            patch.object(supervisor, "_incident_values", return_value=[]),
            patch.object(supervisor, "_incident_workload", return_value={}),
            patch.object(supervisor, "_revoke_exact_approval", return_value=True),
            patch.object(supervisor, "_incident_sweep", side_effect=sweep),
            patch("lumi_eggcracker.supervisor.incident_store.find_match", return_value=None),
            patch(
                "lumi_eggcracker.supervisor.incident_store.create",
                return_value=incident,
            ),
            patch(
                "lumi_eggcracker.supervisor.incident_store.update",
                return_value=incident,
            ),
        ):
            supervisor._post_containment_response(receipt)

    def test_incident_sweep_reentry_returns_without_lock_deadlock(self) -> None:
        supervisor = self._instance()
        supervisor.incident_sweep_lock = threading.Lock()
        supervisor.incident_sweep_active = True
        started = time.monotonic()
        supervisor._incident_sweep("i" * 24)
        self.assertLess(time.monotonic() - started, 0.1)

    def test_incident_cache_avoids_revalidating_unchanged_records(self) -> None:
        supervisor = self._instance()
        supervisor.incident_health_healthy = True
        supervisor.incident_response_lock = threading.RLock()
        with tempfile.TemporaryDirectory() as temporary:
            supervisor.incidents = Path(temporary)
            with patch(
                "lumi_eggcracker.supervisor.incident_store.compact",
                wraps=incidents.compact,
            ) as compact:
                self.assertEqual([], supervisor._incident_values())
                self.assertEqual([], supervisor._incident_values())
            self.assertEqual(1, compact.call_count)

    def test_incident_cache_adopts_atomic_write_without_full_reload(self) -> None:
        supervisor = self._instance()
        supervisor.incident_health_healthy = True
        supervisor.incident_response_lock = threading.RLock()
        with tempfile.TemporaryDirectory() as temporary:
            supervisor.incidents = Path(temporary)
            original = incidents.create(
                supervisor.incidents,
                receipt={
                    "event_id": "a" * 24,
                    "observed": {"argv_sha256": "b" * 64, "uid": 2001},
                    "executable": {"sha256": "c" * 64},
                },
                policy={"version": "1.0.0"},
                catalogue_sha256="d" * 64,
                source_commit="e" * 40,
                version="1.0.0",
                trigger="UNAPPROVED_AI_MATCH",
                profile="content.gguf-llama",
                evidence=["gguf-v3"],
                match={
                    "argv_sha256": "b" * 64,
                    "executable_sha256": "c" * 64,
                    "profile": "content.gguf-llama",
                    "uid": 2001,
                },
                workload={
                    "boot_id": "f" * 36,
                    "cgroup": "/system.slice/lumi-eggcracker-workload-"
                    + "a" * 24
                    + ".service",
                    "cgroup_device": 1,
                    "cgroup_inode": 2,
                    "run_id": "a" * 24,
                    "unit": "lumi-eggcracker-workload-" + "a" * 24 + ".service",
                    "uid": 2001,
                },
                approval=None,
            )
            supervisor._incident_values()
            updated = incidents.update(
                supervisor.incidents,
                original,
                acknowledgement={"monotonic_ns": 1, "uid": 0},
                state="ACKNOWLEDGED",
            )
            supervisor._cache_incident_write(updated)
            with patch(
                "lumi_eggcracker.supervisor.incident_store.compact",
                side_effect=AssertionError("unchanged store was reparsed"),
            ):
                self.assertEqual(
                    "ACKNOWLEDGED", supervisor._incident_values()[0]["state"]
                )

    def test_incident_cache_revalidates_and_rejects_in_place_tamper(self) -> None:
        supervisor = self._instance()
        supervisor.incident_health_healthy = True
        supervisor.incident_response_lock = threading.RLock()
        with tempfile.TemporaryDirectory() as temporary:
            supervisor.incidents = Path(temporary)
            value = incidents.create(
                supervisor.incidents,
                receipt={
                    "event_id": "a" * 24,
                    "observed": {"argv_sha256": "b" * 64, "uid": 2001},
                    "executable": {"sha256": "c" * 64},
                },
                policy={"version": "1.0.0"},
                catalogue_sha256="d" * 64,
                source_commit="e" * 40,
                version="1.0.0",
                trigger="UNAPPROVED_AI_MATCH",
                profile="content.gguf-llama",
                evidence=["gguf-v3"],
                match={
                    "argv_sha256": "b" * 64,
                    "executable_sha256": "c" * 64,
                    "profile": "content.gguf-llama",
                    "uid": 2001,
                },
                workload={
                    "boot_id": "f" * 36,
                    "cgroup": "/system.slice/lumi-eggcracker-workload-"
                    + "a" * 24
                    + ".service",
                    "cgroup_device": 1,
                    "cgroup_inode": 2,
                    "run_id": "a" * 24,
                    "unit": "lumi-eggcracker-workload-" + "a" * 24 + ".service",
                    "uid": 2001,
                },
                approval=None,
            )
            supervisor._incident_values()
            path = supervisor.incidents / f"{value['incident_id']}.json"
            tampered = json.loads(path.read_text(encoding="utf-8"))
            tampered["state"] = "CLEARED"
            path.write_text(json.dumps(tampered) + "\n", encoding="utf-8")
            with self.assertRaises(JsonInputError):
                supervisor._incident_values()
            self.assertFalse(supervisor.incident_health_healthy)

    def test_group_refresh_replaces_preexec_gate_snapshot_without_losing_evidence(self) -> None:
        identity = ProcessIdentity(10, 100)
        stale = ProcessSnapshot(
            identity,
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3", "_gate"),
            ("0::/system.slice/lumi-eggcracker-workload-" + "a" * 24 + ".service",),
            (),
            (),
        )
        current = ProcessSnapshot(
            identity,
            2001,
            "/opt/llama-cli",
            "llama-cli",
            ("llama-cli", "-m", "/models/qwen"),
            stale.cgroups,
            (),
            (),
        )
        evidence = ArtifactEvidence("gguf-v3", "GGUF", 1, 2, 4096, "a" * 64)
        candidate = _EvidenceCandidate(stale, (evidence,), (), 123)

        with patch("lumi_eggcracker.supervisor.process_snapshot", return_value=current):
            refreshed = Supervisor._refresh_group((candidate,))

        self.assertEqual(1, len(refreshed))
        self.assertEqual(current, refreshed[0].snapshot)
        self.assertEqual((evidence,), refreshed[0].content)
        with patch("lumi_eggcracker.supervisor.process_snapshot", return_value=None):
            self.assertEqual((), Supervisor._refresh_group((candidate,)))

    def test_selected_match_contains_every_process_in_exact_run_cgroup(self) -> None:
        supervisor = self._instance()
        run_id = "a" * 24
        selected = f"/system.slice/lumi-eggcracker-workload-{run_id}.service"
        evidence_id = ProcessIdentity(10, 100)
        broker_id = ProcessIdentity(11, 101)
        replacement_id = ProcessIdentity(12, 102)
        canary_id = ProcessIdentity(13, 103)
        evidence = ProcessSnapshot(
            evidence_id,
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            ("0::" + selected,),
            (),
            (),
        )
        broker = ProcessSnapshot(
            broker_id,
            2001,
            "/usr/bin/broker",
            "broker",
            ("broker",),
            ("0::" + selected,),
            (),
            (),
        )
        replacement = ProcessSnapshot(
            replacement_id,
            2001,
            "/usr/bin/replacement",
            "replacement",
            ("replacement",),
            ("0::" + selected,),
            (),
            (),
        )
        canary = ProcessSnapshot(
            canary_id,
            2001,
            "/usr/bin/canary",
            "canary",
            ("canary",),
            ("0::/system.slice/unrelated-canary.service",),
            (),
            (),
        )
        group = (_EvidenceCandidate(evidence, (), (), 1),)

        targets = supervisor._discovery_containment_targets(
            group,
            {
                evidence_id: evidence,
                broker_id: broker,
                replacement_id: replacement,
                canary_id: canary,
            },
        )

        self.assertEqual({evidence_id, broker_id, replacement_id}, targets)

    def test_complete_identity_is_not_suppressed_or_broadened_by_64_partials(self) -> None:
        supervisor = self._instance()
        supervisor.catalogue = load_bundled()
        parent = ProcessIdentity(10, 100)
        parent_snapshot = ProcessSnapshot(
            parent, 2001, "/usr/bin/worker", "worker", ("worker",), (), (), ()
        )
        content = ArtifactEvidence(
            "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
        )
        runtime = RuntimeEvidence(
            "pytorch-bridge-aten-pair-pinned-cpu",
            "PyTorch/ATen",
            "BUILD_ID_PAIR",
            (),
        )
        snapshots = {parent: parent_snapshot}
        candidates: list[_EvidenceCandidate] = []
        for offset in range(64):
            identity = ProcessIdentity(100 + offset, 1000 + offset)
            current = ProcessSnapshot(
                identity,
                2001,
                "/usr/bin/python3",
                "python3",
                ("python3",),
                (),
                (),
                (),
                parent=parent,
            )
            snapshots[identity] = current
            candidates.append(_EvidenceCandidate(current, (content,), (), 1))
        complete_identity = ProcessIdentity(999, 1999)
        complete_snapshot = ProcessSnapshot(
            complete_identity,
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            (),
            (),
            (),
            parent=parent,
        )
        snapshots[complete_identity] = complete_snapshot
        candidates.append(
            _EvidenceCandidate(complete_snapshot, (content,), (runtime,), 2)
        )

        correlated = supervisor._correlate(candidates, snapshots)
        self.assertEqual(1, len(correlated))
        self.assertEqual(65, len(correlated[0][0]))

        groups = supervisor._content_groups(candidates, snapshots)
        self.assertEqual(1, len(groups))
        self.assertEqual(
            (complete_identity,),
            tuple(item.snapshot.identity for item in groups[0].witness),
        )
        self.assertEqual(65, len(groups[0].scope))

    def test_complete_sibling_storm_is_one_full_component_enforcement_scope(self) -> None:
        supervisor = self._instance()
        supervisor.catalogue = load_bundled()
        parent = ProcessIdentity(20, 200)
        snapshots = {
            parent: ProcessSnapshot(
                parent, 2001, "/usr/bin/worker", "worker", ("worker",), (), (), ()
            )
        }
        content = ArtifactEvidence(
            "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
        )
        runtime = RuntimeEvidence(
            "pytorch-bridge-aten-pair-pinned-cpu",
            "PyTorch/ATen",
            "BUILD_ID_PAIR",
            (),
        )
        candidates: list[_EvidenceCandidate] = []
        for offset in range(18):
            current_identity = ProcessIdentity(21 + offset, 201 + offset)
            current = ProcessSnapshot(
                current_identity,
                2001,
                "/usr/bin/python3",
                "python3",
                ("python3",),
                (),
                (),
                (),
                parent=parent,
            )
            snapshots[current_identity] = current
            candidates.append(
                _EvidenceCandidate(current, (content,), (runtime,), 1)
            )

        groups = supervisor._content_groups(candidates, snapshots)

        self.assertEqual(1, len(groups))
        self.assertEqual(18, len(groups[0].witness))
        self.assertEqual(18, len(groups[0].scope))
        self.assertEqual(
            {parent, *(candidate.snapshot.identity for candidate in candidates)},
            supervisor._discovery_containment_targets(groups[0].scope, snapshots),
        )

    def test_partial_peer_is_scope_but_not_witness_for_complete_identity(self) -> None:
        supervisor = self._instance()
        supervisor.catalogue = load_bundled()
        parent = ProcessIdentity(40, 400)
        snapshots = {
            parent: ProcessSnapshot(
                parent, 2001, "/usr/bin/worker", "worker", ("worker",), (), (), ()
            )
        }
        content = ArtifactEvidence(
            "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
        )
        runtime = RuntimeEvidence(
            "pytorch-bridge-aten-pair-pinned-cpu",
            "PyTorch/ATen",
            "BUILD_ID_PAIR",
            (),
        )
        complete_identity = ProcessIdentity(41, 401)
        partial_identity = ProcessIdentity(42, 402)
        complete = ProcessSnapshot(
            complete_identity,
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            (),
            (),
            (),
            parent=parent,
        )
        partial = ProcessSnapshot(
            partial_identity,
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            (),
            (),
            (),
            parent=parent,
        )
        snapshots.update({complete_identity: complete, partial_identity: partial})

        groups = supervisor._content_groups(
            [
                _EvidenceCandidate(complete, (content,), (runtime,), 1),
                _EvidenceCandidate(partial, (content,), (), 1),
            ],
            snapshots,
        )

        self.assertEqual(1, len(groups))
        self.assertEqual((complete_identity,), tuple(item.snapshot.identity for item in groups[0].witness))
        self.assertEqual(
            {complete_identity, partial_identity},
            {item.snapshot.identity for item in groups[0].scope},
        )

    def test_redundant_complete_sibling_pairs_expand_enforcement_scope(self) -> None:
        supervisor = self._instance()
        supervisor.catalogue = load_bundled()
        parent = ProcessIdentity(20, 200)
        snapshots = {
            parent: ProcessSnapshot(
                parent,
                2001,
                "/usr/bin/worker",
                "worker",
                ("worker",),
                (),
                (),
                (),
            )
        }
        content = ArtifactEvidence(
            "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
        )
        runtime = RuntimeEvidence(
            "pytorch-bridge-aten-pair-pinned-cpu",
            "PyTorch/ATen",
            "BUILD_ID_PAIR",
            (),
        )
        candidates: list[_EvidenceCandidate] = []
        for offset, evidence in enumerate(
            ((content, None), (content, None), (None, runtime), (None, runtime))
        ):
            identity = ProcessIdentity(21 + offset, 201 + offset)
            current = ProcessSnapshot(
                identity,
                2001,
                "/usr/bin/python3",
                "python3",
                ("python3",),
                (),
                (),
                (),
                parent=parent,
            )
            snapshots[identity] = current
            candidates.append(
                _EvidenceCandidate(
                    current,
                    (evidence[0],) if evidence[0] is not None else (),
                    (evidence[1],) if evidence[1] is not None else (),
                    1,
                )
            )

        groups = supervisor._content_groups(candidates, snapshots)
        self.assertGreaterEqual(len(groups), 1)
        for group in groups:
            self.assertLessEqual(len(group.witness), 3)
            self.assertEqual(4, len(group.scope))
            self.assertEqual(
                {parent, *(candidate.snapshot.identity for candidate in candidates)},
                supervisor._discovery_containment_targets(group.scope, snapshots),
            )

    def test_heartbeat_stops_when_no_scan_completed_within_health_bound(self) -> None:
        supervisor = self._instance()
        supervisor.discovery_thread = type(
            "LiveThread", (), {"is_alive": lambda _self: True}
        )()
        supervisor.last_heartbeat_sent = 0.0
        supervisor.last_scan_completed_ns = time.monotonic_ns() - 2_000_000_000
        supervisor.discovery_failures = 0
        with patch("lumi_eggcracker.supervisor.socket.socket") as socket_factory:
            supervisor._heartbeat()
        socket_factory.assert_not_called()

    def test_detection_receipt_fault_latches_unhealthy_and_stops_heartbeat(self) -> None:
        supervisor = self._instance()
        supervisor.detections = Path(".")
        with patch(
            "lumi_eggcracker.supervisor.write_atomic", side_effect=OSError("read only")
        ), self.assertRaises(OSError):
            supervisor._store_detection({"event_id": "a" * 24})
        self.assertFalse(supervisor.receipt_persistence_healthy)

        supervisor.discovery_thread = type(
            "LiveThread", (), {"is_alive": lambda _self: True}
        )()
        supervisor.last_heartbeat_sent = 0.0
        supervisor.last_scan_completed_ns = time.monotonic_ns()
        supervisor.discovery_failures = 0
        supervisor.enforcement_saturation_until_ns = 0
        with patch("lumi_eggcracker.supervisor.socket.socket") as socket_factory:
            supervisor._heartbeat()
        socket_factory.assert_not_called()

        supervisor.catalogue = load_bundled()
        supervisor.quarantine_root = Path("/owned")
        supervisor.last_scan_duration_ns = 1
        with patch(
            "lumi_eggcracker.supervisor.Path.is_file", return_value=True
        ), patch(
            "lumi_eggcracker.supervisor.Path.read_text", return_value="pids"
        ), patch("lumi_eggcracker.supervisor.pidfd_available", return_value=True):
            doctor = supervisor.handle({"action": "doctor", "args": {}})
        self.assertEqual("UNSUPPORTED", doctor["result"])
        self.assertFalse(doctor["discovery"]["healthy"])
        self.assertFalse(doctor["discovery"]["receipt_persistence_healthy"])

    def test_synchronous_startup_scan_is_immediately_heartbeat_eligible(self) -> None:
        supervisor = self._instance()
        supervisor.discovery_thread = type(
            "LiveThread", (), {"is_alive": lambda _self: True}
        )()
        supervisor.last_heartbeat_sent = 0.0
        supervisor.last_scan_completed_ns = 0
        supervisor.last_scan_duration_ns = 0
        supervisor.discovery_failures = 2
        supervisor.enforcement_saturation_until_ns = 0
        supervisor.heartbeat_sequence = 0
        with patch.object(supervisor, "_scan_once") as scan:
            supervisor._scan_synchronously()
        scan.assert_called_once_with(synchronous=True)
        self.assertGreater(supervisor.last_scan_completed_ns, 0)
        self.assertGreaterEqual(supervisor.last_scan_duration_ns, 0)
        self.assertEqual(0, supervisor.discovery_failures)
        with patch("lumi_eggcracker.supervisor.socket.AF_UNIX", 1, create=True), patch(
            "lumi_eggcracker.supervisor.socket.SOCK_DGRAM", 2, create=True
        ), patch("lumi_eggcracker.supervisor.socket.socket") as socket_factory:
            supervisor._heartbeat()
        socket_factory.assert_called_once()
        self.assertEqual(1, supervisor.heartbeat_sequence)

    def test_discovery_window_generation_survives_supervisor_restart(self) -> None:
        with tempfile.TemporaryDirectory() as raw:
            path = Path(raw) / "discovery-progress.json"
            first = self._instance()
            first.discovery_progress_path = path
            first._reserve_discovery_window()
            second = self._instance()
            second.discovery_progress_path = path
            second._reserve_discovery_window()

            self.assertEqual(0, first.discovery_window_generation)
            self.assertEqual(1, second.discovery_window_generation)
            self.assertEqual(64, second._window_start(64))
            second.content_scan_tick = 2
            self.assertEqual(128, second._window_start(64))
            self.assertEqual(
                2, json.loads(path.read_text(encoding="utf-8"))["generation"]
            )

    def test_invalid_discovery_progress_fails_closed(self) -> None:
        with tempfile.TemporaryDirectory() as raw:
            path = Path(raw) / "discovery-progress.json"
            path.write_text('{"generation": true}', encoding="utf-8")
            supervisor = self._instance()
            supervisor.discovery_progress_path = path

            with self.assertRaises(JsonInputError):
                supervisor._reserve_discovery_window()

    def test_containment_orders_direct_kill_before_receipt_state_and_cleanup(self) -> None:
        supervisor = self._instance()
        response = {"result": "TERMINATED", "cleanup": {"attempted": False}}
        with patch("lumi_eggcracker.supervisor.validate_identity", return_value=Path("/owned")), patch("lumi_eggcracker.supervisor.kill_path", return_value=(11, 12)), patch("lumi_eggcracker.supervisor.verify_empty", return_value=(13, EmptyProof(True, 1, 0, []))), patch.object(supervisor, "_cleanup", side_effect=lambda _: supervisor.operations.append("cleanup") or {}), patch("lumi_eggcracker.supervisor.make_receipt", return_value=response), patch("lumi_eggcracker.supervisor.write_atomic"), patch.object(supervisor, "_store", side_effect=lambda *_: supervisor.operations.append("durable-state")):
            result = supervisor._contain(record(), "OPERATOR", 10)
        self.assertEqual("TERMINATED", result["result"])
        self.assertEqual("cgroup.kill", supervisor.operations[0])
        self.assertLess(
            supervisor.operations.index("durable-state"),
            supervisor.operations.index("durable-receipt"),
        )
        self.assertLess(supervisor.operations.index("cgroup.kill"), supervisor.operations.index("durable-receipt"))
        self.assertLess(supervisor.operations.index("durable-receipt"), supervisor.operations.index("cleanup"))

    def test_owned_containment_claim_covers_kill_and_empty_proof(self) -> None:
        supervisor = self._instance()
        item = record()

        def kill_claimed(_path: Path) -> tuple[int, int]:
            self.assertIn(item["cgroup"], supervisor.owned_containment_cgroups)
            return 11, 12

        def verify_claimed(_identity: object) -> tuple[int, EmptyProof]:
            self.assertIn(item["cgroup"], supervisor.owned_containment_cgroups)
            return 13, EmptyProof(True, 1, 0, [])

        with (
            patch("lumi_eggcracker.supervisor.validate_identity", return_value=Path("/owned")),
            patch("lumi_eggcracker.supervisor.kill_path", side_effect=kill_claimed),
            patch("lumi_eggcracker.supervisor.verify_empty", side_effect=verify_claimed),
            patch.object(supervisor, "_cleanup", return_value={}),
            patch(
                "lumi_eggcracker.supervisor.make_receipt",
                return_value={"result": "TERMINATED"},
            ),
            patch("lumi_eggcracker.supervisor.write_atomic"),
            patch.object(supervisor, "_store"),
        ):
            supervisor._contain(item, "OPERATOR", 10)

        self.assertNotIn(item["cgroup"], supervisor.owned_containment_cgroups)

    def test_receipt_persistence_fault_is_terminal_but_never_reports_success(self) -> None:
        supervisor = self._instance()
        item = record()
        stored: list[str] = []
        with patch("lumi_eggcracker.supervisor.validate_identity", return_value=Path("/owned")), patch("lumi_eggcracker.supervisor.kill_path", return_value=(11, 12)), patch("lumi_eggcracker.supervisor.verify_empty", return_value=(13, EmptyProof(True, 1, 0, []))), patch.object(supervisor, "_write_receipt", side_effect=OSError("disk full")), patch.object(supervisor, "_store", side_effect=lambda value: stored.append(str(value["state"]))), patch.object(supervisor, "_cleanup") as cleanup, self.assertRaisesRegex(JsonInputError, "receipt persistence failed"):
            supervisor._contain(item, "OPERATOR", 10)
        self.assertEqual("CONTAINED_RECEIPT_FAILED", item["state"])
        self.assertIn("CONTAINED_RECEIPT_FAILED", stored)
        cleanup.assert_not_called()

    def test_exact_empty_cgroup_allows_normal_completion_without_systemctl(self) -> None:
        supervisor = self._instance()
        saved: list[dict[str, object]] = []
        with patch("lumi_eggcracker.supervisor.events_from_fd", return_value={"populated": 0}), patch.object(supervisor, "_store", side_effect=lambda value: saved.append(value.copy())):
            self.assertTrue(supervisor._complete_allowed(record(), 42))
        self.assertEqual("COMPLETED_ALLOWED", saved[0]["state"])

    def test_collected_empty_cgroup_allows_normal_completion_with_exact_proof(self) -> None:
        supervisor = self._instance()
        item = record()
        with patch.object(supervisor, "_watch_once", side_effect=JsonInputError("owned cgroup is unavailable")), patch("lumi_eggcracker.supervisor.verify_empty", return_value=(1, EmptyProof(True, 0, 0, []))), patch.object(supervisor, "_store") as stored:
            supervisor._watch(item)
        stored.assert_called_once()
        self.assertEqual("COMPLETED_ALLOWED", item["state"])

    def test_recovery_accepts_only_an_exact_empty_collected_cgroup(self) -> None:
        supervisor = self._instance()
        item = record()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root
            supervisor.names = root / "names"
            supervisor.names.mkdir()
            (supervisor.runs / ("a" * 24 + ".json")).write_text(__import__("json").dumps(item), encoding="utf-8")
            with patch("lumi_eggcracker.supervisor.boot_id", return_value="b" * 36), patch.object(supervisor, "_contain", side_effect=JsonInputError("containment failed: owned cgroup is unavailable")), patch("lumi_eggcracker.supervisor.verify_empty", return_value=(1, EmptyProof(True, 0, 0, []))), patch.object(supervisor, "_mark_completed") as completed:
                supervisor._recover()
        completed.assert_called_once_with(item)

    def test_recovery_promotes_collected_failure_after_exact_empty_proof(self) -> None:
        supervisor = self._instance()
        item = record()

        def fail_containment(value: dict[str, object], _trigger: str) -> None:
            value["state"] = "CONTAINMENT_FAILED"
            raise JsonInputError("containment failed: owned cgroup is unavailable")

        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root
            supervisor.names = root / "names"
            supervisor.names.mkdir()
            (supervisor.runs / ("a" * 24 + ".json")).write_text(
                __import__("json").dumps(item), encoding="utf-8"
            )
            with patch("lumi_eggcracker.supervisor.boot_id", return_value="b" * 36), patch.object(supervisor, "_contain", side_effect=fail_containment), patch(
                "lumi_eggcracker.supervisor.verify_empty",
                return_value=(1, EmptyProof(True, 0, 0, [])),
            ), patch.object(supervisor, "_mark_completed") as completed:
                supervisor._recover()
        self.assertEqual("RUNNING", item["state"])
        completed.assert_called_once_with(item)

    def test_recovered_empty_transition_is_durable_and_terminal(self) -> None:
        supervisor = self._instance()
        item = record()
        item["state"] = "CONTAINMENT_FAILED"
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            supervisor.runs.mkdir()
            supervisor.names.mkdir()
            with patch.object(supervisor, "_teardown_terminal_boundary"):
                supervisor._complete_recovered_empty(item)
            stored = load_run(supervisor.runs, "a" * 24)
        self.assertEqual("COMPLETED_ALLOWED", stored["state"])
        self.assertNotIn(stored["cgroup"], supervisor.active_cgroups)

    def test_recovery_reconciles_absent_prior_boot_workload(self) -> None:
        supervisor = self._instance()
        item = record()
        with tempfile.TemporaryDirectory() as raw:
            supervisor.runs = Path(raw)
            (supervisor.runs / ("a" * 24 + ".json")).write_text(
                json.dumps(item), encoding="utf-8"
            )
            with (
                patch("lumi_eggcracker.supervisor.boot_id", return_value="c" * 36),
                patch.object(
                    supervisor, "_record_cgroup_present", return_value=False
                ),
                patch.object(supervisor, "_complete_recovered_empty") as completed,
                patch.object(supervisor, "_contain") as contain,
            ):
                supervisor._recover()
        completed.assert_called_once_with(item)
        contain.assert_not_called()

    def test_recovery_rejects_prior_boot_cgroup_name_collision(self) -> None:
        supervisor = self._instance()
        item = record()
        with tempfile.TemporaryDirectory() as raw:
            supervisor.runs = Path(raw)
            (supervisor.runs / ("a" * 24 + ".json")).write_text(
                json.dumps(item), encoding="utf-8"
            )
            with (
                patch("lumi_eggcracker.supervisor.boot_id", return_value="c" * 36),
                patch.object(
                    supervisor, "_record_cgroup_present", return_value=True
                ),
                self.assertRaisesRegex(JsonInputError, "prior-boot workload"),
            ):
                supervisor._recover()

    def test_recovery_does_not_erase_receipt_persistence_failure(self) -> None:
        supervisor = self._instance()
        item = record()

        def fail_receipt(value: dict[str, object], _trigger: str) -> None:
            value["state"] = "CONTAINED_RECEIPT_FAILED"
            raise JsonInputError("contained but receipt persistence failed")

        with tempfile.TemporaryDirectory() as raw:
            supervisor.runs = Path(raw)
            (supervisor.runs / ("a" * 24 + ".json")).write_text(
                json.dumps(item), encoding="utf-8"
            )
            with (
                patch("lumi_eggcracker.supervisor.boot_id", return_value="b" * 36),
                patch.object(supervisor, "_contain", side_effect=fail_receipt),
                patch("lumi_eggcracker.supervisor.verify_empty") as empty,
                self.assertRaisesRegex(JsonInputError, "receipt persistence"),
            ):
                supervisor._recover()
        empty.assert_not_called()

    def test_recovery_reclaims_exact_terminal_offline_boundary(self) -> None:
        supervisor = self._instance()
        supervisor.boundary_cleanup_healthy = True
        item = record()
        item["state"] = "TERMINATED"
        item["boundary"] = {"recorded": True}
        supervisor.runs = MagicMock()
        supervisor.runs.glob.return_value = [Path("a" * 24 + ".json")]
        with (
            patch("lumi_eggcracker.supervisor.load_run", return_value=item),
            patch.object(supervisor, "_teardown_boundary") as teardown,
        ):
            supervisor._recover()
        teardown.assert_called_once_with(item)
        self.assertTrue(supervisor.boundary_cleanup_healthy)

    def test_recovery_reports_terminal_boundary_identity_cleanup_failure(self) -> None:
        supervisor = self._instance()
        supervisor.boundary_cleanup_healthy = True
        item = record()
        item["state"] = "CONTAINED_RECEIPT_FAILED"
        item["boundary"] = {"recorded": True}
        supervisor.runs = MagicMock()
        supervisor.runs.glob.return_value = [Path("a" * 24 + ".json")]
        with (
            patch("lumi_eggcracker.supervisor.load_run", return_value=item),
            patch.object(
                supervisor,
                "_teardown_boundary",
                side_effect=JsonInputError("boundary identity drifted"),
            ),
        ):
            supervisor._recover()
        self.assertFalse(supervisor.boundary_cleanup_healthy)

    def test_status_is_read_only(self) -> None:
        supervisor = self._instance()
        item = record()
        with patch.object(supervisor, "_load", return_value=item), patch.object(supervisor, "_store") as stored:
            value = supervisor.handle({"action": "status", "args": {"name": "demo"}})
        self.assertEqual("RUNNING", value["state"])
        stored.assert_not_called()

    def test_status_resolves_retained_terminal_history(self) -> None:
        supervisor = self._instance()
        item = record()
        item["state"] = "COMPLETED_ALLOWED"
        with patch.object(supervisor, "_load", side_effect=JsonInputError("gone")), patch.object(supervisor, "_latest_by_name", return_value=item), patch.object(supervisor, "_store") as stored:
            value = supervisor.handle({"action": "status", "args": {"name": "demo"}})
        self.assertEqual("COMPLETED_ALLOWED", value["state"])
        stored.assert_not_called()

    def test_history_empty_result_is_explicitly_read_only(self) -> None:
        supervisor = self._instance()
        supervisor.runs = MagicMock()
        supervisor.runs.glob.return_value = []
        with (
            patch.object(supervisor, "_incident_status", side_effect=AssertionError),
            patch.object(supervisor, "_store") as stored,
            patch.object(supervisor, "_prune_terminal_records") as pruned,
            patch.object(supervisor, "_teardown_boundary") as teardown,
        ):
            self.assertEqual(
                {"runs": []},
                supervisor.handle({"action": "history", "args": {"limit": 20}}),
            )
        stored.assert_not_called()
        pruned.assert_not_called()
        teardown.assert_not_called()

    def test_history_returns_only_terminal_records_in_deterministic_order(self) -> None:
        supervisor = self._instance()
        active = retained_record("a" * 24, "RUNNING", 900)
        completed = retained_record("b" * 24, "COMPLETED_ALLOWED", 20)
        terminated_low_id = retained_record("c" * 24, "TERMINATED", 20)
        terminated_high_id = retained_record("d" * 24, "TERMINATED", 20)
        values = {
            item["run_id"]: item
            for item in (active, completed, terminated_low_id, terminated_high_id)
        }
        supervisor.runs = MagicMock()
        supervisor.runs.glob.return_value = [
            Path(f"{run_id}.json") for run_id in values
        ]

        def load(_runs: Path, run_id: str) -> dict[str, object]:
            return values[run_id]

        with (
            patch("lumi_eggcracker.supervisor.load_run", side_effect=load),
            patch.object(supervisor, "_incident_status", side_effect=AssertionError),
        ):
            response = supervisor.handle({"action": "history", "args": {"limit": 3}})

        self.assertEqual(
            ["d" * 24, "c" * 24, "b" * 24],
            [item["run_id"] for item in response["runs"]],
        )
        self.assertEqual(
            ["TERMINATED", "TERMINATED", "COMPLETED_ALLOWED"],
            [item["state"] for item in response["runs"]],
        )
        for item in response["runs"]:
            self.assertEqual(
                {
                    "exec_policy_digest",
                    "exec_policy_id",
                    "incident_lockdown",
                    "name",
                    "network_mode",
                    "run_id",
                    "state",
                    "unit",
                    "workload_uid",
                },
                set(item),
            )
            self.assertNotIn("created_monotonic_ns", item)

    def test_history_filters_each_supported_state(self) -> None:
        supervisor = self._instance()
        states = (
            "COMPLETED_ALLOWED",
            "TERMINATED",
            "CONTAINMENT_FAILED",
            "CONTAINED_RECEIPT_FAILED",
        )
        values = {
            f"{index:024x}": retained_record(f"{index:024x}", state, index)
            for index, state in enumerate(states, start=1)
        }
        supervisor.runs = MagicMock()
        supervisor.runs.glob.return_value = [
            Path(f"{run_id}.json") for run_id in values
        ]

        def load(_runs: Path, run_id: str) -> dict[str, object]:
            return values[run_id]

        for index, state in enumerate(states, start=1):
            with self.subTest(state=state), patch(
                "lumi_eggcracker.supervisor.load_run", side_effect=load
            ):
                response = supervisor.handle(
                    {"action": "history", "args": {"limit": 20, "state": state}}
                )
            self.assertEqual([f"{index:024x}"], [item["run_id"] for item in response["runs"]])
            self.assertEqual([state], [item["state"] for item in response["runs"]])

    def test_history_filters_before_sort_and_limit(self) -> None:
        supervisor = self._instance()
        older_id = "a" * 24
        older_abnormal = retained_record(older_id, "CONTAINMENT_FAILED", 1)
        newer_successes = [
            retained_record(f"{index:024x}", "COMPLETED_ALLOWED", index + 100)
            for index in range(1, 23)
        ]
        values = {
            item["run_id"]: item for item in [older_abnormal, *newer_successes]
        }
        supervisor.runs = MagicMock()
        supervisor.runs.glob.return_value = [
            Path(f"{run_id}.json") for run_id in values
        ]

        def load(_runs: Path, run_id: str) -> dict[str, object]:
            return values[run_id]

        with patch("lumi_eggcracker.supervisor.load_run", side_effect=load) as load_run:
            response = supervisor.handle(
                {
                    "action": "history",
                    "args": {"limit": 1, "state": "CONTAINMENT_FAILED"},
                }
            )

        self.assertEqual([older_id], [item["run_id"] for item in response["runs"]])
        self.assertEqual("CONTAINMENT_FAILED", response["runs"][0]["state"])
        self.assertEqual(len(values), load_run.call_count)

    def test_history_rejects_invalid_limit_types_ranges_and_extra_arguments(self) -> None:
        supervisor = self._instance()
        cases = (
            {},
            {"limit": 0},
            {"limit": 21},
            {"limit": True},
            {"limit": False},
            {"limit": 1.0},
            {"limit": "1"},
            {"limit": 1, "extra": None},
            {"limit": 1, "state": "RUNNING"},
            {"limit": 1, "state": "terminated"},
            {"limit": 1, "state": None},
            {"limit": 1, "state": True},
            {"limit": 1, "state": "TERMINATED", "extra": None},
        )
        for args in cases:
            with self.subTest(args=args), self.assertRaises(JsonInputError):
                supervisor.handle({"action": "history", "args": args})

    def test_history_rejects_malformed_retained_record_instead_of_proving_empty(self) -> None:
        supervisor = self._instance()
        supervisor.runs = MagicMock()
        valid_id = "a" * 24
        malformed_id = "b" * 24
        supervisor.runs.glob.return_value = [
            Path(f"{valid_id}.json"),
            Path(f"{malformed_id}.json"),
        ]

        def load(_runs: Path, run_id: str) -> dict[str, object]:
            if run_id == valid_id:
                return retained_record(valid_id, "COMPLETED_ALLOWED", 1)
            raise JsonInputError("private malformed record")

        for args in (
            {"limit": 20},
            {"limit": 20, "state": "TERMINATED"},
        ):
            with self.subTest(args=args), patch(
                "lumi_eggcracker.supervisor.load_run", side_effect=load
            ), self.assertRaisesRegex(
                JsonInputError, "retained workload history is unavailable"
            ):
                supervisor.handle({"action": "history", "args": args})

    def test_history_does_not_read_receipts_or_run_mutation_paths(self) -> None:
        supervisor = self._instance()
        item = retained_record("a" * 24, "TERMINATED", 1)
        supervisor.runs = MagicMock()
        supervisor.runs.glob.return_value = [Path(f"{'a' * 24}.json")]
        supervisor.receipts = MagicMock()
        supervisor.receipts.glob.side_effect = AssertionError("receipt read")
        with (
            patch("lumi_eggcracker.supervisor.load_run", return_value=item),
            patch("lumi_eggcracker.supervisor.load_regular_json", side_effect=AssertionError),
            patch.object(supervisor, "_store") as stored,
            patch.object(supervisor, "_prune_terminal_records") as pruned,
            patch.object(supervisor, "_teardown_boundary") as teardown,
            patch.object(supervisor, "_contain") as contain,
        ):
            response = supervisor.handle({"action": "history", "args": {"limit": 20}})
        self.assertEqual(["a" * 24], [value["run_id"] for value in response["runs"]])
        stored.assert_not_called()
        pruned.assert_not_called()
        teardown.assert_not_called()
        contain.assert_not_called()

    def test_terminal_run_keeps_name_tombstone(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            item = record()
            supervisor._store(item)
            self.assertTrue((supervisor.runs / ("a" * 24 + ".json")).is_file())
            self.assertTrue((supervisor.names / "demo.json").is_file())
            item["state"] = "TERMINATED"
            supervisor._store(item)
            self.assertEqual(
                {"run_id": "a" * 24},
                __import__("json").loads(
                    (supervisor.names / "demo.json").read_text(encoding="utf-8")
                ),
            )

    def test_terminal_name_tombstone_is_not_listed_as_active(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            item = record()
            supervisor._store(item)
            item["state"] = "TERMINATED"
            supervisor._store(item)
            self.assertEqual({"runs": []}, supervisor.handle({"action": "list", "args": {}}))

    def test_backfill_reserves_names_from_legacy_terminal_records(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            supervisor.runs.mkdir()
            supervisor.names.mkdir()
            item = record()
            item["state"] = "TERMINATED"
            (supervisor.runs / ("a" * 24 + ".json")).write_text(
                __import__("json").dumps(item), encoding="utf-8"
            )
            supervisor._backfill_name_tombstones()
            self.assertTrue((supervisor.names / "demo.json").is_file())

    def test_store_rejects_a_second_run_using_a_reserved_name(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            first = record()
            supervisor._store(first)
            first["state"] = "TERMINATED"
            supervisor._store(first)
            replacement_id = "c" * 24
            replacement = {
                **record(),
                "cgroup": f"/system.slice/lumi-eggcracker-workload-{replacement_id}.service",
                "run_id": replacement_id,
                "unit": f"lumi-eggcracker-workload-{replacement_id}.service",
            }
            with self.assertRaisesRegex(JsonInputError, "name index collision"):
                supervisor._store(replacement)
            self.assertFalse((supervisor.runs / ("c" * 24 + ".json")).exists())

    def test_terminal_pruning_never_removes_the_record_just_persisted(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            historical = {
                **record(),
                "cgroup": "/system.slice/lumi-eggcracker-workload-"
                + "b" * 24
                + ".service",
                "created_monotonic_ns": 10_000,
                "name": "historical",
                "run_id": "b" * 24,
                "state": "TERMINATED",
                "unit": "lumi-eggcracker-workload-" + "b" * 24 + ".service",
            }
            current = {
                **record(),
                "created_monotonic_ns": 1,
                "state": "TERMINATED",
            }
            with patch("lumi_eggcracker.supervisor.MAX_TERMINAL_RECORDS", 1):
                supervisor._store(historical)
                supervisor._store(current)

            self.assertTrue((supervisor.runs / ("a" * 24 + ".json")).is_file())
            self.assertFalse((supervisor.runs / ("b" * 24 + ".json")).exists())

    def test_terminal_pruning_retains_boundary_cleanup_authority(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            supervisor.runs = Path(raw)
            pending_path = supervisor.runs / ("a" * 24 + ".json")
            clean_path = supervisor.runs / ("b" * 24 + ".json")
            pending_path.write_text("{}", encoding="utf-8")
            clean_path.write_text("{}", encoding="utf-8")
            pending = {
                **record(),
                "boundary": {"recorded": True},
                "created_monotonic_ns": 1,
                "state": "TERMINATED",
            }
            clean = {
                **record(),
                "created_monotonic_ns": 2,
                "run_id": "b" * 24,
                "state": "TERMINATED",
            }
            boundary = MagicMock()
            boundary.namespace_mounts_present.return_value = True

            with (
                patch("lumi_eggcracker.supervisor.MAX_TERMINAL_RECORDS", 0),
                patch(
                    "lumi_eggcracker.supervisor.load_run",
                    side_effect=lambda _root, run_id: (
                        pending if run_id == "a" * 24 else clean
                    ),
                ),
                patch(
                    "lumi_eggcracker.supervisor.OfflineBoundary.from_record",
                    return_value=boundary,
                ),
            ):
                supervisor._prune_terminal_records()

            self.assertTrue(pending_path.is_file())
            self.assertFalse(clean_path.exists())

    def test_terminal_boundary_teardown_retries_pruning_after_identity_release(
        self,
    ) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            supervisor.runs = Path(raw)
            current_path = supervisor.runs / ("a" * 24 + ".json")
            historical_path = supervisor.runs / ("b" * 24 + ".json")
            current_path.write_text("{}", encoding="utf-8")
            historical_path.write_text("{}", encoding="utf-8")
            current = {
                **record(),
                "boundary": {"recorded": True},
                "created_monotonic_ns": 1,
                "state": "TERMINATED",
            }
            historical = {
                **record(),
                "boundary": {"recorded": True},
                "created_monotonic_ns": 10_000,
                "name": "historical",
                "run_id": "b" * 24,
                "state": "TERMINATED",
                "unit": "lumi-eggcracker-workload-" + "b" * 24 + ".service",
            }
            boundary = MagicMock()
            boundary.namespace_mounts_present.return_value = False

            with (
                patch("lumi_eggcracker.supervisor.MAX_TERMINAL_RECORDS", 1),
                patch.object(
                    supervisor,
                    "_teardown_boundary",
                    return_value={"removed": True},
                ) as teardown,
                patch(
                    "lumi_eggcracker.supervisor.load_run",
                    side_effect=lambda _root, run_id: (
                        current if run_id == "a" * 24 else historical
                    ),
                ),
                patch(
                    "lumi_eggcracker.supervisor.OfflineBoundary.from_record",
                    return_value=boundary,
                ),
            ):
                result = supervisor._teardown_terminal_boundary(current)

            self.assertEqual({"removed": True}, result)
            teardown.assert_called_once_with(current)
            self.assertTrue(current_path.is_file())
            self.assertFalse(historical_path.exists())

    def test_terminal_boundary_teardown_does_not_prune_after_cleanup_failure(
        self,
    ) -> None:
        supervisor = self._instance()
        item = {**record(), "boundary": {"recorded": True}, "state": "TERMINATED"}
        with (
            patch.object(
                supervisor,
                "_teardown_boundary",
                side_effect=JsonInputError("boundary identity drifted"),
            ),
            patch.object(supervisor, "_prune_terminal_records") as prune,
            self.assertRaisesRegex(JsonInputError, "identity drifted"),
        ):
            supervisor._teardown_terminal_boundary(item)
        prune.assert_not_called()

    def test_autonomous_kill_terminal_state_wins_completion_race(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            item = record()
            supervisor._store(item)
            stale_watcher_record = item.copy()
            terminated = supervisor._mark_discovered_runs_terminated(
                {str(item["cgroup"])}
            )
            self.assertEqual(1, len(terminated))
            self.assertEqual(item["run_id"], terminated[0]["run_id"])
            self.assertEqual("TERMINATED", terminated[0]["state"])
            self.assertEqual(
                "TERMINATED", load_run(supervisor.runs, str(item["run_id"]))["state"]
            )
            self.assertFalse(supervisor._mark_completed(stale_watcher_record))
            self.assertEqual(
                "TERMINATED", load_run(supervisor.runs, str(item["run_id"]))["state"]
            )

    def test_live_admitted_pid_outside_empty_cgroup_cannot_complete_allowed(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            supervisor.launches = root / "launches"
            item = record()
            supervisor._store(item)
            provenance = {
                "cgroup": item["cgroup"],
                "pid": 42,
                "run_id": item["run_id"],
                "start_time": 100,
            }
            with (
                patch(
                    "lumi_eggcracker.supervisor.load_launch_provenance",
                    return_value=[provenance],
                ),
                patch(
                    "lumi_eggcracker.supervisor.process_identity",
                    return_value=ProcessIdentity(42, 100),
                ),
            ):
                self.assertFalse(supervisor._mark_completed(item.copy()))
            self.assertEqual(
                "RUNNING", load_run(supervisor.runs, str(item["run_id"]))["state"]
            )

    def test_discovery_attributes_moved_exact_launch_identity_to_owned_run(self) -> None:
        supervisor = self._instance()
        supervisor.launches = Path("launches")
        item = record()
        snapshot = MagicMock(
            identity=ProcessIdentity(42, 100),
            cgroups=("0::/r2-43-escape",),
        )
        provenance = {
            "cgroup": item["cgroup"],
            "pid": 42,
            "run_id": item["run_id"],
            "start_time": 100,
        }
        with patch(
            "lumi_eggcracker.supervisor.load_launch_provenance",
            return_value=[provenance],
        ):
            self.assertEqual(
                {item["cgroup"]}, supervisor._discovered_run_cgroups(snapshot, ())
            )

    def test_affected_approval_is_captured_from_the_exact_owned_run(self) -> None:
        supervisor = self._instance()
        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            supervisor.launches = root / "launches"
            item = record()
            supervisor._store(item)
            approval = {"name": "exact-approval"}
            with patch.object(
                supervisor, "_approval_identity", return_value=approval
            ) as approval_identity:
                self.assertIs(
                    approval,
                    supervisor._approval_for_discovered_cgroups(
                        {str(item["cgroup"])}
                    ),
                )
            approval_identity.assert_called_once_with(item["run_id"])

    def test_autonomous_kill_reclaims_owned_boundary_after_receipt_publish(self) -> None:
        supervisor = self._instance()
        supervisor.quarantine_root = Path("/sys/fs/cgroup/quarantine")
        target = ProcessIdentity(42, 100)
        snapshot = MagicMock(identity=target)
        detected = MagicMock(profile="content.gguf-llama")
        terminated_record = record()
        terminated_record["state"] = "TERMINATED"
        terminated_record["boundary"] = {"recorded": True}
        order: list[str] = []

        with (
            patch.object(
                supervisor,
                "_discovered_run_cgroups",
                return_value={str(terminated_record["cgroup"])},
            ),
            patch("lumi_eggcracker.supervisor.contain_many", return_value=MagicMock()),
            patch.object(
                supervisor,
                "_detection_receipt",
                return_value={
                    "event_id": "e" * 24,
                    "trigger": {"kind": "UNAPPROVED_AI_MATCH"},
                },
            ),
            patch.object(
                supervisor,
                "_mark_discovered_runs_terminated",
                return_value=(terminated_record,),
            ),
            patch.object(
                supervisor,
                "_store_detection",
                side_effect=lambda _receipt: order.append("publish"),
            ),
            patch.object(
                supervisor,
                "_teardown_boundary",
                side_effect=lambda _record: order.append("teardown"),
            ) as teardown,
            patch.object(supervisor, "_post_containment_response"),
        ):
            supervisor._enforce_discovery(
                snapshot,
                detected,
                (),
                (),
                1,
                2,
                None,
                "e" * 24,
                targets={target},
            )

        self.assertEqual(["publish", "teardown"], order)
        teardown.assert_called_once_with(terminated_record)

    def test_autonomous_boundary_cleanup_failure_fails_health_not_kill(self) -> None:
        supervisor = self._instance()
        supervisor.boundary_cleanup_healthy = True
        item = record()
        item["state"] = "TERMINATED"
        item["boundary"] = {"recorded": True}
        with patch.object(
            supervisor,
            "_teardown_boundary",
            side_effect=JsonInputError("boundary identity drifted"),
        ):
            supervisor._cleanup_discovered_run_boundaries((item,))
        self.assertFalse(supervisor.boundary_cleanup_healthy)

    def test_health_rejects_terminal_run_with_namespace_mounts(self) -> None:
        supervisor = self._instance()
        item = record()
        item["state"] = "TERMINATED"
        item["boundary"] = {"recorded": True}
        supervisor.runs = MagicMock()
        supervisor.runs.glob.return_value = [Path("a" * 24 + ".json")]
        boundary = MagicMock()
        boundary.namespace_mounts_present.return_value = True
        with (
            patch("lumi_eggcracker.supervisor.load_run", return_value=item),
            patch(
                "lumi_eggcracker.supervisor.OfflineBoundary.from_record",
                return_value=boundary,
            ),
        ):
            self.assertFalse(supervisor._stored_boundary_cleanup_healthy())

    def test_autonomous_kill_marks_owned_run_before_publishing_receipt(self) -> None:
        supervisor = self._instance()
        supervisor.quarantine_root = Path("/sys/fs/cgroup/quarantine")
        target = ProcessIdentity(42, 100)
        snapshot = MagicMock(identity=target)
        detected = MagicMock(profile="content.gguf-llama")
        containment = MagicMock()
        order: list[str] = []
        terminated_record = record()
        terminated_record["state"] = "TERMINATED"
        affected_approval = {
            "argv_sha256": "a" * 64,
            "bound_input_sha256": ["b" * 64],
            "created_monotonic_ns": 1,
            "executable_sha256": "c" * 64,
            "name": "approved-run",
            "uid": 2001,
        }

        def mark(cgroups: set[str]) -> tuple[dict[str, object], ...]:
            order.append("mark")
            self.assertEqual({str(terminated_record["cgroup"])}, cgroups)
            return (terminated_record,)

        with (
            patch.object(
                supervisor,
                "_discovered_run_cgroups",
                return_value={
                    "/system.slice/lumi-eggcracker-workload-"
                    + "a" * 24
                    + ".service"
                },
            ),
            patch(
                "lumi_eggcracker.supervisor.contain_many",
                side_effect=lambda *_args, **_kwargs: order.append("contain")
                or containment,
            ),
            patch.object(
                supervisor,
                "_detection_receipt",
                return_value={
                    "event_id": "e" * 24,
                    "trigger": {"kind": "UNAPPROVED_AI_MATCH"},
                },
            ),
            patch.object(
                supervisor,
                "_approval_for_discovered_cgroups",
                side_effect=lambda _cgroups: order.append("approval")
                or affected_approval,
            ),
            patch.object(
                supervisor,
                "_mark_discovered_runs_terminated",
                side_effect=mark,
            ),
            patch.object(
                supervisor,
                "_store_detection",
                side_effect=lambda _receipt: order.append("publish"),
            ),
            patch.object(supervisor, "_post_containment_response") as respond,
        ):
            supervisor._enforce_discovery(
                snapshot,
                detected,
                (),
                (),
                1,
                2,
                None,
                "e" * 24,
                targets={target},
            )

        self.assertEqual(["contain", "approval", "mark", "publish"], order)
        response_receipt = respond.call_args.args[0]
        self.assertEqual("e" * 24, response_receipt["event_id"])
        self.assertIs(terminated_record, respond.call_args.kwargs["record"])
        self.assertIs(affected_approval, respond.call_args.kwargs["approval"])

    def test_owned_containment_claim_suppresses_redundant_autonomous_kill(self) -> None:
        supervisor = self._instance()
        supervisor.quarantine_root = Path("/sys/fs/cgroup/quarantine")
        target = ProcessIdentity(42, 100)
        snapshot = MagicMock(identity=target)
        detected = MagicMock(profile="content.gguf-llama")
        cgroup = "/system.slice/lumi-eggcracker-workload-" + "a" * 24 + ".service"
        supervisor.owned_containment_cgroups.add(cgroup)
        supervisor.discovery_active.add(target)

        with (
            patch.object(
                supervisor,
                "_discovered_run_cgroups",
                return_value={cgroup},
            ),
            patch("lumi_eggcracker.supervisor.contain_many") as contain,
            patch.object(supervisor, "_store_detection") as store_detection,
        ):
            supervisor._enforce_discovery(
                snapshot,
                detected,
                (),
                (),
                1,
                2,
                None,
                "e" * 24,
                targets={target},
            )

        contain.assert_not_called()
        store_detection.assert_not_called()
        self.assertNotIn(target, supervisor.discovery_active)
        self.assertIn(cgroup, supervisor.owned_containment_cgroups)

    def test_autonomous_kill_barrier_blocks_transient_allowed_completion(self) -> None:
        supervisor = self._instance()
        supervisor.quarantine_root = Path("/sys/fs/cgroup/quarantine")
        target = ProcessIdentity(42, 100)
        snapshot = MagicMock(identity=target)
        detected = MagicMock(profile="content.gguf-llama")
        containment = MagicMock()
        completion_results: list[bool] = []

        with tempfile.TemporaryDirectory() as raw:
            root = Path(raw)
            supervisor.runs = root / "runs"
            supervisor.names = root / "names"
            item = record()
            supervisor._store(item)

            def contain_while_watcher_observes_empty(*_args: object, **_kwargs: object) -> object:
                completion_results.append(supervisor._mark_completed(item.copy()))
                self.assertEqual(
                    "RUNNING",
                    load_run(supervisor.runs, str(item["run_id"]))["state"],
                )
                return containment

            with (
                patch.object(
                    supervisor,
                    "_discovered_run_cgroups",
                    return_value={str(item["cgroup"])},
                ),
                patch(
                    "lumi_eggcracker.supervisor.contain_many",
                    side_effect=contain_while_watcher_observes_empty,
                ),
                patch.object(
                    supervisor,
                    "_detection_receipt",
                    return_value={
                        "event_id": "e" * 24,
                        "trigger": {"kind": "UNAPPROVED_AI_MATCH"},
                    },
                ),
                patch.object(supervisor, "_store_detection"),
                patch.object(supervisor, "_post_containment_response"),
            ):
                supervisor._enforce_discovery(
                    snapshot,
                    detected,
                    (),
                    (),
                    1,
                    2,
                    None,
                    "e" * 24,
                    targets={target},
                )

            self.assertEqual([False], completion_results)
            self.assertEqual(
                "TERMINATED",
                load_run(supervisor.runs, str(item["run_id"]))["state"],
            )

    def test_correlation_requires_live_same_uid_parent_or_sibling_relation(self) -> None:
        supervisor = self._instance()
        root = ProcessIdentity(10, 100)
        left_identity = ProcessIdentity(11, 101)
        right_identity = ProcessIdentity(12, 102)
        left = ProcessSnapshot(left_identity, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), (), parent=root)
        right = ProcessSnapshot(right_identity, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), (), parent=root)
        root_snapshot = ProcessSnapshot(root, 2001, "/usr/bin/worker", "worker", ("worker",), (), (), ())
        content = ArtifactEvidence("safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64)
        runtime = RuntimeEvidence("pytorch-bridge-aten-pair-pinned-cpu", "PyTorch/ATen", "BUILD_ID_PAIR", ())
        left_candidate = _EvidenceCandidate(left, (content,), (), 1)
        right_candidate = _EvidenceCandidate(right, (), (runtime,), 2)
        related, boundary = supervisor._related(left_candidate, right_candidate, {root: root_snapshot, left_identity: left, right_identity: right})
        self.assertTrue(related)
        self.assertEqual("sibling", boundary)
        unrelated_root = ProcessIdentity(1, 1)
        right_unrelated = ProcessSnapshot(right_identity, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), (), parent=unrelated_root)
        root_snapshot = ProcessSnapshot(unrelated_root, 0, "/sbin/init", "init", ("init",), (), (), ())
        related, _boundary = supervisor._related(left_candidate, _EvidenceCandidate(right_unrelated, (), (runtime,), 2), {unrelated_root: root_snapshot, left_identity: left, right_identity: right_unrelated})
        self.assertFalse(related)

    def test_related_pair_is_synthesized_but_unrelated_pair_is_not(self) -> None:
        supervisor = self._instance()
        common = ProcessIdentity(30, 300)
        left_identity = ProcessIdentity(31, 301)
        right_identity = ProcessIdentity(32, 302)
        left = ProcessSnapshot(left_identity, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), (), parent=common)
        right = ProcessSnapshot(right_identity, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), (), parent=common)
        root = ProcessSnapshot(common, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), ())
        content = ArtifactEvidence("safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64)
        bridge = RuntimeEvidence(PYTORCH_BRIDGE_EVIDENCE_ID, "PyTorch/ATen", "BUILD_ID", ())
        aten = RuntimeEvidence(PYTORCH_ATEN_EVIDENCE_ID, "PyTorch/ATen", "BUILD_ID", ())
        related_groups = supervisor._correlate(
            [_EvidenceCandidate(left, (content,), (bridge,), 1), _EvidenceCandidate(right, (), (aten,), 2)],
            {common: root, left_identity: left, right_identity: right},
        )
        self.assertEqual(1, len(related_groups))
        group = related_groups[0][0]
        runtimes = with_pytorch_pair(item for candidate in group for item in candidate.runtimes)
        self.assertIsNotNone(
            match(
                load_bundled(),
                left,
                evidence={
                    "MODEL_CONTENT": {content.evidence_id},
                    "MODEL_RUNTIME": {item.evidence_id for item in runtimes},
                },
            )
        )
        unrelated_root = ProcessIdentity(40, 400)
        unrelated = ProcessSnapshot(right_identity, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), (), parent=unrelated_root)
        unrelated_groups = supervisor._correlate(
            [_EvidenceCandidate(left, (content,), (bridge,), 1), _EvidenceCandidate(unrelated, (), (aten,), 2)],
            {common: root, left_identity: left, unrelated_root: ProcessSnapshot(unrelated_root, 0, "/sbin/init", "init", ("init",), (), (), ()), right_identity: unrelated},
        )
        self.assertEqual(2, len(unrelated_groups))
        self.assertTrue(
            all(
                match(
                    load_bundled(),
                    group[0].snapshot,
                    evidence={
                        "MODEL_CONTENT": {item.evidence_id for item in group[0].content},
                        "MODEL_RUNTIME": {item.evidence_id for item in with_pytorch_pair(group[0].runtimes)},
                    },
                )
                is None
                for group, _boundary in unrelated_groups
            )
        )

    def test_orphaned_roles_in_exact_active_selected_cgroup_are_related(self) -> None:
        supervisor = self._instance()
        selected = "/system.slice/lumi-eggcracker-workload-" + "a" * 24 + ".service"
        supervisor.active_cgroups = {selected}
        init = ProcessIdentity(1, 1)
        left_identity = ProcessIdentity(31, 301)
        right_identity = ProcessIdentity(32, 302)
        left = ProcessSnapshot(
            left_identity,
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            ("0::" + selected,),
            (),
            (),
            parent=init,
        )
        right = ProcessSnapshot(
            right_identity,
            2001,
            "/usr/bin/python3",
            "python3",
            ("python3",),
            ("0::" + selected,),
            (),
            (),
            parent=init,
        )
        content = ArtifactEvidence(
            "safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64
        )
        runtime = RuntimeEvidence(
            "pytorch-bridge-aten-pair-pinned-cpu",
            "PyTorch/ATen",
            "BUILD_ID_PAIR",
            (),
        )
        left_candidate = _EvidenceCandidate(left, (content,), (), 1)
        right_candidate = _EvidenceCandidate(right, (), (runtime,), 2)

        with patch("lumi_eggcracker.supervisor.Path.is_symlink", return_value=False), patch(
            "lumi_eggcracker.supervisor.Path.is_dir", return_value=True
        ), patch("lumi_eggcracker.supervisor.Path.is_file", return_value=True):
            related, boundary = supervisor._related(
                left_candidate,
                right_candidate,
                {left_identity: left, right_identity: right},
            )
            self.assertTrue(related)
            self.assertEqual("owned-cgroup", boundary)

            supervisor.active_cgroups.clear()
            related, _boundary = supervisor._related(
                left_candidate,
                right_candidate,
                {left_identity: left, right_identity: right},
            )
            self.assertFalse(related)

    def test_unrelated_same_uid_partial_candidates_do_not_join_by_common_root(self) -> None:
        supervisor = self._instance()
        common = ProcessIdentity(1, 1)
        left_id = ProcessIdentity(21, 201)
        right_id = ProcessIdentity(22, 202)
        left = ProcessSnapshot(left_id, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), (), parent=common)
        right = ProcessSnapshot(right_id, 2001, "/usr/bin/python3", "python3", ("python3",), (), (), (), parent=common)
        root = ProcessSnapshot(common, 0, "/sbin/init", "init", ("init",), (), (), ())
        content = ArtifactEvidence("safetensors-v1", "SAFETENSORS", 1, 2, 4096, "a" * 64)
        runtime = RuntimeEvidence("pytorch-bridge-aten-pair-pinned-cpu", "PyTorch/ATen", "BUILD_ID_PAIR", ())
        with patch.object(supervisor, "_owned_cgroup", return_value=None):
            groups = supervisor._correlate(
                [_EvidenceCandidate(left, (content,), (), 1), _EvidenceCandidate(right, (), (runtime,), 2)],
                {common: root, left_id: left, right_id: right},
            )
        self.assertEqual(2, len(groups))
        self.assertTrue(all(len(value[0]) == 1 for value in groups))