Drain checkpoint readers before deciding final sealed state

docs/final-reader-ledger.org

@@ -0,0 +1,13 @@
+#+TITLE: Final checkpoint reader ordering (issue 70)
+* 1. Task
+On base c09d361, drain the runner-owned capture before deciding whether a clean exit can seal.
+* 2. Existing machinery
+Checkpointer.final stops scheduling and waits for inflight under one checkpoint timeout, then passes remaining budget to begin. run currently scans holders first. holders_of includes the current process, intentionally. Existing final deadline tests and InProcess fixture cover captures and stalls. Search: direct callers; GitNexus impact failed loading libssl.so.3.
+* 3. Definitions
+Sealed requires clean non-misplaced exit and no observed database holders. Crash remains distinct from sealed recovery. Final capture itself opens readers after the quiescence check.
+* 4. Invariants
+No holder exemption, extra timeout, remote stat, or relaxed restore gate. A stalled inflight capture skips final publication. Holder scanning spends the same remaining final budget.
+* 5. Reuse
+Move the existing holder check into Checkpointer.final after inflight drains, before remaining-budget calculation and begin. run supplies the clean-exit candidate class. Reuse holders_of unchanged.
+* 6. Questions
+None. Verify owned reader release, persistent external holder, crash candidate, exhausted deadline, real restore and reuse. Self-review only.

sucoder/codex_state_runner.py

@@ -3182,6 +3182,13 @@ def final(self, cls):
             log("final %s capture skipped: the durable generations listing never completed, so nothing could be published; "
                 "last durable generation: %s%s" % (cls, self.describe_last(), self.describe_unsynced()))
             return "the final capture was skipped: the durable generations listing never completed"
+        # Quiesce owned readers before class selection (final-reader ledger, sections 3-5).
+        if cls == "sealed":
+            holders = holders_of(self.db_dir)
+            if holders:
+                cls = "crash"
+                log("pids %s still hold descriptors under %s after Codex exited; the final generation is a crash candidate, not sealed" % (
+                    ", ".join(str(pid) for pid in holders[:4]), self.db_dir))
         remaining = deadline - time.monotonic()
         if remaining <= 0:
             log("final %s capture skipped: the %ds shutdown budget was spent waiting for the capture in flight; last durable "
@@ -3807,12 +3814,6 @@ def forward(signum, frame):
         status.update("exited", exit=code, outcome=result.get("outcome"))
         log("Codex exited with %d (%s)" % (code, result.get("outcome", "unknown")))
     cls = "sealed" if code == 0 and not misplaced else "crash"
-    if cls == "sealed":
-        holders = holders_of(db_dir)
-        if holders:
-            cls = "crash"
-            log("pids %s still hold descriptors under %s after Codex exited; the final generation is a crash candidate, not sealed" % (
-                ", ".join(str(pid) for pid in holders[:4]), db_dir))
     skipped = checkpointer.final(cls)
     request = drain.settle(code, skipped)
     if request is not None and request["mode"] == "renew":

tests/test_codex_state_runner.py

@@ -4164,3 +4164,40 @@ def grown(st):
         assert b.quit() == 0
     finally:
         b.close()
+
+@pytest.mark.parametrize("candidate,held,expected", [("sealed", False, "sealed"), ("sealed", True, "crash"), ("crash", False, "crash")])
+def test_final_class_checks_holders_after_owned_capture_drains(runner, tmp_path, monkeypatch, candidate, held, expected):
+    h = InProcess(runner, tmp_path, timeout=2)
+    ready, release = threading.Event(), threading.Event()
+    def reader():
+        connection = sqlite3.connect(str(h.db_dir / "state_5.sqlite"))
+        connection.execute("SELECT name FROM sqlite_master").fetchall()
+        ready.set()
+        release.wait(5)
+        connection.close()
+        with h.cp.inflight_lock:
+            h.cp.inflight = None
+    h.cp.inflight = {"generation": "000001-reader"}
+    thread = threading.Thread(target=reader)
+    thread.start()
+    assert ready.wait(2)
+    assert os.getpid() in runner.holders_of(str(h.db_dir))
+    original = runner.holders_of
+    observed = []
+    def holders(directory):
+        assert h.cp.inflight is None
+        observed.append(original(directory))
+        return [999999] if held else observed[-1]
+    monkeypatch.setattr(runner, "holders_of", holders)
+    classes = []
+    h.cp.capture = lambda generation, trigger, cls, deadline, final_pass: classes.append(cls)
+    timer = threading.Timer(0.2, release.set)
+    timer.start()
+    try:
+        assert h.cp.final(candidate) is None
+        assert classes == [expected]
+        assert observed == ([[]] if candidate == "sealed" else [])
+    finally:
+        release.set()
+        thread.join(3)
+        timer.join()