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]>

control_plane_db.py

@@ -27,12 +27,12 @@ import uuid
 from contextlib import contextmanager
 from dataclasses import dataclass
 from datetime import datetime, timedelta, timezone
-from typing import Any, Iterator, Sequence
+from typing import Any, Iterator, Mapping, Sequence
 
 import dependency_graph
 import gitea_audit
 
-SCHEMA_VERSION = 5
+SCHEMA_VERSION = 6
 
 # Assignable work kinds only — raw monitoring incidents are never work items.
 WORK_KINDS = frozenset({"issue", "pr"})
@@ -419,6 +419,7 @@ class ControlPlaneDB:
                 self._migrate_incident_links_null_scope(conn)
                 self._migrate_lease_lifecycle_columns(conn)
                 self._migrate_session_ownership_columns(conn)
+                self._migrate_session_lifecycle_columns(conn)
                 self._migrate_usage_events_table(conn)
                 conn.execute(
                     "INSERT OR REPLACE INTO schema_meta(key, value) VALUES (?, ?)",
@@ -812,6 +813,7 @@ class ControlPlaneDB:
         pid: int | None = None,
         status: str = "active",
         controller_instance_id: str | None = None,
+        owner_process_started_at: str | None = None,
     ) -> dict[str, Any]:
         """Register/refresh a session row.
 
@@ -821,16 +823,22 @@ class ControlPlaneDB:
         controller instance can. It is never overwritten with ``None``, so a
         heartbeat from a caller that does not supply one cannot erase
         ownership.
+
+        *owner_process_started_at* (#969) is the OS start time of the owner
+        process at registration time. When present it is retained across
+        heartbeats (never cleared by ``None``) so later PID-reuse checks do
+        not depend on a live ``ps`` probe of a long-dead process.
         """
         now = _ts()
         instance = (controller_instance_id or "").strip() or None
+        proc_start = (owner_process_started_at or "").strip() or None
         with self._tx() as conn:
             existing = conn.execute(
                 "SELECT session_id FROM sessions WHERE session_id = ?",
                 (session_id,),
             ).fetchone()
             if existing:
-                if instance is None:
+                if instance is None and proc_start is None:
                     conn.execute(
                         """
                         UPDATE sessions
@@ -840,7 +848,21 @@ class ControlPlaneDB:
                         """,
                         (role, profile, namespace, pid, now, status, session_id),
                     )
