Handle collected empty workload cgroups

src/lumi_eggcracker/supervisor.py

@@ -684,6 +684,9 @@ class Supervisor:
         events = events_from_fd(cgroup_events_fd)
         if events.get("populated") != 0:
             return False
+        return self._mark_completed(record)
+
+    def _mark_completed(self, record: dict[str, Any]) -> bool:
         lock = self._new_lock(record["run_id"])
         with lock:
             if record["state"] not in ACTIVE_STATES:
@@ -727,6 +730,14 @@ class Supervisor:
                 self._watch_once(record, ready)
                 return
             except (JsonInputError, OSError, RuntimeError) as error:
+                # A collected transient unit is accepted only through the same
+                # exact-empty proof used after a direct cgroup kill.  It is
+                # not inferred from systemd's inactive state.
+                if "owned cgroup is unavailable" in str(error):
+                    _empty_ns, proof = verify_empty(identity_from_run(record))
+                    if proof.complete:
+                        self._mark_completed(record)
+                        return
                 failure = error
                 if attempt == 0:
                     continue

tests/test_supervisor.py

@@ -59,6 +59,14 @@ class SupervisorTests(unittest.TestCase):
             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_status_is_read_only(self) -> None:
         supervisor = self._instance()
         item = record()