fix: reserve workload names after termination

LIMITATIONS.md

@@ -46,4 +46,6 @@ behavioural modelling, telemetry upload, malware prevention or EDR.
 
 Approvals are exact and deliberately narrow: only a qualified native llama runtime or root-owned CPython executing one absolute regular script is supported. Workload UID, resolved executable identity and complete argv digest must match at the protected pre-exec `start` gate. Python `-c`, `-m`, stdin/interactive, relative-script, package-directory, symlink and unsupported-interpreter forms are rejected. The Python script is identity/digest bound and launched from an immutable per-run root-owned stage; changing it aborts `start`. This does not claim a general dependency-closure analyser or make approved workloads a sandbox. A direct or third-party launch cannot consume an approval, and descendants or siblings do not inherit one. Revoking an approval affects future protected launches, not a workload already running under that approval.
 
+Operator-supplied workload names are single-use for the lifetime of an installation. A name remains reserved after completion or containment so a stale status or kill command cannot resolve to a replacement workload. Clean uninstallation removes those reservations with the rest of Eggcracker's local state.
+
 The explicit `start` command remains non-interactive. It has no terminal attachment, current-working-directory forwarding, timeout or output streaming. It supports one active selected workload globally; PID, memory and CPU limits are explicit. Detection receipts are bounded to 1000 entries.

src/lumi_eggcracker/supervisor.py

@@ -383,22 +383,26 @@ class Supervisor:
         record = validate_run(record)
         if not hasattr(self, "active_cgroups"):
             self.active_cgroups = set()
-        if record["state"] in ACTIVE_STATES:
-            self.active_cgroups.add(record["cgroup"])
-        else:
-            self.active_cgroups.discard(record["cgroup"])
-        self.operations.append("durable-state")
-        write_atomic(run_path(self.runs, record["run_id"]), record)
         pointer = name_path(self.names, record["name"])
-        if record["state"] in ACTIVE_STATES:
-            write_atomic(pointer, {"run_id": record["run_id"]})
-        elif pointer.exists():
+        expected_pointer = {"run_id": record["run_id"]}
+        pointer_missing = not pointer.exists() and not pointer.is_symlink()
+        if not pointer_missing:
             try:
                 current = load_regular_json(pointer)
-                if current == {"run_id": record["run_id"]}:
-                    pointer.unlink()
             except JsonInputError:
                 raise JsonInputError("workload name index is invalid")
+            if current != expected_pointer:
+                raise JsonInputError("workload name index collision")
+        if pointer_missing:
+            # A reserved name is safe even if the following record write
+            # fails; allowing reuse after a partial start is not.
+            write_atomic(pointer, expected_pointer)
+        self.operations.append("durable-state")
+        write_atomic(run_path(self.runs, record["run_id"]), record)
+        if record["state"] in ACTIVE_STATES:
+            self.active_cgroups.add(record["cgroup"])
+        else:
+            self.active_cgroups.discard(record["cgroup"])
         if record["state"] not in ACTIVE_STATES and hasattr(self, "launches"):
             provenance_path(self.launches, record["run_id"]).unlink(missing_ok=True)
             self._clear_stage(record["run_id"])
@@ -445,7 +449,7 @@ class Supervisor:
         return record
 
     def _latest_by_name(self, name: str) -> dict[str, Any]:
-        """Resolve a completed run only when its name is no longer active."""
+        """Resolve retained history for an unavailable or completed name."""
         candidates: list[dict[str, Any]] = []
         for path in self.runs.glob("*.json"):
             try:
@@ -458,6 +462,33 @@ class Supervisor:
             raise JsonInputError("workload name is unavailable")
         return max(candidates, key=lambda item: int(item["created_monotonic_ns"]))
 
