Preserve newly terminated run history

CHANGELOG.md

@@ -16,6 +16,9 @@
 - Link a successful autonomous incident to the exact owned workload record
   after containment, allowing recovery tooling to identify the selected run
   without replacing the authoritative quarantine kill receipt.
+- Never prune the terminal run record in the transaction that just persisted
+  it, avoiding cross-boot monotonic-clock ordering from erasing the newest kill
+  status under a full retained history.
 
 ## 1.0.0 (release candidate)
 

src/lumi_eggcracker/supervisor.py

@@ -407,7 +407,7 @@ class Supervisor:
         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"])
-        self._prune_terminal_records()
+        self._prune_terminal_records(protected_run_id=record["run_id"])
 
     @staticmethod
     def _clear_stage(run_id: str) -> None:
@@ -428,7 +428,7 @@ class Supervisor:
             child.unlink()
         root.rmdir()
 
-    def _prune_terminal_records(self) -> None:
+    def _prune_terminal_records(self, *, protected_run_id: str | None = None) -> None:
         records: list[tuple[int, Path]] = []
         for path in self.runs.glob("*.json"):
             try:
@@ -437,7 +437,13 @@ class Supervisor:
                 continue
             if value["state"] not in ACTIVE_STATES:
                 records.append((int(value["created_monotonic_ns"]), path))
-        for _, path in sorted(records, reverse=True)[MAX_TERMINAL_RECORDS:]:
+        remove_count = max(0, len(records) - MAX_TERMINAL_RECORDS)
+        removable = [
+            value
+            for value in sorted(records)
+            if value[1].stem != protected_run_id
+        ]
+        for _, path in removable[:remove_count]:
             path.unlink(missing_ok=True)
 
     def _load(self, name: str) -> dict[str, Any]:

tests/test_supervisor.py

@@ -895,6 +895,35 @@ class SupervisorTests(unittest.TestCase):
                 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_autonomous_kill_terminal_state_wins_completion_race(self) -> None:
         supervisor = self._instance()
         with tempfile.TemporaryDirectory() as raw: