fix(reconcile): retire workflow session rows whose owners are no longer live

Implement a sanctioned session lifecycle for post-restart reconciliation so
active session rows with dead or reused owner PIDs can be terminalized
without deleting history. Protect live owners, live leases, and live
client-managed sessions; record durable session_retired audit events; keep
cleanup idempotent under concurrent reconciles.

Closes #969

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

tests/test_control_plane_db.py

@@ -37,7 +37,7 @@ class ControlPlaneDBTest(unittest.TestCase):
             rows = dict(conn.execute("SELECT key, value FROM schema_meta").fetchall())
         finally:
             conn.close()
-        self.assertEqual(rows["schema_version"], "5")
+        self.assertEqual(rows["schema_version"], "6")
         self.assertIn("DB coordinates", rows["architecture"])
         self.assertIn("bridge", rows["architecture"].lower())
 
@@ -868,7 +868,7 @@ class SessionCheckpointTest(unittest.TestCase):
             conn.close()
         self.assertIn("session_checkpoints", names)
         record = self._write()
-        self.assertEqual(record["checkpoint_schema_version"], 5)
+        self.assertEqual(record["checkpoint_schema_version"], 6)
 
     # AC2 — checkpoints written for multi-role session fixtures.
     def test_multi_role_fixtures_each_get_a_row(self) -> None:

tests/test_issue_969_session_lifecycle.py

@@ -0,0 +1,480 @@
+"""Tests for dead-owner / PID-reuse session retirement (#969)."""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import threading
+import unittest
+from concurrent.futures import ThreadPoolExecutor, as_completed
+from datetime import datetime, timedelta, timezone
+
+import control_plane_db as cpd
+import post_restart_reconcile as prr
+import session_lifecycle as sl
+
+NOW = datetime(2026, 7, 29, 12, 0, 0, tzinfo=timezone.utc)
+EARLIER = NOW - timedelta(hours=2)
+LATER = NOW + timedelta(minutes=5)
+
+
+def _session(
+    session_id: str,
+    *,
+    pid: int | None = 4242,
+    status: str = "active",
+    started_at: datetime = EARLIER,
+    last_heartbeat_at: datetime | None = None,
+    client_managed: bool = False,
+    owner_process_started_at: datetime | None = None,
+    role: str = "author",
+) -> dict:
+    hb = last_heartbeat_at if last_heartbeat_at is not None else started_at
+    row = {
+        "session_id": session_id,
+        "role": role,
+        "profile": "prgs-author",
+        "pid": pid,
+        "status": status,
+        "started_at": cpd._ts(started_at),
+        "last_heartbeat_at": cpd._ts(hb),
+        "client_managed": client_managed,
+    }
+    if owner_process_started_at is not None:
+        row["owner_process_started_at"] = cpd._ts(owner_process_started_at)
+    return row
+
+
+def _alive(pids: set[int]):
+    def _check(pid):
+        try:
+            return int(pid) in pids
+        except (TypeError, ValueError):
+            return False
+
+    return _check
+
+
+def _starts(mapping: dict[int, datetime]):
+    def _probe(pid):
+        try:
+            return mapping.get(int(pid))
+        except (TypeError, ValueError):
+            return None
+
+    return _probe
+
+
+class ClassifyDeadOwnerTests(unittest.TestCase):
+    def test_dead_owner_is_stale_and_retireable(self) -> None:
+        c = sl.classify_session(
+            _session("ghost", pid=2_000_000_000),
+            now=NOW,
+            pid_checker=_alive(set()),
+            process_start_probe=_starts({}),
+        )
+        self.assertEqual(c.classification, sl.CLASS_STALE)
+        self.assertTrue(c.retireable)
+        self.assertIn(c.reason, {sl.REASON_DEAD_OWNER, sl.REASON_HEARTBEAT_STALE_DEAD})
+
+    def test_missing_pid_is_stale(self) -> None:
+        c = sl.classify_session(
+            _session("no-pid", pid=None),
+            now=NOW,
+            pid_checker=_alive(set()),
+        )
+        self.assertEqual(c.classification, sl.CLASS_STALE)
+        self.assertEqual(c.reason, sl.REASON_MISSING_PID)
+        self.assertTrue(c.retireable)
+
+
+class PidReuseTests(unittest.TestCase):
+    def test_pid_reuse_marks_stale_not_live(self) -> None:
+        # Process with same PID started AFTER the session was recorded.
+        c = sl.classify_session(
+            _session("reused", pid=77, started_at=EARLIER, last_heartbeat_at=EARLIER),
+            now=NOW,
+            pid_checker=_alive({77}),
+            process_start_probe=_starts({77: LATER}),
+        )
+        self.assertEqual(c.classification, sl.CLASS_STALE)
+        self.assertEqual(c.reason, sl.REASON_PID_REUSE)
+        self.assertTrue(c.pid_reused)
+        self.assertTrue(c.retireable)
+
+    def test_matching_process_start_is_live(self) -> None:
+        c = sl.classify_session(
+            _session(
+                "same-proc",
+                pid=88,
+                started_at=EARLIER,
+                last_heartbeat_at=NOW - timedelta(seconds=30),
+                owner_process_started_at=EARLIER - timedelta(seconds=5),
+            ),
+            now=NOW,
+            pid_checker=_alive({88}),
+            process_start_probe=_starts({88: EARLIER - timedelta(seconds=5)}),
+        )
+        self.assertEqual(c.classification, sl.CLASS_LIVE)
+        self.assertFalse(c.retireable)
+
+
+class LiveOwnerAndLeaseTests(unittest.TestCase):
+    def test_live_owner_not_retired(self) -> None:
+        c = sl.classify_session(
+            _session(
+                "live",
+                pid=os.getpid(),
+                last_heartbeat_at=NOW - timedelta(seconds=10),
+            ),
+            now=NOW,
+            pid_checker=_alive({os.getpid()}),
+            process_start_probe=_starts({os.getpid(): EARLIER}),
+        )
+        self.assertEqual(c.classification, sl.CLASS_LIVE)
+        self.assertFalse(c.retireable)
+
+    def test_live_lease_blocks_retirement_even_if_pid_dead(self) -> None:
+        c = sl.classify_session(
+            _session("leased", pid=99999),
+            now=NOW,
+            pid_checker=_alive(set()),
+            live_lease_sessions={"leased"},
+        )
+        self.assertEqual(c.classification, sl.CLASS_PROTECTED)
+        self.assertEqual(c.reason, sl.REASON_LIVE_LEASE)
+        self.assertFalse(c.retireable)
+
+    def test_client_managed_live_never_retired(self) -> None:
+        c = sl.classify_session(
+            _session(
+                "client",
+                pid=55,
+                client_managed=True,
+                last_heartbeat_at=NOW - timedelta(seconds=5),
+            ),
+            now=NOW,
+            pid_checker=_alive({55}),
+            process_start_probe=_starts({55: EARLIER}),
+        )
+        self.assertEqual(c.classification, sl.CLASS_LIVE)
+        self.assertEqual(c.reason, sl.REASON_CLIENT_MANAGED_LIVE)
+        self.assertFalse(c.retireable)
+
+    def test_client_managed_dead_pid_is_retireable(self) -> None:
+        # Dead client process is not a live client-managed session.
+        c = sl.classify_session(
+            _session("client-dead", pid=56, client_managed=True),
+            now=NOW,
+            pid_checker=_alive(set()),
+        )
+        self.assertEqual(c.classification, sl.CLASS_STALE)
+        self.assertTrue(c.retireable)
+
+
+class TerminalAndDisconnectedTests(unittest.TestCase):
+    def test_already_terminal_not_retireable(self) -> None:
+        c = sl.classify_session(
+            _session("done", status="retired", pid=1),
+            now=NOW,
+            pid_checker=_alive(set()),
+        )
+        self.assertEqual(c.classification, sl.CLASS_TERMINAL)
+        self.assertFalse(c.retireable)
+
+    def test_alive_stale_heartbeat_is_disconnected_not_retired(self) -> None:
+        c = sl.classify_session(
+            _session(
+                "quiet",
+                pid=66,
+                last_heartbeat_at=NOW - timedelta(hours=5),
+            ),
+            now=NOW,
+            pid_checker=_alive({66}),
+            process_start_probe=_starts({66: EARLIER}),
+        )
+        self.assertEqual(c.classification, sl.CLASS_DISCONNECTED)
+        self.assertFalse(c.retireable)
+
+
+class FleetAndApplyTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.db_path = os.path.join(self._tmp.name, "cp.sqlite3")
+        self.db = cpd.ControlPlaneDB(self.db_path)
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def test_mixed_fleet_and_apply_retires_only_stale(self) -> None:
+        live_pid = os.getpid()
+        # Use wall-clock "now" so upsert timestamps align with classification.
+        moment = datetime.now(timezone.utc)
+        proc_start = moment - timedelta(hours=1)
+        self.db.upsert_session(
+            session_id="s-live",
+            role="author",
+            pid=live_pid,
+            status="active",
+            owner_process_started_at=cpd._ts(proc_start),
+        )
+        self.db.upsert_session(
+            session_id="s-dead", role="reviewer", pid=2_000_000_001, status="active"
+        )
+        self.db.upsert_session(
+            session_id="s-ended", role="merger", pid=3, status="ended"
+        )
+
+        sessions = self.db.list_sessions(limit=50)
+        report = sl.classify_sessions(
+            sessions,
+            leases=[],
+            now=moment,
+            pid_checker=_alive({live_pid}),
+            process_start_probe=_starts({live_pid: proc_start}),
+        )
+        self.assertGreaterEqual(report.stale_count, 1)
+        self.assertTrue(
+            any(c.session_id == "s-dead" and c.retireable for c in report.classifications)
+        )
+        self.assertTrue(
+            any(
+                c.session_id == "s-live" and not c.retireable
+                for c in report.classifications
+            )
+        )
+
+        first = sl.apply_session_retirements(
+            self.db, report, dry_run=False, actor_session_id="actor-1", now=moment
+        )
+        self.assertGreaterEqual(first["retired_count"], 1)
+        # After retirement, active list should exclude s-dead.
+        active = {
+            s["session_id"]
+            for s in self.db.list_sessions(statuses=("active",), limit=50)
+        }
+        self.assertNotIn("s-dead", active)
+        self.assertIn("s-live", active)
+
+        # Repeated cleanup is idempotent.
+        report2 = sl.classify_sessions(
+            self.db.list_sessions(limit=50),
+            leases=[],
+            now=moment,
+            pid_checker=_alive({live_pid}),
+            process_start_probe=_starts({live_pid: proc_start}),
+        )
+        second = sl.apply_session_retirements(
+            self.db, report2, dry_run=False, actor_session_id="actor-1", now=moment
+        )
+        # No double-retirement of the same row as a new mutation.
+        self.assertEqual(second["retired_count"], 0)
+
+        # Durable audit event present.
+        import sqlite3
+
+        conn = sqlite3.connect(self.db_path)
+        try:
+            events = conn.execute(
+                "SELECT event_type, message FROM events WHERE event_type = ?",
+                ("session_retired",),
+            ).fetchall()
+        finally:
+            conn.close()
+        self.assertTrue(events)
+        self.assertTrue(any("s-dead" in (m or "") for _, m in events))
+
+    def test_live_lease_blocks_db_retirement(self) -> None:
+        self.db.upsert_session(
+            session_id="s-leased", role="author", pid=2_000_000_002, status="active"
+        )
+        self.db.upsert_work_item(
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=969,
+        )
+        result = self.db.assign_and_lease(
+            session_id="s-leased",
+            role="author",
+            remote="prgs",
+            org="org",
+            repo="repo",
+            kind="issue",
+            number=969,
+        )
+        self.assertEqual(result.outcome, "assigned")
+
+        # Inventory-style lease with explicit live freshness (authoritative for
+        # the pure classifier). DB apply also blocks on the active lease row.
+        leases = [
+            {
+                "lease_id": result.lease_id,
+                "session_id": "s-leased",
+                "status": "active",
+                "freshness": {"freshness": "active"},
+            }
+        ]
+        report = sl.classify_sessions(
+            self.db.list_sessions(statuses=("active",), limit=20),
+            leases=leases,
+            now=NOW,
+            pid_checker=_alive(set()),
+        )
+        # Classifier protects via live lease set.
+        self.assertTrue(
+            any(
+                c.session_id == "s-leased" and c.classification == sl.CLASS_PROTECTED
+                for c in report.classifications
+            )
+        )
+        apply = sl.apply_session_retirements(
+            self.db, report, dry_run=False, actor_session_id="actor", now=NOW
+        )
+        self.assertEqual(apply["retired_count"], 0)
+        active = {
+            s["session_id"]
+            for s in self.db.list_sessions(statuses=("active",), limit=20)
+        }
+        self.assertIn("s-leased", active)
+
+        # Direct DB CAS also refuses while an active lease row remains.
+        blocked = self.db.retire_session(
+            session_id="s-leased",
+            reason=sl.REASON_DEAD_OWNER,
+            actor_session_id="actor",
+            now=NOW,
+        )
+        self.assertEqual(blocked["outcome"], "blocked")
+        self.assertEqual(blocked["reason"], "live_lease")
+
+    def test_concurrent_retirement_is_idempotent(self) -> None:
+        for i in range(20):
+            self.db.upsert_session(
+                session_id=f"ghost-{i}",
+                role="author",
+                pid=3_000_000 + i,
+                status="active",
+            )
+
+        def _worker() -> dict:
+            return sl.retire_stale_sessions(
+                self.db,
+                dry_run=False,
+                actor_session_id=f"actor-{threading.get_ident()}",
+                now=NOW,
+                pid_checker=_alive(set()),
+                process_start_probe=_starts({}),
+                session_limit=100,
+            )
+
+        outcomes = []
+        with ThreadPoolExecutor(max_workers=4) as pool:
+            futs = [pool.submit(_worker) for _ in range(4)]
+            for fut in as_completed(futs):
+                outcomes.append(fut.result())
+
+        total_retired = sum(o["apply"]["retired_count"] for o in outcomes)
+        # Exactly one successful retirement per ghost row across all workers.
+        self.assertEqual(total_retired, 20)
+        active = {
+            s["session_id"]
+            for s in self.db.list_sessions(statuses=("active",), limit=100)
+        }
+        for i in range(20):
+            self.assertNotIn(f"ghost-{i}", active)
+
+
+class ReconcileIntegrationTests(unittest.TestCase):
+    def test_unresolved_until_retired_then_resolved(self) -> None:
+        inv = {
+            "inventory_complete": True,
+            "incomplete_reasons": [],
+            "service_health": {"healthy": True},
+            "clients": [{"session_id": "c1", "connected": True}],
+            "sessions": [
+                _session("ghost", pid=2_000_000_099, last_heartbeat_at=EARLIER),
+            ],
+            "leases": [],
+            "checkpoints_available": False,
+            "worktree_bindings": [],
+            "pending_mutations": [],
+            "capabilities": {"stale": False},
+            "boot_head_sha": "a" * 40,
+            "current_head_sha": "a" * 40,
+            "queue_state": {"safe_to_resume": True},
+        }
+        proof = prr.reconcile_after_restart(inv, now=NOW, mode=prr.MODE_LOG_ONLY)
+        sess = next(i for i in proof.items if i.dimension == prr.DIM_SESSIONS)
+        self.assertEqual(sess.status, prr.ITEM_UNRESOLVED)
+        self.assertIn("ghost", sess.details.get("orphan_session_ids") or [])
+
+        # After retirement inventory (no active orphans) resolves.
+        inv2 = dict(inv)
+        inv2["sessions"] = []
+        inv2["session_fleet"] = {
+            "retireable_session_ids": [],
+            "sessions_dimension_resolved": True,
+            "live_count": 0,
+            "stale_count": 0,
+        }
+        proof2 = prr.reconcile_after_restart(inv2, now=NOW, mode=prr.MODE_LOG_ONLY)
+        sess2 = next(i for i in proof2.items if i.dimension == prr.DIM_SESSIONS)
+        self.assertEqual(sess2.status, prr.ITEM_RESOLVED)
+
+    def test_legacy_orphan_key_still_populated(self) -> None:
+        inv = {
+            "inventory_complete": True,
+            "service_health": {"healthy": True},
+            "clients": [],
+            "sessions": [_session("ghost", pid=2_000_000_100)],
+            "leases": [],
+            "checkpoints_available": False,
+            "worktree_bindings": [],
+            "pending_mutations": [],
+            "capabilities": {"stale": False},
+            "boot_head_sha": "a" * 40,
+            "current_head_sha": "a" * 40,
+            "queue_state": {"safe_to_resume": True},
+        }
+        proof = prr.reconcile_after_restart(inv, now=NOW)
+        sess = next(i for i in proof.items if i.dimension == prr.DIM_SESSIONS)
+        self.assertIn("orphan_session_ids", sess.details)
+
+
+class SchemaMigrationTests(unittest.TestCase):
+    def test_lifecycle_columns_present(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            path = os.path.join(tmp, "cp.sqlite3")
+            db = cpd.ControlPlaneDB(path)
+            db.upsert_session(session_id="s1", role="author", pid=1)
+            out = db.retire_session(
+                session_id="s1",
+                reason=sl.REASON_DEAD_OWNER,
+                actor_session_id="tester",
+                now=NOW,
+            )
+            self.assertEqual(out["outcome"], "retired")
+            rows = db.list_sessions(limit=5)
+            # May not appear under active filter
+            all_rows = db.list_sessions(limit=5)
+            # Re-open raw to check columns
+            import sqlite3
+
+            conn = sqlite3.connect(path)
+            try:
+                cols = {r[1] for r in conn.execute("PRAGMA table_info(sessions)")}
+                version = conn.execute(
+                    "SELECT value FROM schema_meta WHERE key='schema_version'"
+                ).fetchone()[0]
+            finally:
+                conn.close()
+            self.assertIn("retired_at", cols)
+            self.assertIn("retire_reason", cols)
+            self.assertIn("owner_process_started_at", cols)
+            self.assertEqual(version, "6")
+
+
+if __name__ == "__main__":
+    unittest.main()