+    def _backfill_name_tombstones(self) -> None:
+        """Reserve names created by releases that removed terminal pointers."""
+        latest: dict[str, dict[str, Any]] = {}
+        for path in self.runs.glob("*.json"):
+            try:
+                record = load_run(self.runs, path.stem)
+            except JsonInputError:
+                continue
+            prior = latest.get(record["name"])
+            if prior is None or int(record["created_monotonic_ns"]) > int(
+                prior["created_monotonic_ns"]
+            ):
+                latest[record["name"]] = record
+        for name, record in latest.items():
+            pointer = name_path(self.names, name)
+            if pointer.exists() or pointer.is_symlink():
+                value = load_regular_json(pointer)
+                run_id = value.get("run_id")
+                if (
+                    set(value) != {"run_id"}
+                    or not isinstance(run_id, str)
+                    or not RUN_ID.fullmatch(run_id)
+                ):
+                    raise JsonInputError("workload name index is invalid")
+                continue
+            write_atomic(pointer, {"run_id": record["run_id"]})
+
     def _receipt_path(self, event_id: str) -> Path:
         return self.receipts / f"{event_id}.json"
 
@@ -531,6 +562,7 @@ class Supervisor:
             incident_store.compact(self.incidents)
         except (JsonInputError, OSError):
             self.incident_health_healthy = False
+        self._backfill_name_tombstones()
         self._recover()
         self.quarantine_root = self._prepare_quarantine()
         self._scan_synchronously()
@@ -2532,7 +2564,12 @@ class Supervisor:
         if not self._incident_status()["healthy"]:
             raise JsonInputError("local lockdown state is unavailable; root recovery is required")
         with self.start_lock, self.approval_lock:
-            if name_path(self.names, name).exists() or self._active_exists():
+            reserved_name = name_path(self.names, name)
+            if (
+                reserved_name.exists()
+                or reserved_name.is_symlink()
+                or self._active_exists()
+            ):
                 raise JsonInputError(
                     "one protected workload is already active or name is unavailable"
                 )
@@ -2980,11 +3017,7 @@ class Supervisor:
             }
         if action == "start":
             return self._start(args)
-        if action == "status" and set(args) == {"name"}:
-            try:
-                record = self._load(args["name"])
-            except JsonInputError:
-                record = self._latest_by_name(args["name"])
+        def public_run(record: dict[str, Any]) -> dict[str, Any]:
             return {
                 "name": record["name"],
                 "run_id": record["run_id"],
@@ -3000,11 +3033,20 @@ class Supervisor:
                 "incident_lockdown": self._incident_status()["lockdown"],
                 "workload_uid": record["workload_uid"],
             }
+
+        if action == "status" and set(args) == {"name"}:
+            try:
+                record = self._load(args["name"])
+            except JsonInputError:
+                record = self._latest_by_name(args["name"])
+            return public_run(record)
         if action == "list" and not args:
-            runs = [
-                self.handle({"action": "status", "args": {"name": path.stem}})
-                for path in sorted(self.names.glob("*.json"))
-            ]
+            runs = []
+            for path in sorted(self.names.glob("*.json")):
+                try:
+                    runs.append(public_run(self._load(path.stem)))
+                except JsonInputError:
+                    continue
             return {"runs": runs}
         if action in {"approve", "revoke", "exec_policy_create", "exec_policy_revoke"} and not self._incident_status()["healthy"]:
             raise JsonInputError("local lockdown state is unavailable; root recovery is required")

tests/test_supervisor.py

@@ -817,7 +817,7 @@ class SupervisorTests(unittest.TestCase):
         self.assertEqual("RUNNING", value["state"])
         stored.assert_not_called()
 
-    def test_status_resolves_the_latest_terminal_run_after_name_reuse_is_enabled(self) -> None:
+    def test_status_resolves_retained_terminal_history(self) -> None:
         supervisor = self._instance()
         item = record()
         item["state"] = "COMPLETED_ALLOWED"
@@ -826,7 +826,7 @@ class SupervisorTests(unittest.TestCase):
         self.assertEqual("COMPLETED_ALLOWED", value["state"])
         stored.assert_not_called()
 
-    def test_run_records_are_keyed_by_run_id_and_name_pointer_is_removed_when_terminal(self) -> None:
+    def test_terminal_run_keeps_name_tombstone(self) -> None:
         supervisor = self._instance()
         with tempfile.TemporaryDirectory() as raw:
             root = Path(raw)
@@ -838,7 +838,61 @@ class SupervisorTests(unittest.TestCase):
             self.assertTrue((supervisor.names / "demo.json").is_file())
             item["state"] = "TERMINATED"
             supervisor._store(item)
-            self.assertFalse((supervisor.names / "demo.json").exists())
+            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_autonomous_kill_terminal_state_wins_completion_race(self) -> None:
         supervisor = self._instance()