fix(reconcile): server-enforced, revalidated missing-worktree cleanup (#970 review 644 B1-B5)

Addresses the five blocking findings of review 644 on PR #972.

B1 — live ownership and status revalidation. resolve_missing_worktree_binding
now re-reads the authoritative lease, session, checkpoint, and issue-lock rows
immediately before mutating and diffs them against the snapshot the audit
recorded (binding path and identity, lease id/status/session/owner pid,
checkpoint path/status, live-session evidence, trustworthy ownership evidence,
issue-lock state). Any drift fails closed without mutation, and the binding is
reclassified from the live values rather than the audit snapshot. A candidate
carrying no audited snapshot is refused rather than trusted.

B2 — server-enforced cleanup authorization. Apply mode no longer accepts a
client-supplied operator_authorized boolean; it is rejected outright at the MCP
tool and in the module (#709 F1 / review 434). Authorization is now the
project's own reconciliation cleanup gate, required at both the task-capability
boundary (new reconciler-only reconcile_missing_worktree_bindings capability,
gitea.branch.delete, role-exclusive) and the production mutation boundary
(an authorized audit_reconciliation_mode cleanup phase, re-checked at the point
of mutation so a forged authorization mapping cannot stand in for the gate).
Dry-run remains available to any gitea.read profile and stays non-mutating.
Existing role, repository, parity, and provenance gates are unchanged.

B3 — expected-path compare-and-swap. retire_session_checkpoint_worktree_path
now requires expected_path and performs a guarded update keyed on the stored
path, refusing without mutation when the stored path was moved, replaced, or
concurrently changed, when the row is unknown, or when a selector matches more
than one checkpoint. retire_lease_worktree_path gains the same treatment plus
optional status/session/owner-pid compare-and-swap, and its UPDATE is keyed on
the audited path. Both report an idempotent already_retired outcome instead of
falsely reporting a retirement.

B4 — live-session and issue-lock evidence. session_active is now derived from
the control-plane sessions table instead of never being set, along two axes:
genuine liveness (recorded active, PID not dead, heartbeat fresh — the rule
reused from restart_coordinator) and weaker but still trustworthy recorded
ownership. A non-terminal lease now protects its binding regardless of whether
the recorded PID is alive, so dead-PID evidence alone can no longer retire a
lease the control plane still holds. The previously unused issue_lock_store is
now read: a live durable issue lock binding the path or branch blocks cleanup,
and locks whose own paths are missing are reported for release through their
own lifecycle rather than retired here.

B5 — adversarial regression coverage. The suite now drives the registered MCP
tools through mcp_server, the real ControlPlaneDB, and the real cleanup gate,
covering lease status/ownership/session/path drift, expected-path mismatch,
concurrent recreation, an unauthorized caller submitting operator_authorized,
wrong profile and missing capability, live-session and trustworthy-owner
evidence, conflicting issue locks, non-mutating dry-run, exact-binding-only
retirement, preservation of unrelated worktrees and git metadata, idempotent
re-execution, and worktrees-dimension resolution.

All original #970 acceptance criteria are preserved, including the distinctions
between deleted paths, moved paths, unavailable hosts or mounts, transient
filesystem failures, live ownership, and concurrent recreation.

Tests: focused #970 suite 47 passed. Adjacent suites (capability role
invariants, audit reconciliation mode, control plane DB, lease lifecycle,
reconciler cleanup integration, delete-branch capability, restart coordinator,
bootstrap lock contract) 252 passed / 93 subtests. Full suite 28 failed /
6000 passed, an exact match of the pre-change baseline's 28 failing test ids
at 3f584352 (28 failed / 5960 passed).

Closes #970

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, Mapping, Sequence
+from typing import Any, Iterator, Sequence
 
 import dependency_graph
 import gitea_audit
 
-SCHEMA_VERSION = 6
+SCHEMA_VERSION = 5
 
 # Assignable work kinds only — raw monitoring incidents are never work items.
 WORK_KINDS = frozenset({"issue", "pr"})
@@ -280,6 +280,22 @@ def _ts(dt: datetime | None = None) -> str:
     return value.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace("+00:00", "Z")
 
 
+def _realpath_or_raw(value: str | None) -> str:
+    """Normalize a filesystem path for compare-and-swap equality (#970).
+
+    Symlinks and ``..`` segments must not make two spellings of the same path
+    look different, but an unresolvable path must still compare as itself
+    rather than collapsing to empty — an empty result means "no path given".
+    """
+    text = (value or "").strip()
+    if not text:
+        return ""
+    try:
+        return os.path.realpath(os.path.abspath(text))
+    except Exception:
+        return text
+
+
 def _parse_ts(value: str | None) -> datetime | None:
     if not value:
         return None
@@ -419,7 +435,6 @@ 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 (?, ?)",
@@ -813,7 +828,6 @@ 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.
 
@@ -823,22 +837,16 @@ 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 and proc_start is None:
+                if instance is None:
                     conn.execute(
                         """
                         UPDATE sessions
@@ -848,21 +856,7 @@ class ControlPlaneDB:
                         """,
                         (role, profile, namespace, pid, now, status, session_id),
                     )
-                elif instance is None:
+                else:
-                    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
@@ -876,33 +870,18 @@ 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, owner_process_started_at
+                        controller_instance_id
-                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
                     """,
                     (
                         session_id, role, profile, namespace, pid, now, now,
-                        status, instance, proc_start,
+                        status, instance,
                     ),
                 )
             row = conn.execute(
@@ -918,165 +897,6 @@ 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,
         *,
@@ -1835,29 +1655,6 @@ 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,
         *,
@@ -2132,6 +1929,168 @@ class ControlPlaneDB:
                 ),
             )
 
+    def retire_lease_worktree_path(
+        self,
+        lease_id: str,
+        *,
+        expected_path: str | None = None,
+        expected_status: str | None = None,
+        expected_session_id: str | None = None,
+        expected_owner_pid: int | None = None,
+        reason: str = "missing_worktree_path_retired",
+    ) -> dict[str, Any]:
+        """Retire a missing worktree_path binding from a control-plane lease (#970).
+
+        Clears worktree_path on the lease row, updates provenance_json with
+        durable retirement audit proof, and writes a worktree_binding_retired
+        event.
+
+        The update is a compare-and-swap (#970 review 644 B1/B3): the caller
+        states the exact path it audited and, when known, the lease status,
+        owning session, and owner pid it classified against. Every stated value
+        must still match the stored row, and the ``UPDATE`` itself is keyed on
+        the stored ``worktree_path``, so a concurrent writer that moved or
+        replaced the binding between audit and apply loses the race instead of
+        having its value silently overwritten. A mismatch raises and mutates
+        nothing.
+
+        ``expected_path`` is mandatory: a retirement that does not name the path
+        it intends to clear cannot be safe against concurrent recreation.
+        """
+        now_s = _ts()
+        expected_norm = _realpath_or_raw(expected_path)
+        if not expected_norm:
+            raise ControlPlaneError(
+                f"cannot retire lease {lease_id} worktree_path: expected_path is "
+                "required for compare-and-swap retirement (fail closed)"
+            )
+        with self._tx() as conn:
+            cols = self._lease_columns(conn)
+            row = conn.execute(
+                "SELECT * FROM leases WHERE lease_id = ?",
+                (lease_id,),
+            ).fetchone()
+            if not row:
+                raise ControlPlaneError(f"unknown lease_id {lease_id}")
+
+            record = dict(row)
+            current_wt = (record.get("worktree_path") or "").strip()
+
+            if not current_wt:
+                # Idempotent: the binding this caller audited is already gone.
+                return {
+                    "lease_id": lease_id,
+                    "retired": False,
+                    "already_retired": True,
+                    "prior_worktree_path": "",
+                    "expected_worktree_path": expected_path,
+                    "reason": reason,
+                    "compare_and_swap": {
+                        "matched": True,
+                        "outcome": "already_retired",
+                    },
+                }
+
+            if _realpath_or_raw(current_wt) != expected_norm:
+                raise ControlPlaneError(
+                    f"cannot retire lease {lease_id} worktree_path: expected "
+                    f"'{expected_path}' does not match current '{current_wt}' "
+                    "(fail closed)"
+                )
+
+            for field, expected_value in (
+                ("status", expected_status),
+                ("session_id", expected_session_id),
+            ):
+                if expected_value is None:
+                    continue
+                current_value = record.get(field)
+                if str(current_value or "").strip() != str(expected_value).strip():
+                    raise ControlPlaneError(
+                        f"cannot retire lease {lease_id} worktree_path: lease "
+                        f"{field} changed since audit (expected "
+                        f"'{expected_value}', found '{current_value}'); fail closed"
+                    )
+
+            if expected_owner_pid is not None:
+                current_pid = record.get("owner_pid")
+                if current_pid is not None and int(current_pid) != int(expected_owner_pid):
+                    raise ControlPlaneError(
+                        f"cannot retire lease {lease_id} worktree_path: lease "
+                        f"owner_pid changed since audit (expected "
+                        f"{expected_owner_pid}, found {current_pid}); fail closed"
+                    )
+
+            # Parse and update provenance_json
+            raw_prov = record.get("provenance_json") or "{}"
+            try:
+                prov = json.loads(raw_prov) if isinstance(raw_prov, str) else dict(raw_prov)
+            except Exception:
+                prov = {}
+            if not isinstance(prov, dict):
+                prov = {}
+
+            prior_path = current_wt
+            prov.update({
+                "worktree_path_retired": True,
+                "retired_worktree_path": prior_path,
+                "retired_at": now_s,
+                "retirement_reason": reason,
+                "retired_from_status": record.get("status"),
+                "retired_from_session_id": record.get("session_id"),
+                "worktree_path": "",
+            })
+            prov_json = json.dumps(prov)
+
+            if "worktree_path" in cols:
+                # CAS: keyed on the exact stored path this caller audited.
+                cur = conn.execute(
+                    "UPDATE leases SET worktree_path = '', provenance_json = ? "
+                    "WHERE lease_id = ? AND worktree_path = ?",
+                    (prov_json, lease_id, record.get("worktree_path")),
+                )
+                if cur.rowcount != 1:
+                    raise ControlPlaneError(
+                        f"cannot retire lease {lease_id} worktree_path: "
+                        "compare-and-swap matched no row (concurrent change); "
+                        "fail closed"
+                    )
+            else:
+                conn.execute(
+                    "UPDATE leases SET provenance_json = ? WHERE lease_id = ?",
+                    (prov_json, lease_id),
+                )
+
+            conn.execute(
+                """
+                INSERT INTO events(work_item_id, event_type, message, created_at)
+                VALUES (?, 'worktree_binding_retired', ?, ?)
+                """,
+                (
+                    record["work_item_id"],
+                    f"lease {lease_id} worktree_path '{prior_path}' retired: {reason}",
+                    now_s,
+                ),
+            )
+
+            return {
+                "lease_id": lease_id,
+                "retired": True,
+                "already_retired": False,
+                "prior_worktree_path": prior_path,
+                "expected_worktree_path": expected_path,
+                "retired_at": now_s,
+                "reason": reason,
+                "compare_and_swap": {
+                    "matched": True,
+                    "outcome": "retired",
+                    "expected_status": expected_status,
+                    "expected_session_id": expected_session_id,
+                    "expected_owner_pid": expected_owner_pid,
+                },
+            }
+
+
     def abandon_lease(
         self,
         *,
@@ -3270,3 +3229,123 @@ class ControlPlaneDB:
             "live_lease_id": None if live_lease_id is None else str(live_lease_id),
             "reconcile_action": "reconcile_required" if stale else "safe_to_resume",
         }
+
+    def retire_session_checkpoint_worktree_path(
+        self,
+        session_id: str,
+        *,
+        checkpoint_id: str | None = None,
+        expected_path: str | None = None,
+        expected_status: str | None = None,
+        reason: str = "missing_worktree_path_retired",
+    ) -> dict[str, Any]:
+        """Retire a missing worktree_path from session_checkpoints (#970).
+
+        Compare-and-swap, mirroring :meth:`retire_lease_worktree_path` (#970
+        review 644 B3). ``expected_path`` names the exact stored path the caller
+        audited; the guarded ``UPDATE`` is keyed on that stored value, so a
+        checkpoint whose path was moved, replaced, or concurrently rewritten
+        after the audit is refused without mutation rather than blindly cleared.
+
+        Exactly one checkpoint row is targeted: by ``checkpoint_id`` when given,
+        otherwise by ``session_id``, which must identify a single row.
+        """
+        now_s = _ts()
+        expected_norm = _realpath_or_raw(expected_path)
+        if not expected_norm:
+            raise ControlPlaneError(
+                "cannot retire session checkpoint worktree_path: expected_path "
+                "is required for compare-and-swap retirement (fail closed)"
+            )
+        if not checkpoint_id and not (session_id or "").strip():
+            raise ControlPlaneError(
+                "cannot retire session checkpoint worktree_path: checkpoint_id "
+                "or session_id is required (fail closed)"
+            )
+
+        with self._tx() as conn:
+            if checkpoint_id:
+                selector_sql = "SELECT * FROM session_checkpoints WHERE checkpoint_id = ?"
+                selector_params: tuple[Any, ...] = (checkpoint_id,)
+                selector_desc = f"checkpoint_id '{checkpoint_id}'"
+            else:
+                selector_sql = "SELECT * FROM session_checkpoints WHERE session_id = ?"
+                selector_params = (session_id,)
+                selector_desc = f"session_id '{session_id}'"
+
+            rows = [dict(r) for r in conn.execute(selector_sql, selector_params).fetchall()]
+            if not rows:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path: no "
+                    f"checkpoint matches {selector_desc} (fail closed)"
+                )
+            if len(rows) > 1:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path: "
+                    f"{selector_desc} matches {len(rows)} checkpoints; supply an "
+                    "exact checkpoint_id (fail closed)"
+                )
+
+            record = rows[0]
+            target_checkpoint_id = record.get("checkpoint_id")
+            current_wt = (record.get("worktree_path") or "").strip()
+
+            if not current_wt:
+                # Idempotent: the binding this caller audited is already gone.
+                return {
+                    "session_id": session_id,
+                    "checkpoint_id": target_checkpoint_id,
+                    "retired": False,
+                    "already_retired": True,
+                    "prior_worktree_path": "",
+                    "expected_worktree_path": expected_path,
+                    "reason": reason,
+                    "compare_and_swap": {
+                        "matched": True,
+                        "outcome": "already_retired",
+                    },
+                }
+
+            if _realpath_or_raw(current_wt) != expected_norm:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path for "
+                    f"{selector_desc}: expected '{expected_path}' does not match "
+                    f"current '{current_wt}' (fail closed)"
+                )
+
+            if expected_status is not None:
+                current_status = record.get("status")
+                if str(current_status or "").strip() != str(expected_status).strip():
+                    raise ControlPlaneError(
+                        f"cannot retire session checkpoint worktree_path for "
+                        f"{selector_desc}: status changed since audit (expected "
+                        f"'{expected_status}', found '{current_status}'); fail closed"
+                    )
+
+            cur = conn.execute(
+                "UPDATE session_checkpoints SET worktree_path = '', updated_at = ? "
+                "WHERE checkpoint_id = ? AND worktree_path = ?",
+                (now_s, target_checkpoint_id, record.get("worktree_path")),
+            )
+            if cur.rowcount != 1:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path for "
+                    f"{selector_desc}: compare-and-swap matched no row "
+                    "(concurrent change); fail closed"
+                )
+
+            return {
+                "session_id": session_id,
+                "checkpoint_id": target_checkpoint_id,
+                "retired": True,
+                "already_retired": False,
+                "prior_worktree_path": current_wt,
+                "expected_worktree_path": expected_path,
+                "retired_at": now_s,
+                "reason": reason,
+                "compare_and_swap": {
+                    "matched": True,
+                    "outcome": "retired",
+                    "expected_status": expected_status,
+                },
+            }

docs/mcp-tool-inventory.md

@@ -65,6 +65,7 @@ that gates each call, not which tools exist.
 - `gitea_assess_work_issue_duplicate`
 - `gitea_assess_worktree_cleanup_integrity`
 - `gitea_audit_config`
+- `gitea_audit_missing_worktree_bindings`
 - `gitea_audit_runtime_recovery_contamination`
 - `gitea_audit_stable_branch_contamination`
 - `gitea_audit_worktree_cleanup`
@@ -126,16 +127,20 @@ that gates each call, not which tools exist.
 - `gitea_post_heartbeat`
 - `gitea_publish_unpublished_issue_branch`
 - `gitea_quarantine_contaminated_review`
+- `gitea_rebind_dirty_same_claimant_author_session`
 - `gitea_reclaim_expired_workflow_lease`
+- `gitea_reconcile_after_restart`
 - `gitea_reconcile_already_landed_pr`
 - `gitea_reconcile_issue_claims`
 - `gitea_reconcile_merged_cleanups`
+- `gitea_reconcile_missing_worktree_bindings`
 - `gitea_reconcile_superseded_by_merged_pr`
 - `gitea_record_daemon_process_kill_attempt`
 - `gitea_record_irrecoverable_decision_lock_provenance`
 - `gitea_record_pre_review_command`
 - `gitea_record_shell_spawn_outcome`
 - `gitea_record_stable_branch_push_attempt`
+- `gitea_recover_dirty_orphaned_issue_worktree`
 - `gitea_recover_incomplete_bootstrap_lock`
 - `gitea_release_merger_pr_lease`
 - `gitea_release_reviewer_pr_lease`

docs/post-restart-reconcile.md

@@ -19,11 +19,7 @@ The assessor classifies:
 
 - **service_health** — process healthy / parity mutation-safe
 - **clients** — connected client descriptors (optional inventory)
-- **sessions** — active session rows with dead or reused owner pids are
+- **sessions** — active session rows with dead owner pids are unresolved
-  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

@@ -13547,6 +13547,162 @@ def gitea_scan_already_landed_open_prs(
     }
 
 
+@mcp.tool()
+def gitea_audit_missing_worktree_bindings(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: audit and classify worktree bindings whose paths are missing on disk (#970).
+
+    Correlates missing-path bindings from control-plane leases, session checkpoints,
+    and issue locks with repository, host, branch, issue/PR, session, and lease state.
+    Distinguishes deleted worktrees from moved paths, host/mount failures, and live
+    leases/sessions.
+
+    Args:
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with audit counts, missing binding classifications, and resolution status.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    import missing_worktree_reconcile
+    db, _ = _control_plane_db_or_error()
+    root = _canonical_local_git_root()
+    h, o, r = _resolve(remote, host, org, repo)
+
+    return missing_worktree_reconcile.audit_missing_worktree_bindings(
+        db,
+        project_root=root,
+        remote=remote,
+        org=o,
+        repo=r,
+        host=h,
+    )
+
+
+@mcp.tool()
+def gitea_reconcile_missing_worktree_bindings(
+    dry_run: bool = True,
+    # Deprecated: retained so callers that still pass it get an explicit deny.
+    operator_authorized: bool = False,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Audit and safely resolve (retire) missing worktree path bindings (#970).
+
+    Identifies confirmed stale deleted worktree bindings, re-validates them
+    against a fresh authoritative read immediately before mutation to prevent
+    recreation and ownership races, and retires only the exact stale bindings
+    while preserving unrelated worktrees and Git metadata.
+
+    Apply mode (``dry_run=False``) is authorized **server-side** (#970 review
+    644 B2): it requires the reconciler-only ``gitea.branch.delete``
+    capability, the resolved ``reconcile_missing_worktree_bindings`` task
+    capability, and an active cleanup phase minted through
+    ``gitea_authorize_reconciliation_cleanup_phase``. A caller-supplied
+    ``operator_authorized`` is never authorization evidence (#709 F1 /
+    review 434) and is rejected outright. Dry-run stays available to any
+    ``gitea.read`` profile and never mutates.
+
+    Args:
+        dry_run: If True (default), reports planned mutations without modifying state.
+        operator_authorized: Rejected. Authorization is a server-side artifact.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with before/after audit state, resolutions, and dimension status.
+    """
+    import missing_worktree_reconcile
+
+    # Explicitly reject the self-assertable Boolean before anything else, so it
+    # can never combine with a legitimate gate to authorize a mutation.
+    if operator_authorized:
+        return {
+            "success": False,
+            "performed": False,
+            "reason": "operator_authorized_rejected",
+            "reasons": [missing_worktree_reconcile.OPERATOR_AUTHORIZED_REJECTION],
+            "exact_next_action": (
+                "authorize cleanup via gitea_authorize_reconciliation_cleanup_phase "
+                "from a reconciler profile, then re-run with dry_run=False"
+            ),
+        }
+
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    db, db_errs = _control_plane_db_or_error()
+    root = _canonical_local_git_root()
+    h, o, r = _resolve(remote, host, org, repo)
+
+    profile = get_profile()
+    role = (profile.get("role") or profile.get("role_kind") or "").strip().lower()
+
+    cleanup_authorization = None
+    if not dry_run:
+        cleanup_task = missing_worktree_reconcile.CLEANUP_TASK
+        required_permission = task_capability_map.required_permission(cleanup_task)
+        capability_blockers = _profile_operation_gate(required_permission)
+        role_matches = role == task_capability_map.required_role(cleanup_task)
+        cleanup_authorization = missing_worktree_reconcile.assess_cleanup_authorization(
+            role=role,
+            capability_blockers=capability_blockers,
+            operator_authorized=False,
+            task_capability_resolved=(not capability_blockers) and role_matches,
+        )
+        if not cleanup_authorization.get("authorized"):
+            return {
+                "success": False,
+                "performed": False,
+                "reason": "cleanup_authorization_required",
+                "reasons": cleanup_authorization.get("reasons") or [],
+                "cleanup_authorization": cleanup_authorization,
+                "permission_report": _permission_block_report(required_permission),
+                "exact_next_action": (
+                    "resolve the reconcile_missing_worktree_bindings capability "
+                    "from a reconciler profile and authorize cleanup via "
+                    "gitea_authorize_reconciliation_cleanup_phase"
+                ),
+            }
+
+    return missing_worktree_reconcile.reconcile_missing_worktree_bindings(
+        db,
+        project_root=root,
+        remote=remote,
+        org=o,
+        repo=r,
+        host=h,
+        dry_run=dry_run,
+        operator_authorized=False,
+        cleanup_authorization=cleanup_authorization,
+    )
+
+
 @mcp.tool()
 def gitea_audit_worktree_cleanup(
     remote: str = "dadeschools",
@@ -24432,18 +24588,10 @@ 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 / #969).
+    """Gather + classify post-restart state; cache the latest proof (#662)."""
-
-    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)
@@ -24457,48 +24605,13 @@ 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"] = not apply_session_cleanup
+    payload["read_only"] = True
     payload["remote"] = remote
     payload["org"] = o
     payload["repo"] = r
@@ -24507,9 +24620,6 @@ 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
@@ -24536,9 +24646,8 @@ 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 / #969).
+    """Run post-restart MCP reconciliation and return a completion proof (#662).
 
     Gathers live control-plane sessions, leases, worktree bindings, and
     master-parity evidence, then classifies them with the pure
@@ -24546,17 +24655,12 @@ def gitea_reconcile_after_restart(
     machine-readable completion proof listing resolved / unresolved dimensions
     and proposed durable follow-up issues.
 
-    Never restarts MCP, never auto-resumes write mutations, and never creates
+    Read-only by design: never restarts MCP, never auto-resumes write
-    Gitea issues. Default mode is ``log_only``; set
+    mutations, and never creates Gitea issues (those are a separate apply
+    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.
@@ -24576,96 +24680,9 @@ 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,65 +475,24 @@ def reconcile_after_restart(
             )
         )
 
-    # --- sessions (#969: dead-owner / PID-reuse lifecycle) ---------------
+    # --- sessions -------------------------------------------------------
-    # 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 = [
+    orphan_sessions = [
-        L for L in (inventory.get("leases") or []) if isinstance(L, Mapping)
+        s
+        for s in sessions
+        if str(s.get("status") or "").lower() == "active"
+        and s.get("pid") is not None
+        and not lease_lifecycle.is_process_alive(s.get("pid"))
     ]
-    fleet_report = inventory.get("session_fleet")
+    if orphan_sessions:
-    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
-                if str(s.get("status") or "").lower() == "active"
-                and s.get("pid") is not None
-                and not lease_lifecycle.is_process_alive(s.get("pid"))
-            ]
-            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(retireable_ids)} session row(s) with dead/reused owner "
+                f"{len(orphan_sessions)} active session row(s) with dead owner pid",
-                f"await retirement",
                 details={
-                    "orphan_session_ids": retireable_ids,
+                    "orphan_session_ids": [s.get("session_id") for s in orphan_sessions],
-                    "retireable_session_ids": retireable_ids,
                     "total_sessions": len(sessions),
-                    "fleet": fleet_details,
                 },
                 follow_up=True,
             )
@@ -543,12 +502,8 @@ def reconcile_after_restart(
             _item(
                 DIM_SESSIONS,
                 ITEM_RESOLVED,
-                f"{len(sessions)} session row(s) reconciled "
+                f"{len(sessions)} session row(s) reconciled (no dead-pid orphans)",
-                f"(no retireable dead/reused owners)",
+                details={"total_sessions": len(sessions)},
-                details={
-                    "total_sessions": len(sessions),
-                    "fleet": fleet_details,
-                },
             )
         )
 

session_lifecycle.py

@@ -1,703 +0,0 @@
-"""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,9 +153,8 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.push",
         "role": "author",
     },
-    # #662: post-restart reconcile is inventory + pure classification.
+    # #662: post-restart reconcile is read-only inventory + pure classification.
-    # #969: optional apply_session_cleanup retires confirmed-stale session rows
+    # Durable follow-up issue creation is a separate apply path (not this task).
-    # through the same tool; durable follow-up Gitea issues remain separate.
     "reconcile_after_restart": {
         "permission": "gitea.read",
         "role": "author",
@@ -164,15 +163,6 @@ 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",
@@ -396,6 +386,25 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.delete",
         "role": "reconciler",
     },
+    # #970: auditing missing worktree bindings is read-only; retiring one is a
+    # control-plane cleanup mutation and carries the same reconciler-only
+    # authority as any other reconciliation cleanup (review 644 B2).
+    "audit_missing_worktree_bindings": {
+        "permission": "gitea.read",
+        "role": "reconciler",
+    },
+    "gitea_audit_missing_worktree_bindings": {
+        "permission": "gitea.read",
+        "role": "reconciler",
+    },
+    "reconcile_missing_worktree_bindings": {
+        "permission": "gitea.branch.delete",
+        "role": "reconciler",
+    },
+    "gitea_reconcile_missing_worktree_bindings": {
+        "permission": "gitea.branch.delete",
+        "role": "reconciler",
+    },
     "work_issue": {
         "permission": "gitea.pr.create",
         "role": "author",
@@ -663,6 +672,8 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
         "delete_branch",
         "cleanup_merged_pr_branch",
         "reconciliation_cleanup",
+        "reconcile_missing_worktree_bindings",
+        "gitea_reconcile_missing_worktree_bindings",
         "work_issue",
         "work-issue",
     }

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"], "6")
+        self.assertEqual(rows["schema_version"], "5")
         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"], 6)
+        self.assertEqual(record["checkpoint_schema_version"], 5)
 
     # 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

@@ -1,480 +0,0 @@
-"""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()

tests/test_task_capability_role_invariants.py

@@ -153,6 +153,12 @@ EXPECTED_ROLE_EXCLUSIVE_TASKS = frozenset(
         "delete_branch",
         "cleanup_merged_pr_branch",
         "reconciliation_cleanup",
+        # #970 review 644 B2: retiring a missing worktree binding is a
+        # control-plane cleanup mutation, so it carries the same reconciler-only
+        # authority as every other reconciliation cleanup. Permission alone must
+        # not authorize it.
+        "reconcile_missing_worktree_bindings",
+        "gitea_reconcile_missing_worktree_bindings",
         "work_issue",
         "work-issue",
     }