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))