-                else:
+                elif instance is None:
+                    conn.execute(
+                        """
+                        UPDATE sessions
+                        SET role = ?, profile = ?, namespace = ?, pid = ?,
+                            last_heartbeat_at = ?, status = ?,
+                            owner_process_started_at = COALESCE(?, owner_process_started_at)
+                        WHERE session_id = ?
+                        """,
+                        (
+                            role, profile, namespace, pid, now, status,
+                            proc_start, session_id,
+                        ),
+                    )
+                elif proc_start is None:
                     conn.execute(
                         """
                         UPDATE sessions
@@ -854,18 +876,33 @@ class ControlPlaneDB:
                             instance, session_id,
                         ),
                     )
+                else:
+                    conn.execute(
+                        """
+                        UPDATE sessions
+                        SET role = ?, profile = ?, namespace = ?, pid = ?,
+                            last_heartbeat_at = ?, status = ?,
+                            controller_instance_id = ?,
+                            owner_process_started_at = COALESCE(?, owner_process_started_at)
+                        WHERE session_id = ?
+                        """,
+                        (
+                            role, profile, namespace, pid, now, status,
+                            instance, proc_start, session_id,
+                        ),
+                    )
             else:
                 conn.execute(
                     """
                     INSERT INTO sessions(
                         session_id, role, profile, namespace, pid,
                         started_at, last_heartbeat_at, status,
-                        controller_instance_id
+                        controller_instance_id, owner_process_started_at
-                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
                     """,
                     (
                         session_id, role, profile, namespace, pid, now, now,
-                        status, instance,
+                        status, instance, proc_start,
                     ),
                 )
             row = conn.execute(
@@ -881,6 +918,165 @@ class ControlPlaneDB:
                 (_ts(), session_id),
             )
 
+    def retire_session(
+        self,
+        *,
+        session_id: str,
+        reason: str,
+        actor_session_id: str | None = None,
+        details: Mapping[str, Any] | None = None,
+        now: datetime | None = None,
+        terminal_status: str = "retired",
+    ) -> dict[str, Any]:
+        """Terminalize one session row if it is still non-terminal (#969).
+
+        CAS on non-terminal status: concurrent retirements of the same row
+        yield exactly one ``retired`` outcome and subsequent
+        ``already_terminal`` outcomes. Never deletes historical rows. Writes a
+        durable ``session_retired`` event for audit.
+        """
+        moment = _ts(now)
+        reason_s = (reason or "").strip() or "unspecified"
+        term = (terminal_status or "retired").strip().lower() or "retired"
+        terminal_set = {
+            "retired",
+            "ended",
+            "terminal",
+            "dead",
+            "stale",
+            "orphaned",
+        }
+        with self._tx() as conn:
+            row = conn.execute(
+                "SELECT * FROM sessions WHERE session_id = ?",
+                (session_id,),
+            ).fetchone()
+            if row is None:
+                return {
+                    "outcome": "missing",
+                    "reason": reason_s,
+                    "prior_status": None,
+                    "new_status": None,
+                    "details": {"session_id": session_id},
+                }
+            prior = dict(row)
+            prior_status = str(prior.get("status") or "").strip().lower()
+            if prior_status in terminal_set:
+                return {
+                    "outcome": "already_terminal",
+                    "reason": reason_s,
+                    "prior_status": prior_status,
+                    "new_status": prior_status,
+                    "details": {"session_id": session_id, "idempotent": True},
+                }
+
+            # Optional: refuse when an active lease still names this session.
+            active_lease = conn.execute(
+                """
+                SELECT lease_id, status, expires_at FROM leases
+                WHERE session_id = ? AND status = 'active'
+                LIMIT 1
+                """,
+                (session_id,),
+            ).fetchone()
+            if active_lease is not None:
+                return {
+                    "outcome": "blocked",
+                    "reason": "live_lease",
+                    "prior_status": prior_status,
+                    "new_status": prior_status,
+                    "details": {
+                        "session_id": session_id,
+                        "lease_id": active_lease["lease_id"],
+                        "blocker": "active_lease_row",
+                    },
+                }
+
+            cols = {
+                r[1] for r in conn.execute("PRAGMA table_info(sessions)").fetchall()
+            }
+            if "retired_at" in cols and "retire_reason" in cols:
+                conn.execute(
+                    """
+                    UPDATE sessions
+                    SET status = ?, retired_at = ?, retire_reason = ?
+                    WHERE session_id = ? AND status = ?
+                    """,
+                    (term, moment, reason_s, session_id, prior.get("status")),
+                )
+            else:
+                conn.execute(
+                    """
+                    UPDATE sessions
+                    SET status = ?
+                    WHERE session_id = ? AND status = ?
+                    """,
+                    (term, session_id, prior.get("status")),
+                )
+            changed = conn.execute(
+                "SELECT changes()"
+            ).fetchone()[0]
+            if not changed:
+                # Lost CAS race — re-read.
+                refreshed = conn.execute(
+                    "SELECT status FROM sessions WHERE session_id = ?",
+                    (session_id,),
+                ).fetchone()
+                cur = (
+                    str(refreshed["status"]).strip().lower()
+                    if refreshed is not None
+                    else None
+                )
+                return {
+                    "outcome": "already_terminal"
+                    if cur in terminal_set
+                    else "blocked",
+                    "reason": reason_s,
+                    "prior_status": prior_status,
+                    "new_status": cur,
+                    "details": {
+                        "session_id": session_id,
+                        "cas_lost": True,
+                    },
+                }
+
+            detail_payload: dict[str, Any] = {
+                "session_id": session_id,
+                "prior_status": prior_status,
+                "new_status": term,
+                "reason": reason_s,
+                "actor_session_id": actor_session_id,
+                "pid": prior.get("pid"),
+                "role": prior.get("role"),
+                "profile": prior.get("profile"),
+            }
+            if isinstance(details, Mapping):
+                for key, value in details.items():
+                    if key not in detail_payload:
+                        detail_payload[key] = value
+            message = (
+                f"session {session_id} retired reason={reason_s} "
+                f"prior_status={prior_status}"
+            )
+            # events.work_item_id is nullable; session retirement is not work-scoped.
+            conn.execute(
+                """
+                INSERT INTO events(work_item_id, event_type, message, created_at)
+                VALUES (NULL, 'session_retired', ?, ?)
+                """,
+                (message[:2000], moment),
+            )
+            # Also persist a structured JSON line in the message when short enough
+            # by appending a compact summary (full detail stays in return value /
+            # audit log; events.message is human-readable).
+            return {
+                "outcome": "retired",
+                "reason": reason_s,
+                "prior_status": prior_status,
+                "new_status": term,
+                "details": detail_payload,
+            }
+
     def list_sessions(
         self,
         *,
@@ -1639,6 +1835,29 @@ class ControlPlaneDB:
             if name not in cols:
                 conn.execute(f"ALTER TABLE sessions ADD COLUMN {name} {decl}")
 
+    _SESSION_LIFECYCLE_COLUMNS: tuple[tuple[str, str], ...] = (
+        ("owner_process_started_at", "TEXT"),
+        ("retired_at", "TEXT"),
+        ("retire_reason", "TEXT"),
+    )
+
+    def _migrate_session_lifecycle_columns(self, conn: sqlite3.Connection) -> None:
+        """Add session retirement / PID-reuse provenance columns (#969).
+
+        Additive and idempotent. Pre-existing rows migrate with NULL; retirement
+        fills ``retired_at`` / ``retire_reason``, and new upserts may record
+        ``owner_process_started_at`` for stronger identity checks.
+        """
+        cols = {
+            row[1]
+            for row in conn.execute("PRAGMA table_info(sessions)").fetchall()
+        }
+        if not cols:
+            return
+        for name, decl in self._SESSION_LIFECYCLE_COLUMNS:
+            if name not in cols:
+                conn.execute(f"ALTER TABLE sessions ADD COLUMN {name} {decl}")
+
     def list_active_claims(
         self,
         *,

docs/post-restart-reconcile.md

@@ -19,7 +19,11 @@ The assessor classifies:
 
 - **service_health** — process healthy / parity mutation-safe
 - **clients** — connected client descriptors (optional inventory)
-- **sessions** — active session rows with dead owner pids are unresolved
+- **sessions** — active session rows with dead or reused owner pids are
+  unresolved until retired via `#969` (`session_lifecycle` /
+  `apply_session_cleanup=true` on `gitea_reconcile_after_restart`, or
+  `gitea_retire_stale_workflow_sessions`). Live owners, live leases, and live
+  client-managed sessions are never retired.
 - **checkpoints** — soft-depends on #660; skipped with reason when schema absent
 - **leases** — live control-plane leases after restart
 - **capabilities** — master-parity / stale-runtime (#610)

gitea_mcp_server.py

@@ -24432,10 +24432,18 @@ def _run_post_restart_reconcile(
     repo: str | None = None,
     mode: str | None = None,
     limit: int = 200,
+    apply_session_cleanup: bool = False,
 ) -> dict:
-    """Gather + classify post-restart state; cache the latest proof (#662)."""
+    """Gather + classify post-restart state; cache the latest proof (#662 / #969).
+
+    When *apply_session_cleanup* is true, confirmed-stale session rows (dead
+    owner / PID reuse, no live lease) are terminalized through
+    ``session_lifecycle`` before re-classification so the sessions dimension
+    can resolve. Default remains dry (read-only) to preserve #662 rollout.
+    """
     global _POST_RESTART_LAST_PROOF, _POST_RESTART_BOOT_RAN
     import post_restart_reconcile as prr
+    import session_lifecycle as sl
 
     try:
         _h, o, r = _resolve(remote, None, org, repo)
@@ -24449,13 +24457,48 @@ def _run_post_restart_reconcile(
     inventory = _gather_post_restart_inventory(
         remote=remote, org=o, repo=r, limit=limit
     )
+
+    session_cleanup: dict | None = None
+    if apply_session_cleanup:
+        db, db_errs = _control_plane_db_or_error()
+        if db is None:
+            session_cleanup = {
+                "success": False,
+                "reasons": db_errs
+                or ["control-plane DB unavailable; cannot retire sessions"],
+            }
+        else:
+            profile = get_profile()
+            profile_name = (profile.get("profile_name") or "").strip() or "session"
+            actor = f"{profile_name}-{os.getpid()}"
+            session_cleanup = sl.retire_stale_sessions(
+                db,
+                sessions=inventory.get("sessions") or [],
+                leases=inventory.get("leases") or [],
+                dry_run=False,
+                actor_session_id=actor,
+                session_limit=max(1, int(limit)),
+            )
+            # Re-gather active sessions after mutation so classification sees
+            # the post-retirement fleet (retired rows drop out of active list).
+            try:
+                inventory["sessions"] = db.list_sessions(
+                    statuses=("active",), limit=max(1, int(limit))
+                )
+            except Exception as exc:  # noqa: BLE001
+                inventory["incomplete_reasons"] = list(
+                    inventory.get("incomplete_reasons") or []
+                ) + [f"post-retirement session re-list failed: {_redact(str(exc))}"]
+            fleet = (session_cleanup.get("fleet") or {}) if session_cleanup else {}
+            inventory["session_fleet"] = fleet
+
     proof = prr.reconcile_after_restart(
         inventory,
         mode=mode or _post_restart_reconcile_mode(),
     )
     payload = proof.as_dict()
     payload["success"] = True
-    payload["read_only"] = True
+    payload["read_only"] = not apply_session_cleanup
     payload["remote"] = remote
     payload["org"] = o
     payload["repo"] = r
@@ -24464,6 +24507,9 @@ def _run_post_restart_reconcile(
         "proposed_follow_ups lists durable issues the apply path may create; "
         "this tool never creates them (log-only by default, #662 rollout)"
     )
+    if session_cleanup is not None:
+        payload["session_cleanup"] = session_cleanup
+    payload["apply_session_cleanup"] = bool(apply_session_cleanup)
     _POST_RESTART_LAST_PROOF = payload
     _POST_RESTART_BOOT_RAN = True
     return payload
@@ -24490,8 +24536,9 @@ def gitea_reconcile_after_restart(
     repo: str | None = None,
     mode: str | None = None,
     limit: int = 200,
+    apply_session_cleanup: bool = False,
 ) -> dict:
-    """Run post-restart MCP reconciliation and return a completion proof (#662).
+    """Run post-restart MCP reconciliation and return a completion proof (#662 / #969).
 
     Gathers live control-plane sessions, leases, worktree bindings, and
     master-parity evidence, then classifies them with the pure
@@ -24499,12 +24546,17 @@ def gitea_reconcile_after_restart(
     machine-readable completion proof listing resolved / unresolved dimensions
     and proposed durable follow-up issues.
 
-    Read-only by design: never restarts MCP, never auto-resumes write
+    Never restarts MCP, never auto-resumes write mutations, and never creates
-    mutations, and never creates Gitea issues (those are a separate apply
+    Gitea issues. Default mode is ``log_only``; set
-    path). Default mode is ``log_only``; set
     ``GITEA_POST_RESTART_RECONCILE_MODE=enforce`` (or pass ``mode='enforce'``)
     to set ``mutation_hold`` when anything remains unresolved.
 
+    *apply_session_cleanup* (#969): when true, confirmed-stale workflow session
+    rows (dead owner PID / PID reuse, no live lease, not a live client-managed
+    owner) are terminalized through the sanctioned ``session_lifecycle`` path
+    before re-classification. Default false preserves the historical read-only
+    gather+classify behaviour. Does not delete historical rows.
+
     Soft-depends on #660 for session checkpoints: when the checkpoint schema
     module is absent the checkpoints dimension is ``skipped`` with an explicit
     reason rather than inventing a schema.
@@ -24524,9 +24576,96 @@ def gitea_reconcile_after_restart(
         repo=repo,
         mode=mode,
         limit=limit,
+        apply_session_cleanup=bool(apply_session_cleanup),
     )
 
 
+@mcp.tool()
+def gitea_retire_stale_workflow_sessions(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    apply: bool = False,
+    limit: int = 500,
+) -> dict:
+    """Retire workflow session rows whose owners are no longer live (#969).
+
+    Plans (and optionally applies) terminalization of control-plane session
+    rows that are confirmed stale:
+
+    * owner PID absent / dead
+    * PID reused by an unrelated process (process start after session start)
+    * no live workflow lease
+    * not a live client-managed owner
+
+    Default ``apply=false`` is dry-run only. ``apply=true`` performs CAS
+    status updates to ``retired`` and writes durable ``session_retired``
+    events. Idempotent under concurrent reconciliation. Never deletes rows
+    and never touches sessions protected by a live lease or live owner.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "read_only": not apply,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    try:
+        _h, o, r = _resolve(remote, None, org, repo)
+    except ValueError as exc:
+        return {"success": False, "reasons": [str(exc)], "read_only": not apply}
+
+    db, db_errs = _control_plane_db_or_error()
+    if db is None:
+        return {
+            "success": False,
+            "read_only": not apply,
+            "reasons": db_errs or ["control-plane DB unavailable"],
+        }
+
+    import session_lifecycle as sl
+
+    profile = get_profile()
+    profile_name = (profile.get("profile_name") or "").strip() or "session"
+    actor = f"{profile_name}-{os.getpid()}"
+
+    # Prefer lease inventory scoped to the requested repo when available.
+    leases: list[dict] = []
+    try:
+        lease_result = lease_lifecycle.list_active_leases(
+            db,
+            remote=remote if remote in REMOTES else remote,
+            org=o,
+            repo=r,
+            role=None,
+            include_non_active=True,
+            limit=max(1, int(limit)),
+        )
+        leases = list(lease_result.get("leases") or [])
+    except Exception:  # noqa: BLE001
+        try:
+            leases = db.list_leases(statuses=("active",), limit=max(1, int(limit)))
+        except Exception:  # noqa: BLE001
+            leases = []
+
+    result = sl.retire_stale_sessions(
+        db,
+        leases=leases,
+        dry_run=not bool(apply),
+        actor_session_id=actor,
+        session_limit=max(1, int(limit)),
+    )
+    result["read_only"] = not bool(apply)
+    result["apply"] = bool(apply)
+    result["remote"] = remote
+    result["org"] = o
+    result["repo"] = r
+    return result
+
+
 @mcp.tool()
 def gitea_inspect_workflow_lease(
     lease_id: str,

post_restart_reconcile.py

@@ -475,8 +475,37 @@ def reconcile_after_restart(
             )
         )
 
-    # --- sessions -------------------------------------------------------
+    # --- sessions (#969: dead-owner / PID-reuse lifecycle) ---------------
+    # Prefer a precomputed fleet report from the gather/apply path when present;
+    # otherwise classify pure from inventory (injectable checkers stay default).
     sessions = [s for s in (inventory.get("sessions") or []) if isinstance(s, Mapping)]
+    leases_for_sessions = [
+        L for L in (inventory.get("leases") or []) if isinstance(L, Mapping)
+    ]
+    fleet_report = inventory.get("session_fleet")
+    if isinstance(fleet_report, Mapping) and "retireable_session_ids" in fleet_report:
+        fleet_details = dict(fleet_report)
+        retireable_ids = list(fleet_details.get("retireable_session_ids") or [])
+        resolved = bool(fleet_details.get("sessions_dimension_resolved", not retireable_ids))
+    else:
+        try:
+            import session_lifecycle as _sl
+
+            client_managed = inventory.get("client_managed_session_ids")
+            cm_set = None
+            if isinstance(client_managed, (list, tuple, set, frozenset)):
+                cm_set = {str(x) for x in client_managed}
+            fleet = _sl.classify_sessions(
+                sessions,
+                leases=leases_for_sessions,
+                now=started,
+                client_managed_sessions=cm_set,
+            )
+            fleet_details = fleet.as_dict()
+            retireable_ids = list(fleet_details.get("retireable_session_ids") or [])
+            resolved = bool(fleet_details.get("sessions_dimension_resolved"))
+        except Exception as exc:  # noqa: BLE001 — fail closed to legacy signal
+            # Legacy fallback: dead-pid active rows only (pre-#969 behaviour).
             orphan_sessions = [
                 s
                 for s in sessions
@@ -484,15 +513,27 @@ def reconcile_after_restart(
                 and s.get("pid") is not None
                 and not lease_lifecycle.is_process_alive(s.get("pid"))
             ]
-    if orphan_sessions:
+            retireable_ids = [s.get("session_id") for s in orphan_sessions]
+            resolved = not orphan_sessions
+            fleet_details = {
+                "total_sessions": len(sessions),
+                "retireable_session_ids": retireable_ids,
+                "legacy_fallback": True,
+                "fallback_error": str(exc),
+            }
+
+    if not resolved and retireable_ids:
         items.append(
             _item(
                 DIM_SESSIONS,
                 ITEM_UNRESOLVED,
-                f"{len(orphan_sessions)} active session row(s) with dead owner pid",
+                f"{len(retireable_ids)} session row(s) with dead/reused owner "
+                f"await retirement",
                 details={
-                    "orphan_session_ids": [s.get("session_id") for s in orphan_sessions],
+                    "orphan_session_ids": retireable_ids,
+                    "retireable_session_ids": retireable_ids,
                     "total_sessions": len(sessions),
+                    "fleet": fleet_details,
                 },
                 follow_up=True,
             )
@@ -502,8 +543,12 @@ def reconcile_after_restart(
             _item(
                 DIM_SESSIONS,
                 ITEM_RESOLVED,
-                f"{len(sessions)} session row(s) reconciled (no dead-pid orphans)",
+                f"{len(sessions)} session row(s) reconciled "
-                details={"total_sessions": len(sessions)},
+                f"(no retireable dead/reused owners)",
+                details={
+                    "total_sessions": len(sessions),
+                    "fleet": fleet_details,
+                },
             )
         )
 

session_lifecycle.py

@@ -0,0 +1,703 @@
+"""Safe lifecycle for workflow session rows with dead or reused owners (#969).
+
+Post-restart reconciliation previously left hundreds of ``active`` session rows
+with dead owner PIDs permanently unresolved. A PID existence check alone is not
+enough: operating systems reuse PIDs, so an unrelated live process can appear to
+own a historical session.
+
+This module is the pure classification + apply core for session retirement:
+
+* distinguish live, disconnected, stale, protected, and terminal records
+* refuse retirement when a live lease or live verified owner remains
+* protect live client-managed sessions
+* detect PID reuse via process start time vs session start / heartbeat
+* terminalize confirmed-stale rows idempotently with durable audit events
+* stay safe under concurrent reconciles (CAS on status)
+
+Design mirrors ``lease_lifecycle`` / ``post_restart_reconcile``:
+
+* Pure classification accepts injectable checkers so unit tests never touch
+  real processes.
+* Apply mutations go only through :meth:`ControlPlaneDB.retire_session`.
+* Historical rows are never deleted; status moves to a terminal value and an
+  events-row records the action + reason.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+from dataclasses import dataclass, field
+from datetime import datetime, timedelta, timezone
+from typing import Any, Callable, Mapping, Sequence
+
+import control_plane_db as cpd
+import lease_lifecycle
+
+# Session status vocabulary.
+SESSION_STATUS_ACTIVE = "active"
+SESSION_STATUS_RETIRED = "retired"
+SESSION_STATUS_ENDED = "ended"
+SESSION_STATUS_TERMINAL = "terminal"
+
+TERMINAL_SESSION_STATUSES = frozenset(
+    {
+        SESSION_STATUS_RETIRED,
+        SESSION_STATUS_ENDED,
+        SESSION_STATUS_TERMINAL,
+        "dead",
+        "stale",
+        "orphaned",
+    }
+)
+
+# Classification outcomes for one session row.
+CLASS_LIVE = "live"
+CLASS_DISCONNECTED = "disconnected"
+CLASS_STALE = "stale"
+CLASS_PROTECTED = "protected"
+CLASS_TERMINAL = "terminal"
+
+# Stable reason codes (audit + tests).
+REASON_ALREADY_TERMINAL = "already_terminal"
+REASON_LIVE_OWNER = "live_owner"
+REASON_LIVE_LEASE = "live_lease"
+REASON_CLIENT_MANAGED_LIVE = "client_managed_live"
+REASON_DEAD_OWNER = "dead_owner"
+REASON_PID_REUSE = "pid_reuse"
+REASON_HEARTBEAT_STALE_DEAD = "heartbeat_stale_dead_owner"
+REASON_MISSING_PID = "missing_pid_no_lease"
+
+# Event type written to control-plane events table.
+EVENT_SESSION_RETIRED = "session_retired"
+
+# Default heartbeat window before a still-alive PID is treated as disconnected
+# rather than live (does not alone authorize retirement).
+DEFAULT_HEARTBEAT_STALE_SECONDS = 900
+
+# PID reuse: process start must be strictly later than session started_at by
+# more than this skew (ps lstart is second-resolution; clocks can lag).
+PID_REUSE_SKEW = timedelta(seconds=2)
+
+# Live lease freshness values that block retirement.
+_LIVE_LEASE_FRESHNESS = frozenset({"active", "live"})
+
+
+class SessionLifecycleError(RuntimeError):
+    """Fail-closed session lifecycle policy error."""
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _parse_ts(value: str | datetime | None) -> datetime | None:
+    if value is None:
+        return None
+    if isinstance(value, datetime):
+        if value.tzinfo is None:
+            return value.replace(tzinfo=timezone.utc)
+        return value.astimezone(timezone.utc)
+    return cpd._parse_ts(str(value))
+
+
+def _ts(dt: datetime | None = None) -> str:
+    return cpd._ts(dt)
+
+
+def process_start_time(pid: int | None) -> datetime | None:
+    """Return the OS start time of *pid*, or None when it cannot be resolved.
+
+    Uses ``ps -o lstart=`` (POSIX). Failures return None rather than inventing
+    evidence — missing start time never authorizes retirement of a live PID.
+    """
+    if pid is None:
+        return None
+    try:
+        pid_i = int(pid)
+    except (TypeError, ValueError):
+        return None
+    if pid_i <= 0:
+        return None
+    try:
+        proc = subprocess.run(
+            ["ps", "-o", "lstart=", "-p", str(pid_i)],
+            capture_output=True,
+            text=True,
+            check=False,
+            timeout=2,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return None
+    if proc.returncode != 0:
+        return None
+    text = (proc.stdout or "").strip()
+    if not text:
+        return None
+    try:
+        # Example: "Wed Jul 29 09:14:36 2026"
+        naive = datetime.strptime(text, "%a %b %d %H:%M:%S %Y")
+        return naive.replace(tzinfo=timezone.utc)
+    except ValueError:
+        return None
+
+
+def _lease_freshness_label(lease: Mapping[str, Any]) -> str:
+    fr = lease.get("freshness")
+    if isinstance(fr, Mapping):
+        return str(fr.get("freshness") or "").strip().lower()
+    if fr:
+        return str(fr).strip().lower()
+    # Fall back to classifying a raw lease row.
+    try:
+        return str(
+            lease_lifecycle.classify_lease_freshness(lease).get("freshness") or ""
+        ).strip().lower()
+    except Exception:  # noqa: BLE001 — pure classifier must not raise on bad rows
+        status = str(lease.get("status") or "").strip().lower()
+        return status or "unknown"
+
+
+def live_lease_session_ids(
+    leases: Sequence[Mapping[str, Any]] | None,
+    *,
+    now: datetime | None = None,
+    pid_checker: Callable[[int | None], bool] = lease_lifecycle.is_process_alive,
+) -> set[str]:
+    """Session ids that still hold a live (or ambiguous-active) workflow lease."""
+    live: set[str] = set()
+    moment = now or _utc_now()
+    for lease in leases or ():
+        if not isinstance(lease, Mapping):
+            continue
+        status = str(lease.get("status") or "").strip().lower()
+        if status and status not in {
+            lease_lifecycle.LEASE_STATUS_ACTIVE,
+            "",
+        }:
+            # Explicit terminal lease statuses never protect a session.
+            if status in {
+                lease_lifecycle.LEASE_STATUS_RELEASED,
+                lease_lifecycle.LEASE_STATUS_EXPIRED,
+                lease_lifecycle.LEASE_STATUS_ABANDONED,
+            }:
+                continue
+        freshness = _lease_freshness_label(lease)
+        if freshness in _LIVE_LEASE_FRESHNESS or freshness in {"", "unknown"}:
+            # Ambiguous active rows: re-check with authoritative classifier.
+            try:
+                fr = lease_lifecycle.classify_lease_freshness(
+                    lease, now=moment, pid_checker=pid_checker
+                )
+                freshness = str(fr.get("freshness") or "").strip().lower()
+            except Exception:  # noqa: BLE001
+                freshness = "unknown"
+        if freshness in _LIVE_LEASE_FRESHNESS:
+            sid = str(lease.get("session_id") or "").strip()
+            if sid:
+                live.add(sid)
+        elif freshness == "unknown" and status in {
+            lease_lifecycle.LEASE_STATUS_ACTIVE,
+            "",
+        }:
+            # Fail closed: active lease with unknown freshness blocks retirement.
+            sid = str(lease.get("session_id") or "").strip()
+            if sid:
+                live.add(sid)
+    return live
+
+
+@dataclass(frozen=True)
+class SessionClassification:
+    """Classification of one workflow session row."""
+
+    session_id: str
+    classification: str
+    reason: str
+    retireable: bool
+    status: str | None
+    pid: int | None
+    pid_alive: bool | None
+    pid_reused: bool
+    heartbeat_stale: bool
+    has_live_lease: bool
+    client_managed: bool
+    details: dict[str, Any] = field(default_factory=dict)
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "session_id": self.session_id,
+            "classification": self.classification,
+            "reason": self.reason,
+            "retireable": self.retireable,
+            "status": self.status,
+            "pid": self.pid,
+            "pid_alive": self.pid_alive,
+            "pid_reused": self.pid_reused,
+            "heartbeat_stale": self.heartbeat_stale,
+            "has_live_lease": self.has_live_lease,
+            "client_managed": self.client_managed,
+            "details": dict(self.details),
+        }
+
+
+def classify_session(
+    row: Mapping[str, Any],
+    *,
+    now: datetime | None = None,
+    pid_checker: Callable[[int | None], bool] = lease_lifecycle.is_process_alive,
+    process_start_probe: Callable[[int | None], datetime | None] = process_start_time,
+    live_lease_sessions: set[str] | frozenset[str] | None = None,
+    client_managed_sessions: set[str] | frozenset[str] | None = None,
+    heartbeat_stale_seconds: int = DEFAULT_HEARTBEAT_STALE_SECONDS,
+) -> SessionClassification:
+    """Classify one session row for retirement decisions (#969).
+
+    Rules (first match wins where noted):
+
+    1. Non-active / already-terminal status → ``terminal`` (not retireable).
+    2. Session holds a live lease → ``protected`` (never retire).
+    3. PID missing and no live lease → ``stale`` (retireable: missing owner).
+    4. PID alive + process start after session start → ``stale`` (PID reuse).
+    5. PID alive + client-managed → ``live`` protected (never retire).
+    6. PID alive + fresh heartbeat → ``live``.
+    7. PID alive + stale heartbeat → ``disconnected`` (not retireable alone).
+    8. PID dead → ``stale`` (retireable).
+    """
+    moment = now or _utc_now()
+    session_id = str(row.get("session_id") or "").strip()
+    status = str(row.get("status") or "").strip().lower() or None
+    raw_pid = row.get("pid")
+    try:
+        pid = int(raw_pid) if raw_pid is not None else None
+    except (TypeError, ValueError):
+        pid = None
+
+    client_managed = bool(
+        row.get("client_managed")
+        or row.get("is_client_managed")
+        or (
+            client_managed_sessions is not None
+            and session_id in client_managed_sessions
+        )
+    )
+    has_live_lease = bool(
+        live_lease_sessions is not None and session_id in live_lease_sessions
+    )
+
+    hb = _parse_ts(row.get("last_heartbeat_at"))
+    heartbeat_stale = bool(
+        hb is not None
+        and (moment - hb).total_seconds() > max(0, int(heartbeat_stale_seconds))
+    )
+    if hb is None and status == SESSION_STATUS_ACTIVE:
+        # No heartbeat evidence: treat as stale for liveness bookkeeping only.
+        heartbeat_stale = True
+
+    started = _parse_ts(row.get("started_at"))
+    recorded_proc_start = _parse_ts(row.get("owner_process_started_at"))
+
+    if status in TERMINAL_SESSION_STATUSES:
+        return SessionClassification(
+            session_id=session_id,
+            classification=CLASS_TERMINAL,
+            reason=REASON_ALREADY_TERMINAL,
+            retireable=False,
+            status=status,
+            pid=pid,
+            pid_alive=None,
+            pid_reused=False,
+            heartbeat_stale=heartbeat_stale,
+            has_live_lease=has_live_lease,
+            client_managed=client_managed,
+        )
+
+    if has_live_lease:
+        return SessionClassification(
+            session_id=session_id,
+            classification=CLASS_PROTECTED,
+            reason=REASON_LIVE_LEASE,
+            retireable=False,
+            status=status,
+            pid=pid,
+            pid_alive=pid_checker(pid) if pid is not None else None,
+            pid_reused=False,
+            heartbeat_stale=heartbeat_stale,
+            has_live_lease=True,
+            client_managed=client_managed,
+            details={"blocker": "live_workflow_lease"},
+        )
+
+    if pid is None:
+        return SessionClassification(
+            session_id=session_id,
+            classification=CLASS_STALE,
+            reason=REASON_MISSING_PID,
+            retireable=True,
+            status=status,
+            pid=None,
+            pid_alive=False,
+            pid_reused=False,
+            heartbeat_stale=heartbeat_stale,
+            has_live_lease=False,
+            client_managed=client_managed,
+        )
+
+    pid_alive = bool(pid_checker(pid))
+    pid_reused = False
+    proc_start: datetime | None = None
+
+    if pid_alive:
+        proc_start = recorded_proc_start or process_start_probe(pid)
+        # PID reuse: live process started after the session row itself was
+        # created. Anchor on started_at only — last_heartbeat alone is not a
+        # safe bound (synthetic inventories and long-lived processes would
+        # false-positive against a live ``ps`` probe).
+        if proc_start is not None and started is not None:
+            if proc_start > (started + PID_REUSE_SKEW):
+                pid_reused = True
+
+    if pid_reused:
+        return SessionClassification(
+            session_id=session_id,
+            classification=CLASS_STALE,
+            reason=REASON_PID_REUSE,
+            retireable=True,
+            status=status,
+            pid=pid,
+            pid_alive=True,
+            pid_reused=True,
+            heartbeat_stale=heartbeat_stale,
+            has_live_lease=False,
+            client_managed=client_managed,
+            details={
+                "process_started_at": _ts(proc_start) if proc_start else None,
+                "session_started_at": row.get("started_at"),
+                "last_heartbeat_at": row.get("last_heartbeat_at"),
+            },
+        )
+
+    if pid_alive and client_managed:
+        return SessionClassification(
+            session_id=session_id,
+            classification=CLASS_LIVE,
+            reason=REASON_CLIENT_MANAGED_LIVE,
+            retireable=False,
+            status=status,
+            pid=pid,
+            pid_alive=True,
+            pid_reused=False,
+            heartbeat_stale=heartbeat_stale,
+            has_live_lease=False,
+            client_managed=True,
+        )
+
+    if pid_alive and not heartbeat_stale:
+        return SessionClassification(
+            session_id=session_id,
+            classification=CLASS_LIVE,
+            reason=REASON_LIVE_OWNER,
+            retireable=False,
+            status=status,
+            pid=pid,
+            pid_alive=True,
+            pid_reused=False,
+            heartbeat_stale=False,
+            has_live_lease=False,
+            client_managed=client_managed,
+        )
+
+    if pid_alive and heartbeat_stale:
+        # Process still exists but has not heartbeated — disconnected, not
+        # confirmed stale. Do not retire; operator/reconnect owns next step.
+        return SessionClassification(
+            session_id=session_id,
+            classification=CLASS_DISCONNECTED,
+            reason=REASON_LIVE_OWNER,
+            retireable=False,
+            status=status,
+            pid=pid,
+            pid_alive=True,
+            pid_reused=False,
+            heartbeat_stale=True,
+            has_live_lease=False,
+            client_managed=client_managed,
+            details={"note": "alive_pid_stale_heartbeat_not_retired"},
+        )
+
+    # PID dead (or checker said not alive).
+    reason = REASON_DEAD_OWNER
+    if heartbeat_stale:
+        reason = REASON_HEARTBEAT_STALE_DEAD
+    return SessionClassification(
+        session_id=session_id,
+        classification=CLASS_STALE,
+        reason=reason,
+        retireable=True,
+        status=status,
+        pid=pid,
+        pid_alive=False,
+        pid_reused=False,
+        heartbeat_stale=heartbeat_stale,
+        has_live_lease=False,
+        client_managed=client_managed,
+    )
+
+
+@dataclass(frozen=True)
+class SessionFleetReport:
+    """Fleet-wide classification summary for reconcile + apply."""
+
+    classifications: tuple[SessionClassification, ...]
+    live_count: int
+    disconnected_count: int
+    stale_count: int
+    protected_count: int
+    terminal_count: int
+    retireable: tuple[SessionClassification, ...]
+    counts_by_reason: dict[str, int]
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "live_count": self.live_count,
+            "disconnected_count": self.disconnected_count,
+            "stale_count": self.stale_count,
+            "protected_count": self.protected_count,
+            "terminal_count": self.terminal_count,
+            "retireable_count": len(self.retireable),
+            "retireable_session_ids": [c.session_id for c in self.retireable],
+            "counts_by_reason": dict(self.counts_by_reason),
+            "classifications": [c.as_dict() for c in self.classifications],
+            "sessions_dimension_resolved": len(self.retireable) == 0,
+        }
+
+
+def classify_sessions(
+    sessions: Sequence[Mapping[str, Any]] | None,
+    *,
+    leases: Sequence[Mapping[str, Any]] | None = None,
+    now: datetime | None = None,
+    pid_checker: Callable[[int | None], bool] = lease_lifecycle.is_process_alive,
+    process_start_probe: Callable[[int | None], datetime | None] = process_start_time,
+    client_managed_sessions: set[str] | frozenset[str] | None = None,
+    heartbeat_stale_seconds: int = DEFAULT_HEARTBEAT_STALE_SECONDS,
+) -> SessionFleetReport:
+    """Classify a fleet of session rows against live leases (#969)."""
+    moment = now or _utc_now()
+    live_leases = live_lease_session_ids(
+        leases, now=moment, pid_checker=pid_checker
+    )
+    results: list[SessionClassification] = []
+    for row in sessions or ():
+        if not isinstance(row, Mapping):
+            continue
+        results.append(
+            classify_session(
+                row,
+                now=moment,
+                pid_checker=pid_checker,
+                process_start_probe=process_start_probe,
+                live_lease_sessions=live_leases,
+                client_managed_sessions=client_managed_sessions,
+                heartbeat_stale_seconds=heartbeat_stale_seconds,
+            )
+        )
+
+    live_count = sum(1 for c in results if c.classification == CLASS_LIVE)
+    disconnected_count = sum(
+        1 for c in results if c.classification == CLASS_DISCONNECTED
+    )
+    stale_count = sum(1 for c in results if c.classification == CLASS_STALE)
+    protected_count = sum(1 for c in results if c.classification == CLASS_PROTECTED)
+    terminal_count = sum(1 for c in results if c.classification == CLASS_TERMINAL)
+    retireable = tuple(c for c in results if c.retireable)
+    by_reason: dict[str, int] = {}
+    for c in results:
+        by_reason[c.reason] = by_reason.get(c.reason, 0) + 1
+
+    return SessionFleetReport(
+        classifications=tuple(results),
+        live_count=live_count,
+        disconnected_count=disconnected_count,
+        stale_count=stale_count,
+        protected_count=protected_count,
+        terminal_count=terminal_count,
+        retireable=retireable,
+        counts_by_reason=by_reason,
+    )
+
+
+@dataclass(frozen=True)
+class RetirementResult:
+    """Outcome of one session retirement attempt."""
+
+    session_id: str
+    outcome: str  # retired | already_terminal | skipped | blocked | missing
+    reason: str
+    prior_status: str | None = None
+    new_status: str | None = None
+    details: dict[str, Any] = field(default_factory=dict)
+
+    def as_dict(self) -> dict[str, Any]:
+        return {
+            "session_id": self.session_id,
+            "outcome": self.outcome,
+            "reason": self.reason,
+            "prior_status": self.prior_status,
+            "new_status": self.new_status,
+            "details": dict(self.details),
+        }
+
+
+def apply_session_retirements(
+    db: cpd.ControlPlaneDB,
+    report: SessionFleetReport,
+    *,
+    dry_run: bool = False,
+    actor_session_id: str | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Terminalize every retireable session in *report* (idempotent).
+
+    Concurrent reconciles are safe: each retirement is a CAS on
+    ``status='active'`` (or other non-terminal). A second pass that sees the
+    same session already retired records ``already_terminal`` rather than
+    duplicating audit noise beyond a single no-op outcome.
+    """
+    moment = now or _utc_now()
+    results: list[RetirementResult] = []
+    retired = 0
+    already = 0
+    blocked = 0
+    missing = 0
+
+    for classification in report.retireable:
+        if not classification.retireable:
+            continue
+        if dry_run:
+            results.append(
+                RetirementResult(
+                    session_id=classification.session_id,
+                    outcome="skipped",
+                    reason=classification.reason,
+                    prior_status=classification.status,
+                    new_status=SESSION_STATUS_RETIRED,
+                    details={"dry_run": True, **classification.details},
+                )
+            )
+            continue
+
+        applied = db.retire_session(
+            session_id=classification.session_id,
+            reason=classification.reason,
+            actor_session_id=actor_session_id,
+            details={
+                "classification": classification.classification,
+                "pid": classification.pid,
+                "pid_alive": classification.pid_alive,
+                "pid_reused": classification.pid_reused,
+                "client_managed": classification.client_managed,
+                **classification.details,
+            },
+            now=moment,
+        )
+        outcome = str(applied.get("outcome") or "missing")
+        results.append(
+            RetirementResult(
+                session_id=classification.session_id,
+                outcome=outcome,
+                reason=str(applied.get("reason") or classification.reason),
+                prior_status=applied.get("prior_status"),
+                new_status=applied.get("new_status"),
+                details=dict(applied.get("details") or {}),
+            )
+        )
+        if outcome == "retired":
+            retired += 1
+        elif outcome == "already_terminal":
+            already += 1
+        elif outcome == "blocked":
+            blocked += 1
+        else:
+            missing += 1
+
+    # Non-retireable classifications are recorded for audit completeness when
+    # dry-run lists the fleet, but apply only mutates retireable rows.
+    return {
+        "success": True,
+        "dry_run": dry_run,
+        "retired_count": retired,
+        "already_terminal_count": already,
+        "blocked_count": blocked,
+        "missing_count": missing,
+        "planned_count": len(report.retireable),
+        "results": [r.as_dict() for r in results],
+        "audit_action": EVENT_SESSION_RETIRED,
+        "actor_session_id": actor_session_id,
+        "recorded_at": _ts(moment),
+    }
+
+
+def retire_stale_sessions(
+    db: cpd.ControlPlaneDB,
+    *,
+    sessions: Sequence[Mapping[str, Any]] | None = None,
+    leases: Sequence[Mapping[str, Any]] | None = None,
+    dry_run: bool = False,
+    actor_session_id: str | None = None,
+    now: datetime | None = None,
+    pid_checker: Callable[[int | None], bool] = lease_lifecycle.is_process_alive,
+    process_start_probe: Callable[[int | None], datetime | None] = process_start_time,
+    client_managed_sessions: set[str] | frozenset[str] | None = None,
+    heartbeat_stale_seconds: int = DEFAULT_HEARTBEAT_STALE_SECONDS,
+    session_limit: int = 500,
+) -> dict[str, Any]:
+    """End-to-end plan + apply for stale session retirement (#969).
+
+    When *sessions* is omitted the control-plane DB is inventoried (active
+    rows only). Callers that already gathered inventory should pass it.
+    """
+    moment = now or _utc_now()
+    if sessions is None:
+        sessions = db.list_sessions(
+            statuses=(SESSION_STATUS_ACTIVE,),
+            limit=max(1, int(session_limit)),
+        )
+    if leases is None:
+        try:
+            leases = db.list_leases(statuses=("active",), limit=max(1, int(session_limit)))
+        except Exception:  # noqa: BLE001
+            leases = []
+
+    report = classify_sessions(
+        sessions,
+        leases=leases,
+        now=moment,
+        pid_checker=pid_checker,
+        process_start_probe=process_start_probe,
+        client_managed_sessions=client_managed_sessions,
+        heartbeat_stale_seconds=heartbeat_stale_seconds,
+    )
+    apply_result = apply_session_retirements(
+        db,
+        report,
+        dry_run=dry_run,
+        actor_session_id=actor_session_id,
+        now=moment,
+    )
+    return {
+        "success": True,
+        "fleet": report.as_dict(),
+        "apply": apply_result,
+        "sessions_dimension_resolved": (
+            report.as_dict()["sessions_dimension_resolved"]
+            if dry_run
+            else apply_result["planned_count"]
+            == (
+                apply_result["retired_count"]
+                + apply_result["already_terminal_count"]
+            )
+            and apply_result["blocked_count"] == 0
+        ),
+    }

task_capability_map.py

@@ -153,8 +153,9 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.push",
         "role": "author",
     },
-    # #662: post-restart reconcile is read-only inventory + pure classification.
+    # #662: post-restart reconcile is inventory + pure classification.
-    # Durable follow-up issue creation is a separate apply path (not this task).
+    # #969: optional apply_session_cleanup retires confirmed-stale session rows
+    # through the same tool; durable follow-up Gitea issues remain separate.
     "reconcile_after_restart": {
         "permission": "gitea.read",
         "role": "author",
@@ -163,6 +164,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    # #969: explicit plan/apply path for dead-owner session retirement.
+    "retire_stale_workflow_sessions": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
+    "gitea_retire_stale_workflow_sessions": {
+        "permission": "gitea.read",
+        "role": "author",
+    },
     # #644: Phase 2 Web Console recovery tasks.
     "clear_stale_binding": {
         "permission": "gitea.read",

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()