feat(controller): expose instance-level fleet identity and health snapshots

Add a pure fleet snapshot assessor and a read-only controller/reconciler
MCP tool so multi-instance fleets can be enumerated by client_instance_id
without treating shared client_type as duplication. Trusted launcher
instance IDs, classification (missing/unmanifested/collisions/historical),
registry schema extensions, docs, and regression tests for #978.

Closes #978

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

canonical_repository_root.py

@@ -37,6 +37,48 @@ CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT"
 # Candidate git remote names probed when deriving repository identity.
 _IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps")
 
+# Repository-authority modes (#973 B10). Exactly two values are supported.
+# ``mode`` selects how repository authority is established, so an unrecognised
+# value must never be normalised onto one of these: aliasing a trusted mode is
+# precisely the defect. Omitting the argument keeps the documented safe default,
+# ``validation``.
+MODE_VALIDATION = "validation"
+MODE_DERIVATION = "derivation"
+SUPPORTED_MODES: tuple[str, ...] = (MODE_VALIDATION, MODE_DERIVATION)
+
+# Reason code emitted when an unsupported mode is refused. Mirrors the existing
+# ``webui.sanctioned_restart.DENY_UNKNOWN_MODE`` convention so callers and tests
+# can assert the refusal cause rather than string-matching prose.
+DENY_UNKNOWN_MODE = "unknown_mode"
+
+
+def unsupported_mode_reason(mode: object) -> str | None:
+    """Precise rejection reason for *mode*, or None when *mode* is supported.
+
+    Only the two documented string values are accepted, compared exactly — no
+    stripping, no case folding — so misspellings and whitespace variants are
+    refused rather than coerced. An empty string, ``None``, and any non-string
+    are all *explicitly supplied* unsupported values and are refused on the same
+    footing; none of them is normalised to a supported mode. Omitting the
+    argument entirely never reaches here with an unsupported value because the
+    parameter default is ``"validation"``.
+    """
+    if isinstance(mode, str) and mode in SUPPORTED_MODES:
+        return None
+    supported = ", ".join(repr(m) for m in SUPPORTED_MODES)
+    if not isinstance(mode, str):
+        return (
+            f"unsupported repository-authority mode {mode!r} of type "
+            f"{type(mode).__name__}: only {supported} are supported; the mode "
+            "is refused before any repository assessment and no repository "
+            "identity was resolved through it (fail closed)"
+        )
+    return (
+        f"unsupported repository-authority mode {mode!r}: only {supported} are "
+        "supported; the mode is refused before any repository assessment and no "
+        "repository identity was resolved through it (fail closed)"
+    )
+
 
 def configured_canonical_root(
     profile: Mapping | None,
@@ -132,6 +174,7 @@ def assess_canonical_repository_root(
     process_project_root: str,
     remote: str | None = None,
     require_binding: bool = False,
+    mode: str = "validation",
 ) -> dict:
     """Validate the canonical repository root binding, failing closed on forgery.
 
@@ -140,15 +183,41 @@ def assess_canonical_repository_root(
     ``configured`` (whether a cross-repo binding was declared),
     ``resolved_slug`` and ``source``.
 
-    Without a configured binding the single-repo default is preserved: the
+    *mode* accepts exactly the two values in :data:`SUPPORTED_MODES`:
-    canonical root is derived from *process_project_root* and never blocks
+    - ``"validation"`` (default): an independently trusted expected repository
-    (unless *require_binding* explicitly demands one).
+      slug is known (or required). Candidate root's observed identity must match.
+      Unprovable or missing expected identity fails closed when *require_binding* is True.
+    - ``"derivation"``: caller is deriving the canonical repository identity.
+      No expected slug exists yet by design. Derivation succeeds if the configured
+      root exists, is a git repository, and carries a resolvable git remote identity.
 
-    With a configured binding the path must exist, be a git repository, and —
+    Every other explicitly supplied value — unknown strings, misspellings, the
-    when *expected_slug* is known — carry a matching repository identity. A
+    empty string, ``None``, and non-strings — is refused with ``proven`` False,
-    mismatched or (when *require_binding*) unprovable identity is a forged or
+    ``block`` True, and ``reason_code`` :data:`DENY_UNKNOWN_MODE` (#973 B10).
-    conflicting binding and fails closed.
+    Omitting *mode* entirely keeps the documented ``"validation"`` default.
     """
+    # #973 B10: refuse an unsupported repository-authority mode as the very first
+    # act, before any candidate-root existence check, path or symlink resolution,
+    # git top-level discovery, remote-URL or repository-identity discovery, and
+    # before any expected-versus-observed comparison or validation/derivation
+    # behaviour. An unsupported mode previously fell into the catch-all ``else``
+    # on the configured-root path (silently receiving validation semantics) and
+    # skipped the identity comparison entirely on the single-repository default
+    # path (strictly weaker than validation), so matching identities could return
+    # ``proven`` True. No repository identity may be resolved through a mode the
+    # contract does not define.
+    mode_reason = unsupported_mode_reason(mode)
+    if mode_reason is not None:
+        return _assessment(
+            proven=False,
+            reasons=[mode_reason],
+            configured=bool((configured_value or "").strip()),
+            canonical_repo_root=None,
+            resolved_slug=None,
+            source=source,
+            reason_code=DENY_UNKNOWN_MODE,
+        )
+
     process_root = os.path.realpath(process_project_root)
     declared = (configured_value or "").strip()
 
@@ -168,12 +237,31 @@ def assess_canonical_repository_root(
             )
         # Single-repo default: canonical root follows the install checkout.
         derived = resolve_repo_toplevel(process_root) or process_root
+        resolved_slug = repository_identity_slug(derived, remote=remote)
+        reasons: list[str] = []
+        # #973 B10: the allowlist above guarantees *mode* is one of the two
+        # supported values here, so an unsupported value can no longer skip this
+        # identity comparison and end up strictly weaker than validation.
+        if mode == MODE_VALIDATION and expected_slug:
+            expected = expected_slug.strip()
+            if resolved_slug and resolved_slug.lower() != expected.lower():
+                reasons.append(
+                    f"canonical repository root identity mismatch: '{derived}' resolves "
+                    f"to repository '{resolved_slug}' but expected repository identity "
+                    f"is '{expected}' (forged or conflicting binding, fail closed)"
+                )
+            elif not resolved_slug and require_binding:
+                reasons.append(
+                    f"canonical repository root '{derived}' has no resolvable git "
+                    f"remote identity to confirm authorization for '{expected}' "
+                    "(fail closed)"
+                )
         return _assessment(
-            proven=True,
+            proven=not reasons,
-            reasons=[],
+            reasons=reasons,
             configured=False,
             canonical_repo_root=derived,
-            resolved_slug=None,
+            resolved_slug=resolved_slug,
             source=None,
         )
 
@@ -207,6 +295,17 @@ def assess_canonical_repository_root(
 
     resolved_slug = repository_identity_slug(toplevel, remote=remote)
     reasons: list[str] = []
+
+    if mode == MODE_DERIVATION:
+        if not resolved_slug:
+            reasons.append(
+                f"configured canonical repository root '{toplevel}' has no resolvable "
+                "git remote identity (fail closed)"
+            )
+    else:
+        # #973 B10: MODE_VALIDATION only. This arm is no longer a catch-all — the
+        # allowlist above admits no third value, so an unsupported mode can no
+        # longer silently receive validation semantics here.
         expected = (expected_slug or "").strip() or None
         if expected:
             if resolved_slug and resolved_slug.lower() != expected.lower():
@@ -221,6 +320,12 @@ def assess_canonical_repository_root(
                     f"remote identity to confirm authorization for '{expected}' "
                     "(fail closed)"
                 )
+        elif require_binding:
+            reasons.append(
+                f"canonical repository root '{toplevel}' has configured value "
+                f"'{configured_value}' but authoritative expected repository identity "
+                "is unprovable or missing (fail closed)"
+            )
 
     return _assessment(
         proven=not reasons,
@@ -252,10 +357,19 @@ def _assessment(
     proven: bool,
     reasons: list[str],
     configured: bool,
-    canonical_repo_root: str,
+    canonical_repo_root: str | None,
     resolved_slug: str | None,
     source: str | None,
+    reason_code: str | None = None,
 ) -> dict:
+    """Build the assessment payload.
+
+    ``canonical_repo_root`` is None only when the assessment refused to resolve
+    one at all — today exactly the unsupported-mode refusal (#973 B10), which
+    must not derive a trusted repository identity through an undefined mode.
+    ``reason_code`` is a machine-checkable refusal cause; None for every
+    ordinary (non-coded) outcome.
+    """
     return {
         "proven": proven,
         "block": not proven,
@@ -264,4 +378,5 @@ def _assessment(
         "canonical_repo_root": canonical_repo_root,
         "resolved_slug": resolved_slug,
         "source": source,
+        "reason_code": reason_code,
     }

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/instance-fleet-identity.md

@@ -0,0 +1,171 @@
+# Instance-level fleet identity and health snapshots
+
+Issue **#978**. Companion primitives: **#948** (worker ownership), **#975**
+(heartbeat lifecycle), **#951** (restart receipts).
+
+## Why this exists
+
+The multi-instance fleet gate (#963) needs a *native* answer to:
+
+* which application launches are live;
+* which of the five namespace workers belong to each launch;
+* whether two Codex (or Claude, Grok, …) launches are distinct;
+* whether any live collision is real rather than a shared client type.
+
+Process tables, PID proximity, configuration files, and direct SQLite access
+are **not** production evidence. The sanctioned surface is the read-only MCP
+tool `gitea_snapshot_instance_fleet` on **controller** and **reconciler**
+namespaces.
+
+## Identity hierarchy
+
+| Identity | Scope | Who generates it | Lifetime |
+| --- | --- | --- | --- |
+| `client_type` | Application family (`codex`, `claude_code`, `gemini`, `grok`, …) | Launcher sets `GITEA_MCP_CLIENT` | Stable for the product |
+| `client_instance_id` | One running application launch | **Trusted launcher**, once per launch, as `GITEA_MCP_CLIENT_INSTANCE` | Fresh launch → new ID; reconnect of same launch → same ID; full app restart → new ID |
+| `fleet_run_id` | Operator-approved enrollment / canary cohort | Operator / controller sets `GITEA_MCP_FLEET_RUN_ID` | Duration of the approved rollout |
+| `namespace` | `author` \| `reviewer` \| `merger` \| `controller` \| `reconciler` | Profile / MCP server binding | Process lifetime |
+| `worker_id` / `worker_identity` | One namespace worker process | Runtime registry at first registration | New on worker restart; not reused |
+| `session_id` | Client session ownership | Launcher `GITEA_MCP_CLIENT_SESSION` or runtime | Session lifetime |
+| `generation_id` | One daemon launch | Runtime at process boot | New on process restart |
+| `process_identity` / PID | OS process | Runtime | Process lifetime |
+
+### Rules
+
+1. Multiple active instances **may** share the same `client_type`.
+2. Every application launch receives a **distinct** `client_instance_id`.
+3. All five namespace workers of one launch report the **same**
+   `client_instance_id`.
+4. Each namespace worker has a **distinct** `worker_identity`, process
+   identity, generation, and PID.
+5. Instance identity is **never** inferred from PID proximity, timestamps, or
+   client type alone.
+6. Live reuse of one `client_instance_id` with conflicting workers fails closed.
+7. Sharing only a profile or `client_type` is **not** a duplicate.
+
+This deliberately **replaces** any permanent `exactly_one_per_profile` fleet
+model (#949 assumption) as the operating rule for multi-instance fleets.
+
+## How five workers join one instance
+
+1. The host starts one application instance (for example one Codex session).
+2. The trusted launcher mints `client_instance_id` (see
+   `mcp_fleet_snapshot.generate_client_instance_id`) and injects:
+
+   ```text
+   GITEA_MCP_CLIENT=<client_type>
+   GITEA_MCP_CLIENT_INSTANCE=<client_instance_id>
+   GITEA_MCP_CLIENT_SESSION=<session_id>          # optional but recommended
+   GITEA_MCP_FLEET_RUN_ID=<enrollment id>         # when on an approved canary
+   ```
+
+3. The launcher starts the five MCP namespace processes (or attaches five
+   role profiles) with **that same environment**.
+4. Each worker registers once into the worker registry with its own
+   `worker_identity`, `namespace`, `generation_id`, and PID, but the shared
+   `client_instance_id`.
+
+Worker reconnect (same process, same registration) keeps the instance ID.
+Worker restart (new process) mints a new worker identity and generation but
+must still receive the same `GITEA_MCP_CLIENT_INSTANCE` from the parent
+application if it is the same launch. Full application restart mints a new
+`client_instance_id`.
+
+## Approved fleet manifest
+
+An operator (or controller enrollment step) obtains an approved manifest as a
+list of expected instances, for example:
+
+```json
+{
+  "instances": [
+    {
+      "client_type": "codex",
+      "client_instance_id": "inst-codex-20260730T120000Z-abc123def456",
+      "fleet_run_id": "canary-963-2026-07-30",
+      "namespaces": ["author", "reviewer", "merger", "controller", "reconciler"]
+    },
+    {
+      "client_type": "codex",
+      "client_instance_id": "inst-codex-20260730T120100Z-fed654cba321",
+      "fleet_run_id": "canary-963-2026-07-30",
+      "namespaces": ["author", "reviewer", "merger", "controller", "reconciler"]
+    }
+  ]
+}
+```
+
+Pass that list as `expected_manifest` to `gitea_snapshot_instance_fleet`.
+When a manifest is supplied:
+
+* instances on the manifest but not live → `missing_expected`;
+* live instances not on the manifest → `unmanifested`;
+* both are **active blockers** for fleet-gate safety.
+
+Without a manifest, the snapshot still enumerates the live fleet and classifies
+identity collisions; it does not invent enrollment policy.
+
+## Snapshot consistency
+
+Each snapshot includes:
+
+* `snapshot_at` — UTC timestamp;
+* `consistency_token` / `registry_revision` — content digest over worker
+  identity, instance id, status, heartbeat, fencing, and generation.
+
+Two successive snapshots can prove heartbeat continuity via
+`mcp_fleet_snapshot.compare_snapshot_heartbeats`.
+
+## Classification (active vs historical)
+
+**Active blockers** (make `live_fleet_safe=false`):
+
+* missing expected instance;
+* unmanifested extra instance;
+* duplicate namespace worker within one instance;
+* live `client_instance_id` collision;
+* reused worker / session / generation / process / PID / fencing identity;
+* orphaned or unowned workers;
+* unknown client;
+* foreign-repository workers;
+* old-revision workers;
+* stale workers still marked active;
+* legacy incomplete instance identity.
+
+**Historical dead rows** (`status` released/superseded) are reported as
+`historical_dead` findings with `active_blocker=false`. They never
+automatically make the live fleet unsafe.
+
+## Fail-closed behaviour and recovery
+
+| Condition | Mutation safety | Diagnostic reads | Recovery |
+| --- | --- | --- | --- |
+| Two instances, same `client_type`, distinct IDs | Safe (if otherwise healthy) | Available | None needed |
+| Live reuse of one `client_instance_id` | Unsafe | Available | Stop the colliding launch or re-issue a distinct ID |
+| Two workers, same namespace, one instance | Unsafe | Available | Stop the extra worker |
+| Legacy registration without trusted instance ID | Unsafe for fleet mutation | Available | Relaunch with launcher-issued `GITEA_MCP_CLIENT_INSTANCE` |
+| Historical dead row only | Does not block alone | Available | No action required for fleet safety |
+| Registry unavailable | Snapshot fails closed | N/A | Repair registry path / reconnect namespaces |
+
+Incomplete or untrusted instance identity **cannot** authorize unsafe
+mutation. Diagnostic reads remain available where `gitea.read` allows.
+
+## Permissions
+
+* **Allowed:** `controller`, `reconciler` with `gitea.read`.
+* **Denied:** author, reviewer, merger (even with `gitea.read` for other tools).
+* **No new mutation permissions** are granted to any role.
+
+## Related surfaces
+
+* `mcp_worker_identity` — registry, worker identity, heartbeats (#948, #975).
+* `mcp_fleet_snapshot` — pure snapshot + classification (#978).
+* `gitea_snapshot_instance_fleet` — sanctioned MCP tool (#978).
+* `gitea_get_runtime_context` — single-process view (not fleet-wide).
+
+## Non-goals
+
+* Starting the #963 canary.
+* Purging historical registry rows.
+* Preserving “exactly one process per profile” as the permanent model.
+* Using shell / process-table / SQLite inspection as production fleet evidence.

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`
@@ -154,6 +159,7 @@ that gates each call, not which tools exist.
 - `gitea_sentry_reconcile_issue`
 - `gitea_sentry_watchdog`
 - `gitea_set_issue_labels`
+- `gitea_snapshot_instance_fleet`
 - `gitea_submit_pr_review`
 - `gitea_update_pr_branch_by_merge`
 - `gitea_validate_review_final_report`

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_config.py

@@ -1180,6 +1180,10 @@ RECOGNIZED_GITEA_ENV_KEYS = frozenset({
     "GITEA_SERVER_PROVENANCE",
     "GITEA_AUTHOR_WORKTREE",
     "GITEA_ACTIVE_WORKTREE",
+    "GITEA_REVIEWER_WORKTREE",
+    "GITEA_MERGER_WORKTREE",
+    "GITEA_CANONICAL_REPOSITORY_ROOT",
+    "GITEA_MCP_SESSION_STATE_TTL_HOURS",
     "GITEA_DISABLE_KEYCHAIN",
     "GITEA_CONTROL_PLANE_DB",
     "GITEA_DB_PATH",
@@ -1189,6 +1193,25 @@ RECOGNIZED_GITEA_ENV_KEYS = frozenset({
     "GITEA_IRRECOVERABLE_HMAC_SECRET",
     "GITEA_FORCE_MCP_RUNTIME_CHECK",
     "GITEA_FORCE_CLIENT_MANAGED",
+    # #975: the client-identity inputs the server actually consumes at startup
+    # (CLIENT_NAME_ENV / CLIENT_INSTANCE_ENV / CLIENT_SESSION_ENV in
+    # gitea_mcp_server). Production read them while this allowlist omitted them,
+    # so the peer-env scan classified them as unsupported overrides and the
+    # capability resolver refused every mutation fleet-wide. Named individually
+    # on purpose: no prefix is added, so an unrecognised GITEA_* override is
+    # still refused exactly as it was before.
+    "GITEA_MCP_CLIENT",
+    "GITEA_MCP_CLIENT_INSTANCE",
+    "GITEA_MCP_CLIENT_SESSION",
+    # #978: operator-approved fleet enrollment identity shared by one launch.
+    "GITEA_MCP_FLEET_RUN_ID",
+    "GITEA_MCP_PROCESS_IDENTITY",
+    # #975 review 652 B1: production also consumes HEARTBEAT_INTERVAL_ENV from
+    # mcp_worker_identity via gitea_mcp_server._start_worker_heartbeat. Omitting
+    # it reproduced the same unsupported-env → runtime_reconnect_required
+    # failure mode for the documented operator override. Named individually;
+    # no GITEA_* / GITEA_WORKER_* prefix is added.
+    "GITEA_WORKER_HEARTBEAT_INTERVAL_SECONDS",
 })
 
 RECOGNIZED_GITEA_ENV_PREFIXES = (

gitea_mcp_server.py

@@ -500,10 +500,66 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
     return workspace
 
 
-def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
+def _process_root_git_remote_url(remote_name: str) -> str | None:
+    """Best-effort local ``git remote get-url`` strictly inside ``PROJECT_ROOT``.
+
+    #973 (B8): Must derive expected repository identity from an authority
+    independent of the candidate configured canonical root.
+    """
+    try:
+        proc = subprocess.run(
+            ["git", "remote", "get-url", remote_name],
+            capture_output=True,
+            text=True,
+            cwd=PROJECT_ROOT,
+        )
+        if proc.returncode != 0:
+            return None
+        url = (proc.stdout or "").strip()
+        return url or None
+    except Exception:
+        return None
+
+
+def _resolve_expected_repository_slug(
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> str | None:
+    """Resolve expected repository slug from session context, parameters, or process root.
+
+    Must derive expected repository identity ONLY from trusted sources independent
+    of the candidate configured canonical root (#973 B8).
+
+    Bound session context takes precedence over request parameters so request-supplied
+    coordinates cannot replace a bound canonical root (#741 / #973 B9).
+    """
+    bound = session_ctx.get_session_context() or {}
+    b_org = bound.get("org")
+    b_repo = bound.get("repository")
+    if b_org and b_repo:
+        return session_ctx.format_repository_slug(b_org, b_repo)
+    if org and repo:
+        return session_ctx.format_repository_slug(org, repo)
+    eff_remote = remote or bound.get("remote") or _effective_remote()
+    parsed = remote_repo_guard.parse_org_repo_from_remote_url(
+        _process_root_git_remote_url(eff_remote)
+    )
+    if parsed:
+        return session_ctx.format_repository_slug(parsed[0], parsed[1])
+    return None
+
+
+def _resolve_namespace_mutation_context(
+    worktree_path: str | None = None,
+    remote: str | None = None,
+) -> dict:
     """Canonical namespace workspace + repository root for guards (#460/#510/#706/#618)."""
     role = _effective_workspace_role()
     configured_root, _source = _configured_canonical_root()
+    bound = session_ctx.get_session_context() or {}
+    eff_remote = remote or bound.get("remote") or _effective_remote()
+    expected_slug = _resolve_expected_repository_slug(eff_remote)
     return nwb.resolve_namespace_mutation_context(
         role_kind=role,
         worktree_path=worktree_path,
@@ -516,9 +572,12 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic
         ),
         profile_name=get_profile().get("profile_name"),
         configured_canonical_root=configured_root,
+        expected_slug=expected_slug,
+        remote=eff_remote,
     )
 
 
+
 def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
     """Backward-compatible alias for namespace workspace context."""
     return _resolve_namespace_mutation_context(worktree_path)
@@ -620,6 +679,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
         inspected_root = _get_git_root(workspace)
     process_root = ctx["process_project_root"]
     canonical_root = ctx["canonical_repo_root"]
+    crr_assessment = ctx.get("canonical_root_assessment") or {}
+    resolved_slug = crr_assessment.get("resolved_slug")
+    expected_slug = ctx.get("expected_slug")
     active_root = os.path.realpath(inspected_root or workspace)
     if active_root == canonical_root:
         dirty_scope = "control checkout"
@@ -649,6 +711,10 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
         "workspace_healthy": not bool(
             ctx.get("bound_worktree_missing") or ctx.get("author_worktree_block")
         ),
+        "canonical_root_assessment": crr_assessment,
+        "expected_repository_slug": expected_slug,
+        "observed_repository_identity": resolved_slug,
+        "worktree_registration_result": ctx.get("in_git_worktree_list"),
     }
     if not ctx["roots_aligned"]:
         details["workspace_root_mismatch"] = (
@@ -658,6 +724,7 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
     return details
 
 
+
 def _format_preflight_workspace_details(details: dict) -> str:
     parts = [
         f"MCP server process root: {details.get('mcp_server_process_root')}",
@@ -930,10 +997,7 @@ def _enforce_canonical_repository_root(
     if not configured_value:
         return
 
-    bound = session_ctx.get_session_context() or {}
+    expected_slug = _resolve_expected_repository_slug(remote)
-    expected_slug = session_ctx.format_repository_slug(
-        bound.get("org"), bound.get("repository")
-    )
     assessment = crr.assess_canonical_repository_root(
         configured_value=configured_value,
         source=source,
@@ -1852,8 +1916,16 @@ def _verify_role_mutation_workspace(
             if runtime_reasons:
                 raise RuntimeError("; ".join(runtime_reasons))
     except Exception as exc:
-        if "stale-runtime:" in str(exc):
+        # #975: ``_check_mcp_runtimes_diagnostics`` raises every reason it
-            raise RuntimeError(str(exc))
+        # produces through this one RuntimeError, but only ``stale-runtime:``
+        # was re-raised here — an ``unsupported-env:`` reason was swallowed
+        # while still failing the capability resolver, so this preflight and
+        # the resolver disagreed about the identical diagnostic. Both
+        # authoritative prefixes now propagate the same way. This can only ever
+        # widen what is refused, never widen what is permitted.
+        message = str(exc)
+        if any(prefix in message for prefix in RUNTIME_DIAGNOSTIC_HARD_PREFIXES):
+            raise RuntimeError(message)
         pass
 
     role = _effective_workspace_role()
@@ -1866,6 +1938,9 @@ def _verify_role_mutation_workspace(
     # back to the install checkout and validated Gitea-Tools/branches/ instead
     # of the target repository the namespace is actually bound to.
     _configured_root, _configured_source = _configured_canonical_root()
+    bound = session_ctx.get_session_context() or {}
+    eff_remote = remote or bound.get("remote") or _effective_remote()
+    expected_slug = _resolve_expected_repository_slug(eff_remote, org=org, repo=repo)
     assessment = nwb.assess_namespace_mutation_workspace(
         role_kind=role,
         worktree_path=worktree_path,
@@ -1880,6 +1955,8 @@ def _verify_role_mutation_workspace(
         profile_name=get_profile().get("profile_name"),
         current_branch=git_state.get("current_branch"),
         configured_canonical_root=_configured_root,
+        expected_slug=expected_slug,
+        remote=eff_remote,
     )
     if assessment["block"]:
         raise RuntimeError(
@@ -2184,6 +2261,7 @@ def _canonical_repository_slug(
         process_project_root=PROJECT_ROOT,
         remote=remote,
         require_binding=True,
+        mode="derivation",
     )
     slug = assessment.get("resolved_slug")
     if assessment.get("block") or not slug:
@@ -3414,7 +3492,7 @@ def cleanup_in_progress_for_pr(
 
 # ── Helpers ───────────────────────────────────────────────────────────────────
 
-def _effective_remote(remote: str) -> str:
+def _effective_remote(remote: str = "dadeschools") -> str:
     """If remote is the default ('dadeschools') but the active profile base_url maps to a known remote, use that remote instead."""
     try:
         profile = get_profile()
@@ -13547,6 +13625,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",
@@ -15121,22 +15355,17 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict:
     workspace_root = None
     aligned = None
     canonical_root = None
+    resolved_slug = None
     try:
         ctx = _resolve_namespace_mutation_context(None)
         workspace_root = ctx.get("workspace_path")
         canonical_root = ctx.get("canonical_repo_root")
-        # Alignment keeps its established repository-level meaning (#615 F1):
-        # does this namespace target the repository the process is installed in,
-        # i.e. canonical_repo_root == process_project_root. It is deliberately
-        # NOT path equality between the task workspace and the process root --
-        # the global worktree rule requires task work to live in a branches/
-        # worktree, so that comparison would classify every correctly bound
-        # author, reviewer, and merger session as unsafe.
         aligned = ctx.get("roots_aligned")
+        crr_assessment = ctx.get("canonical_root_assessment") or {}
+        resolved_slug = crr_assessment.get("resolved_slug")
     except Exception:
-        # An unresolvable binding is reported as unknown alignment, never as
-        # proof of alignment.
         aligned = None
+        resolved_slug = None
     try:
         profile_name = get_profile()["profile_name"]
     except Exception:
@@ -15149,6 +15378,7 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict:
         dirty_files=dirty_files,
         active_task_workspace=workspace_root,
         canonical_repository_root=canonical_root,
+        repository_slug=resolved_slug,
         workspace_roots_aligned=aligned,
         profile=profile_name,
         declared_mode=stable_control_runtime.declared_runtime_mode(),
@@ -15492,6 +15722,10 @@ _WORKER_REGISTRY = None
 _WORKER_IDENTITY: str | None = None
 _WORKER_GENERATION: str | None = None
 _WORKER_REGISTRATION_ATTEMPTED = False
+#: #975: the one heartbeat supervisor for this process's registration. One
+#: process registers exactly one worker identity, so there is exactly one
+#: supervisor and it is never replaced.
+_WORKER_HEARTBEAT_SUPERVISOR = None
 
 #: Env a client launcher may set to name itself and its session. Absent values
 #: are reported as unknown; they are never guessed at, because guessing is what
@@ -15499,6 +15733,7 @@ _WORKER_REGISTRATION_ATTEMPTED = False
 CLIENT_NAME_ENV = "GITEA_MCP_CLIENT"
 CLIENT_INSTANCE_ENV = "GITEA_MCP_CLIENT_INSTANCE"
 CLIENT_SESSION_ENV = "GITEA_MCP_CLIENT_SESSION"
+FLEET_RUN_ENV = "GITEA_MCP_FLEET_RUN_ID"
 
 
 def _worker_registry():
@@ -15522,13 +15757,34 @@ def _worker_registry():
 
 
 def _client_identity_hints() -> dict:
-    """What the launcher told us about itself. Unset fields stay unset."""
+    """What the launcher told us about itself (#948 / #975 / #978).
+
+    ``client_instance_id`` is established by the trusted application launcher
+    once per launch and shared by all five namespace workers. When the launcher
+    key is absent we still register under an explicit *legacy* placeholder so
+    diagnostic reads work, but the registration is marked incomplete and cannot
+    authorize multi-instance fleet mutation safety.
+    """
+    import mcp_fleet_snapshot
+
+    client_name = (os.environ.get(CLIENT_NAME_ENV) or "").strip() or None
+    raw_instance = (os.environ.get(CLIENT_INSTANCE_ENV) or "").strip() or None
+    instance = mcp_fleet_snapshot.assess_instance_identity(raw_instance)
+    # Legacy placeholder only when the launcher omitted the key — never invent a
+    # trusted ID from PID proximity.
+    if not instance["client_instance_id"]:
+        placeholder = f"legacy-pid-{os.getpid()}"
+        instance = mcp_fleet_snapshot.assess_instance_identity(
+            placeholder, source="pid_fallback"
+        )
     return {
-        "client_name": (os.environ.get(CLIENT_NAME_ENV) or "").strip() or None,
+        "client_name": client_name,
-        "client_instance_id": (os.environ.get(CLIENT_INSTANCE_ENV) or "").strip()
+        "client_instance_id": instance["client_instance_id"],
-        or f"pid-{os.getpid()}",
+        "instance_id_provenance": instance["provenance"],
+        "instance_identity_trusted": instance["trusted"],
         "session_id": (os.environ.get(CLIENT_SESSION_ENV) or "").strip()
         or f"proc-{os.getpid()}-{_process_boot_head_sha or 'nohead'}",
+        "fleet_run_id": (os.environ.get(FLEET_RUN_ENV) or "").strip() or None,
     }
 
 
@@ -15560,24 +15816,58 @@ def _active_worker_identity() -> str | None:
             identity = mcp_worker_identity.generate_worker_identity(
                 hints["client_name"], hints["session_id"]
             )
+            role = _active_role_kind_safe()
+            profile_name = (
+                (os.environ.get(gitea_config.ENV_PROFILE) or "").strip() or None
+            )
+            # Namespace is the role namespace (author/reviewer/…) when known.
+            namespace = role if role in {
+                "author", "reviewer", "merger", "controller", "reconciler"
+            } else None
+            parity = None
+            try:
+                parity = _current_master_parity()
+            except Exception:
+                parity = None
             outcome = registry.register(
                 worker_identity=identity,
                 client_name=hints["client_name"],
                 client_instance_id=hints["client_instance_id"],
                 session_id=hints["session_id"],
                 generation_id=generation,
-                role=_active_role_kind_safe(),
+                role=role,
-                profile=(os.environ.get(gitea_config.ENV_PROFILE) or "").strip() or None,
+                profile=profile_name,
+                namespace=namespace,
                 remote=(os.environ.get("GITEA_MCP_REMOTE") or "").strip() or None,
                 repository_binding=PROJECT_ROOT,
                 pid=os.getpid(),
                 transport=native.get("bound_transport"),
                 token_fingerprint=native.get("token_fingerprint"),
                 pid_alive_probe=issue_lock_store.is_process_alive,
+                fleet_run_id=hints.get("fleet_run_id"),
+                process_identity=f"pid-{os.getpid()}",
+                startup_revision=(parity or {}).get("startup_head")
+                or _process_boot_head_sha,
+                loaded_revision=(parity or {}).get("local_head")
+                or (parity or {}).get("current_head"),
+                parity_revision=(parity or {}).get("current_head"),
+                live_revision=(parity or {}).get("live_remote_head"),
+                instance_id_provenance=hints.get("instance_id_provenance"),
             )
             if outcome.get("registered"):
                 _WORKER_IDENTITY = identity
                 _WORKER_GENERATION = generation
+                # #975: registration is the only moment identity and fencing
+                # epoch are both known, so the heartbeat supervisor is started
+                # here. Without it ``last_heartbeat_at`` never left
+                # ``started_at`` and every healthy client lost ownership at the
+                # TTL.
+                _start_worker_heartbeat(
+                    identity=identity,
+                    fencing_epoch=outcome.get("fencing_epoch"),
+                    hints=hints,
+                    generation=generation,
+                )
                 return identity
             if not outcome.get("collision"):
                 return None
@@ -15588,6 +15878,78 @@ def _active_worker_identity() -> str | None:
     return None
 
 
+def _start_worker_heartbeat(
+    *,
+    identity: str,
+    fencing_epoch,
+    hints: dict,
+    generation: str | None,
+) -> None:
+    """Attach a heartbeat supervisor to the registration just created (#975).
+
+    Never raises: a supervisor that cannot start leaves the row exactly as
+    ``register()`` wrote it, which is the pre-#975 behaviour, rather than
+    failing the tool call that happened to trigger lazy registration.
+
+    Under pytest the thread is deliberately not started. Tests drive
+    ``WorkerHeartbeatSupervisor`` directly with an injected clock, so the suite
+    proves the lifecycle without leaving background sqlite writers behind.
+    """
+    global _WORKER_HEARTBEAT_SUPERVISOR
+    if _WORKER_HEARTBEAT_SUPERVISOR is not None:
+        return
+    registry = _worker_registry()
+    if registry is None or fencing_epoch is None:
+        return
+    try:
+        supervisor = mcp_worker_identity.WorkerHeartbeatSupervisor(
+            registry,
+            worker_identity=identity,
+            fencing_epoch=int(fencing_epoch),
+            session_id=hints.get("session_id"),
+            generation_id=generation,
+            client_name=hints.get("client_name"),
+            pid=os.getpid(),
+            ttl_seconds=mcp_worker_identity.DEFAULT_HEARTBEAT_TTL_SECONDS,
+            interval_seconds=os.environ.get(
+                mcp_worker_identity.HEARTBEAT_INTERVAL_ENV
+            ),
+        )
+        _WORKER_HEARTBEAT_SUPERVISOR = supervisor
+        if not mcp_daemon_guard.is_pytest_runtime():
+            supervisor.start()
+    except Exception:
+        return
+
+
+def _worker_heartbeat_status() -> dict:
+    """Read-only heartbeat observability for ``gitea_get_runtime_context`` (#975).
+
+    Reports the unsupervised case explicitly rather than omitting the key, so an
+    operator can tell "no supervisor" apart from "supervisor with no beats yet".
+    """
+    supervisor = _WORKER_HEARTBEAT_SUPERVISOR
+    if supervisor is None:
+        return {
+            "supervised": False,
+            "running": False,
+            "reasons": [
+                "no worker heartbeat supervisor is attached to this process; the "
+                "registration is not being renewed and will go stale at its TTL"
+            ],
+        }
+    try:
+        status = dict(supervisor.status())
+    except Exception as exc:
+        return {
+            "supervised": True,
+            "running": False,
+            "reasons": [f"heartbeat status unavailable: {type(exc).__name__}: {exc}"],
+        }
+    status.setdefault("reasons", [])
+    return status
+
+
 def _active_role_kind_safe() -> str | None:
     """Best-effort role for the registry record; never raises into a tool call.
 
@@ -19056,6 +19418,172 @@ def gitea_validate_review_final_report(
     )
 
 
+@mcp.tool()
+def gitea_snapshot_instance_fleet(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    expected_manifest: list | dict | str | None = None,
+    include_historical: bool = True,
+    canonical_repository: str | None = None,
+    expected_live_revision: str | None = None,
+) -> dict:
+    """Read-only: authoritative instance-level fleet identity and health snapshot (#978).
+
+    Controller and reconciler only. Returns a point-in-time snapshot of every
+    registered namespace worker with instance attribution, heartbeat freshness,
+    and structured classification (expected/missing/unmanifested, duplicate
+    namespace workers, identity collisions, foreign repository, old revision,
+    historical dead rows). Sharing only a client type is never a duplicate.
+
+    Does not grant any mutation capability. Does not scan process tables or
+    open foreign databases for production evidence — the worker registry is
+    the sole authority.
+
+    Args:
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Optional host override.
+        org: Optional org override (audit context only).
+        repo: Optional repo override (audit context only).
+        expected_manifest: Optional approved fleet enrollment list
+            (list of dicts with client_instance_id / client_type / fleet_run_id).
+            When supplied, missing and unmanifested instances are classified.
+        include_historical: Include released/superseded rows (default True);
+            historical rows never make the live fleet unsafe by themselves.
+        canonical_repository: Expected repository binding path/slug for
+            foreign-repository classification.
+        expected_live_revision: When set, workers whose recorded revisions
+            differ are classified as old-revision.
+    """
+    import json as _json
+
+    import mcp_fleet_snapshot
+
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "read_only": True,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    profile = get_profile()
+    role = _profile_role_kind(profile)
+    if role not in {"controller", "reconciler"}:
+        return {
+            "success": False,
+            "read_only": True,
+            "allowed": False,
+            "denied_role": role,
+            "required_roles": ["controller", "reconciler"],
+            "mutation_performed": False,
+            "reasons": [
+                f"gitea_snapshot_instance_fleet is restricted to controller and "
+                f"reconciler roles; active role_kind is {role!r}. Author, "
+                "reviewer, and merger profiles keep gitea.read for diagnosis "
+                "elsewhere but do not receive this fleet surface, and no "
+                "unrelated mutation permission is granted."
+            ],
+            "exact_next_action": (
+                "Re-run from a prgs-controller or prgs-reconciler namespace."
+            ),
+        }
+
+    manifest: list | None = None
+    if expected_manifest is not None:
+        raw = expected_manifest
+        if isinstance(raw, str):
+            try:
+                raw = _json.loads(raw)
+            except Exception as exc:
+                return {
+                    "success": False,
+                    "read_only": True,
+                    "reasons": [
+                        f"expected_manifest is not valid JSON: {type(exc).__name__}"
+                    ],
+                }
+        if isinstance(raw, dict):
+            # Accept {"instances": [...]} or a single instance object.
+            if "instances" in raw and isinstance(raw["instances"], list):
+                raw = raw["instances"]
+            else:
+                raw = [raw]
+        if not isinstance(raw, list):
+            return {
+                "success": False,
+                "read_only": True,
+                "reasons": ["expected_manifest must be a list of instance objects"],
+            }
+        manifest = raw
+
+    registry = _worker_registry()
+    if registry is None:
+        return {
+            "success": False,
+            "read_only": True,
+            "reasons": [
+                "worker registry is unavailable; cannot produce an authoritative "
+                "fleet snapshot (fail closed)"
+            ],
+            "exact_next_action": (
+                "Ensure GITEA_WORKER_REGISTRY_DB is writable and re-run after "
+                "workers have registered."
+            ),
+        }
+
+    try:
+        if include_historical:
+            records = registry.list_workers(status=None)
+        else:
+            records = registry.list_workers(status=mcp_worker_identity.STATUS_ACTIVE)
+    except Exception as exc:
+        return {
+            "success": False,
+            "read_only": True,
+            "reasons": [
+                f"failed to read worker registry: {type(exc).__name__}: {_redact(str(exc))}"
+            ],
+        }
+
+    parity = None
+    try:
+        parity = _current_master_parity()
+    except Exception:
+        parity = None
+    live_rev = expected_live_revision or (parity or {}).get("live_remote_head")
+    canon = canonical_repository or PROJECT_ROOT
+
+    snapshot = mcp_fleet_snapshot.snapshot_instance_fleet(
+        records,
+        expected_manifest=manifest,
+        pid_alive_probe=issue_lock_store.is_process_alive,
+        canonical_repository=canon,
+        expected_live_revision=live_rev,
+    )
+    snapshot["role_kind"] = role
+    snapshot["profile"] = profile.get("profile_name")
+    snapshot["remote"] = _effective_remote(remote)
+    snapshot["repository"] = {
+        "org": org,
+        "repo": repo,
+        "canonical_repository": canon,
+    }
+    snapshot["mutation_performed"] = False
+    snapshot["permission_scope"] = {
+        "read_only": True,
+        "granted_operations": ["gitea.read"],
+        "denied_unrelated_mutations": True,
+        "note": (
+            "This capability is strictly observational. It does not authorize "
+            "branch, issue, PR, review, merge, or restart mutations."
+        ),
+    }
+    return snapshot
+
+
 @mcp.tool()
 def gitea_get_runtime_context(
     remote: str = "dadeschools",
@@ -19229,6 +19757,11 @@ def gitea_get_runtime_context(
         "fencing_epoch": provenance_assessment["fencing_epoch"],
         "conflicting_live_sessions": provenance_assessment["conflicting_live_sessions"],
         "provenance_assessment": provenance_assessment,
+        # #975: whether this registration is actually being renewed. Read-only,
+        # and it grants nothing — ownership still comes from the attachment
+        # record above. It exists so "my heartbeat stopped" is diagnosable
+        # before the TTL turns it into session_attachment_missing.
+        "worker_heartbeat": _worker_heartbeat_status(),
         "unconsumed_gitea_env": unconsumed_env,
         "preflight_ready": preflight["preflight_ready"],
         "preflight_block_reasons": preflight["preflight_block_reasons"],
@@ -21737,6 +22270,17 @@ def gitea_route_task_session(
 # self-recovery was removed from the read-only path (was _trigger_mcp_auto_restart).
 # Recovery is owned exclusively by the IDE/client reconnect path.
 
+#: Every reason prefix ``_check_mcp_runtimes_diagnostics`` can emit. Callers
+#: raise its reasons as one RuntimeError, and #975 found that the preflight
+#: re-raise recognised only ``stale-runtime:``, silently dropping
+#: ``unsupported-env:`` while the capability resolver still failed on it. This
+#: lives beside the producer so a newly added reason family cannot be forgotten
+#: by a distant re-raise predicate again.
+RUNTIME_DIAGNOSTIC_HARD_PREFIXES: tuple[str, ...] = (
+    "stale-runtime:",
+    "unsupported-env:",
+)
+
 
 def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
     """Read-only: report missing or stale MCP runtimes (no config or process mutation).
@@ -24432,18 +24976,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 +24993,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 +25008,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 +25034,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 +25043,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 +25068,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,

mcp_fleet_snapshot.py

@@ -0,0 +1,847 @@
+"""Instance-level fleet identity and health snapshots (#978).
+
+#948 established per-worker ownership; #975 made heartbeats keep those rows
+live. Neither surface could enumerate the fleet at *instance* granularity:
+which application launch owns which five namespace workers, whether two
+Codex launches are distinct, or whether a live collision is real rather than
+a shared client type.
+
+This module is pure. Callers supply registry rows (and optional enrichments);
+nothing here opens SQLite, scans process tables, or mutates state. Production
+evidence for the fleet gate is the snapshot returned by the sanctioned
+controller/reconciler tool that wraps this assessor.
+
+Identity hierarchy (highest → lowest):
+
+* ``client_type`` — application family (``codex``, ``claude_code``, …)
+* ``client_instance_id`` — one running application launch (trusted launcher)
+* ``fleet_run_id`` — operator-approved enrollment / canary cohort
+* ``worker_id`` / ``worker_identity`` — one namespace worker process
+* ``namespace`` — author | reviewer | merger | controller | reconciler
+
+Multiple simultaneous instances of the same ``client_type`` are first-class.
+Sharing only a profile or client type is never a duplicate.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import secrets
+from collections import defaultdict
+from datetime import datetime, timezone
+from typing import Any, Callable, Iterable, Mapping
+
+import mcp_worker_identity as mwi
+
+# --- Classification labels ------------------------------------------------
+
+CLASS_EXPECTED = "expected_enrolled"
+CLASS_MISSING = "missing_expected"
+CLASS_UNMANIFESTED = "unmanifested"
+CLASS_DUPLICATE_NAMESPACE = "duplicate_namespace_worker"
+CLASS_INSTANCE_ID_COLLISION = "instance_id_collision"
+CLASS_WORKER_ID_COLLISION = "worker_identity_collision"
+CLASS_SESSION_COLLISION = "session_identity_collision"
+CLASS_GENERATION_COLLISION = "generation_identity_collision"
+CLASS_PROCESS_COLLISION = "process_identity_collision"
+CLASS_PID_COLLISION = "pid_collision"
+CLASS_OWNERSHIP_COLLISION = "ownership_fencing_collision"
+CLASS_ORPHANED = "orphaned_unowned"
+CLASS_UNKNOWN_CLIENT = "unknown_client"
+CLASS_FOREIGN_REPOSITORY = "foreign_repository"
+CLASS_OLD_REVISION = "old_revision"
+CLASS_STALE_WORKER = "stale_orphaned_worker"
+CLASS_LEGACY_INCOMPLETE = "legacy_incomplete_identity"
+CLASS_HISTORICAL = "historical_dead"
+CLASS_HEALTHY = "healthy"
+
+#: Active blockers that make the live fleet unsafe for mutation-gated work.
+ACTIVE_BLOCKER_CLASSES = frozenset(
+    {
+        CLASS_MISSING,
+        CLASS_UNMANIFESTED,
+        CLASS_DUPLICATE_NAMESPACE,
+        CLASS_INSTANCE_ID_COLLISION,
+        CLASS_WORKER_ID_COLLISION,
+        CLASS_SESSION_COLLISION,
+        CLASS_GENERATION_COLLISION,
+        CLASS_PROCESS_COLLISION,
+        CLASS_PID_COLLISION,
+        CLASS_OWNERSHIP_COLLISION,
+        CLASS_ORPHANED,
+        CLASS_UNKNOWN_CLIENT,
+        CLASS_FOREIGN_REPOSITORY,
+        CLASS_OLD_REVISION,
+        CLASS_STALE_WORKER,
+        CLASS_LEGACY_INCOMPLETE,
+    }
+)
+
+SANCTIONED_NAMESPACES = frozenset(
+    {"author", "reviewer", "merger", "controller", "reconciler"}
+)
+
+INSTANCE_ID_PROVENANCE_TRUSTED = "trusted_launcher"
+INSTANCE_ID_PROVENANCE_LEGACY = "legacy_incomplete"
+INSTANCE_ID_PROVENANCE_MISSING = "missing"
+
+CLIENT_INSTANCE_ENV = "GITEA_MCP_CLIENT_INSTANCE"
+FLEET_RUN_ENV = "GITEA_MCP_FLEET_RUN_ID"
+PROCESS_IDENTITY_ENV = "GITEA_MCP_PROCESS_IDENTITY"
+
+_LEGACY_INSTANCE_PREFIXES = ("pid-", "proc-", "legacy-")
+
+
+def _utc_now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _ts(value: datetime) -> str:
+    return value.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def generate_client_instance_id(
+    client_type: str | None,
+    *,
+    launch_nonce: str | None = None,
+    now: datetime | None = None,
+) -> str:
+    """Mint a distinct instance ID for one application launch (#978).
+
+    The trusted launcher (or host that starts all five namespace workers)
+    generates this once per launch and injects it as ``GITEA_MCP_CLIENT_INSTANCE``
+    into every worker environment. Workers never invent their own instance ID
+    from PID proximity or timestamps.
+    """
+    client = mwi.normalize_client_name(client_type)
+    stamp = (now or _utc_now()).astimezone(timezone.utc).strftime(
+        mwi.IDENTITY_TIMESTAMP_FORMAT
+    )
+    nonce = launch_nonce if launch_nonce is not None else secrets.token_hex(16)
+    digest = hashlib.sha256(
+        f"{client}\x1f{stamp}\x1f{nonce}".encode("utf-8")
+    ).hexdigest()[:12]
+    return f"inst-{client}-{stamp}-{digest}"
+
+
+def assess_instance_identity(
+    raw_instance_id: str | None,
+    *,
+    source: str | None = None,
+) -> dict[str, Any]:
+    """Classify whether a client_instance_id is trusted enough for mutation.
+
+    Trusted instance IDs are non-empty and not the pre-#978 PID/proc fallbacks.
+    Incomplete identities remain visible for diagnosis but cannot authorize
+    unsafe mutation.
+    """
+    text = (raw_instance_id or "").strip()
+    if not text:
+        return {
+            "client_instance_id": None,
+            "complete": False,
+            "trusted": False,
+            "provenance": INSTANCE_ID_PROVENANCE_MISSING,
+            "reasons": [
+                "client_instance_id is missing; the trusted launcher must set "
+                f"{CLIENT_INSTANCE_ENV} once per application launch"
+            ],
+        }
+    lowered = text.lower()
+    if lowered.startswith(_LEGACY_INSTANCE_PREFIXES) or source == "pid_fallback":
+        return {
+            "client_instance_id": text,
+            "complete": False,
+            "trusted": False,
+            "provenance": INSTANCE_ID_PROVENANCE_LEGACY,
+            "reasons": [
+                f"client_instance_id {text!r} is a legacy PID/process fallback, "
+                "not a trusted launcher-issued instance identity"
+            ],
+        }
+    return {
+        "client_instance_id": text,
+        "complete": True,
+        "trusted": True,
+        "provenance": INSTANCE_ID_PROVENANCE_TRUSTED,
+        "reasons": [],
+    }
+
+
+def resolve_client_instance_from_env(
+    env: Mapping[str, str] | None = None,
+    *,
+    pid: int | None = None,
+) -> dict[str, Any]:
+    """Resolve instance identity from launcher env without inventing one.
+
+    When the trusted key is absent, return incomplete evidence rather than a
+    silent ``pid-<n>`` identity. Callers that still need a non-empty registry
+    key may choose a legacy placeholder deliberately; they must not treat it as
+    trusted.
+    """
+    source = dict(env or {})
+    raw = (source.get(CLIENT_INSTANCE_ENV) or "").strip()
+    assessment = assess_instance_identity(raw or None)
+    assessment["fleet_run_id"] = (source.get(FLEET_RUN_ENV) or "").strip() or None
+    assessment["process_identity"] = (
+        (source.get(PROCESS_IDENTITY_ENV) or "").strip()
+        or (f"pid-{pid}" if pid is not None else None)
+    )
+    return assessment
+
+
+def _public_worker(record: Mapping[str, Any]) -> dict[str, Any]:
+    return {
+        "worker_id": record.get("worker_identity") or record.get("worker_id"),
+        "worker_identity": record.get("worker_identity") or record.get("worker_id"),
+        "client_type": mwi.normalize_client_name(
+            record.get("client_name") or record.get("client_type")
+        ),
+        "client_instance_id": record.get("client_instance_id"),
+        "fleet_run_id": record.get("fleet_run_id"),
+        "namespace": record.get("namespace"),
+        "profile": record.get("profile"),
+        "declared_role": record.get("role") or record.get("declared_role"),
+        "authenticated_account": record.get("authenticated_account"),
+        "session_id": record.get("session_id"),
+        "generation_id": record.get("generation_id"),
+        "process_identity": record.get("process_identity")
+        or (
+            f"pid-{record['pid']}"
+            if record.get("pid") is not None
+            else None
+        ),
+        "pid": record.get("pid"),
+        "repository_binding": record.get("repository_binding"),
+        "remote": record.get("remote"),
+        "startup_revision": record.get("startup_revision"),
+        "loaded_revision": record.get("loaded_revision"),
+        "parity_revision": record.get("parity_revision"),
+        "live_revision": record.get("live_revision"),
+        "runtime_provenance": record.get("runtime_provenance")
+        or record.get("transport"),
+        "transport": record.get("transport"),
+        "started_at": record.get("started_at"),
+        "last_heartbeat_at": record.get("last_heartbeat_at"),
+        "heartbeat_ttl_seconds": record.get("heartbeat_ttl_seconds"),
+        "fencing_epoch": record.get("fencing_epoch"),
+        "status": record.get("status"),
+        "instance_id_provenance": record.get("instance_id_provenance"),
+    }
+
+
+def _liveness(
+    record: Mapping[str, Any],
+    *,
+    now: datetime | None,
+    pid_alive_probe: Callable[[int | None], bool | None] | None,
+) -> dict[str, Any]:
+    pid_alive = None
+    if pid_alive_probe is not None and record.get("pid") is not None:
+        try:
+            pid_alive = pid_alive_probe(record.get("pid"))
+        except Exception:
+            pid_alive = None
+    return mwi.WorkerRegistry.is_live(record, now=now, pid_alive=pid_alive)
+
+
+def _consistency_token(rows: Iterable[Mapping[str, Any]], snapshot_at: str) -> str:
+    material = [snapshot_at]
+    for row in sorted(
+        rows,
+        key=lambda r: (
+            str(r.get("worker_identity") or ""),
+            str(r.get("last_heartbeat_at") or ""),
+            str(r.get("fencing_epoch") or ""),
+        ),
+    ):
+        material.append(
+            "|".join(
+                [
+                    str(row.get("worker_identity") or ""),
+                    str(row.get("client_instance_id") or ""),
+                    str(row.get("status") or ""),
+                    str(row.get("last_heartbeat_at") or ""),
+                    str(row.get("fencing_epoch") or ""),
+                    str(row.get("generation_id") or ""),
+                ]
+            )
+        )
+    digest = hashlib.sha256("\n".join(material).encode("utf-8")).hexdigest()[:16]
+    return f"fleetrev-{digest}"
+
+
+def build_worker_snapshot_row(
+    record: Mapping[str, Any],
+    *,
+    now: datetime | None = None,
+    pid_alive_probe: Callable[[int | None], bool | None] | None = None,
+    canonical_repository: str | None = None,
+    expected_live_revision: str | None = None,
+    heartbeat_supervised: bool | None = None,
+) -> dict[str, Any]:
+    """One point-in-time worker row for the fleet snapshot."""
+    stamp = now or _utc_now()
+    base = _public_worker(record)
+    identity = assess_instance_identity(
+        base.get("client_instance_id"),
+        source=record.get("instance_id_source"),
+    )
+    liveness = _liveness(record, now=stamp, pid_alive_probe=pid_alive_probe)
+    is_historical = str(record.get("status") or "") != mwi.STATUS_ACTIVE
+    live = bool(liveness.get("live")) and not is_historical
+
+    repo = (base.get("repository_binding") or "").strip() or None
+    foreign_repo = bool(
+        canonical_repository
+        and repo
+        and repo.rstrip("/") != str(canonical_repository).rstrip("/")
+    )
+    old_revision = False
+    if expected_live_revision:
+        for key in ("startup_revision", "loaded_revision", "parity_revision", "live_revision"):
+            rev = (base.get(key) or "").strip()
+            if rev and rev != expected_live_revision:
+                old_revision = True
+                break
+
+    ownership_state = "historical" if is_historical else (
+        "live" if live else "stale"
+    )
+    if live and not identity["trusted"]:
+        ownership_state = "live_untrusted_identity"
+    if live and not base.get("session_id"):
+        ownership_state = "orphaned"
+
+    mutation_safe = bool(
+        live
+        and identity["trusted"]
+        and not foreign_repo
+        and not old_revision
+        and ownership_state == "live"
+        and base.get("client_type") != mwi.UNKNOWN_CLIENT
+    )
+
+    restart_required = bool(
+        old_revision
+        or (live and not liveness.get("heartbeat_fresh", True))
+    )
+
+    return {
+        **base,
+        "instance_identity": identity,
+        "client_instance_id": identity["client_instance_id"] or base.get("client_instance_id"),
+        "instance_id_provenance": identity["provenance"],
+        "instance_identity_trusted": identity["trusted"],
+        "live": live,
+        "historical": is_historical,
+        "liveness": liveness,
+        "heartbeat": {
+            "registered": bool(base.get("last_heartbeat_at")),
+            "supervised": heartbeat_supervised,
+            "age_seconds": liveness.get("heartbeat_age_seconds"),
+            "ttl_seconds": liveness.get("heartbeat_ttl_seconds"),
+            "fresh": liveness.get("heartbeat_fresh"),
+            "last_heartbeat_at": base.get("last_heartbeat_at"),
+        },
+        "fencing": {
+            "fencing_epoch": base.get("fencing_epoch"),
+            "generation_id": base.get("generation_id"),
+        },
+        "ownership_state": ownership_state,
+        "foreign_repository": foreign_repo,
+        "old_revision": old_revision,
+        "stale": not live and not is_historical,
+        "restart_required": restart_required,
+        "mutation_safe": mutation_safe,
+        "conflicting_live_sessions": [],
+    }
+
+
+def _collision_groups(
+    live_rows: list[dict[str, Any]],
+    key_fn,
+) -> dict[str, list[dict[str, Any]]]:
+    groups: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    for row in live_rows:
+        key = key_fn(row)
+        if key is None or key == "" or key == "None":
+            continue
+        groups[str(key)].append(row)
+    return {k: v for k, v in groups.items() if len(v) > 1}
+
+
+def snapshot_instance_fleet(
+    records: Iterable[Mapping[str, Any]],
+    *,
+    expected_manifest: list[Mapping[str, Any]] | None = None,
+    now: datetime | None = None,
+    pid_alive_probe: Callable[[int | None], bool | None] | None = None,
+    canonical_repository: str | None = None,
+    expected_live_revision: str | None = None,
+    registry_revision: str | None = None,
+    known_client_types: Iterable[str] | None = None,
+) -> dict[str, Any]:
+    """Authoritative point-in-time fleet snapshot with classification (#978).
+
+    Historical dead rows are reported separately and never automatically make
+    the live fleet unsafe.
+    """
+    stamp = now or _utc_now()
+    snapshot_at = _ts(stamp)
+    known = {
+        mwi.normalize_client_name(c)
+        for c in (known_client_types or mwi.CLIENT_ALIASES.values())
+    }
+    known.discard(mwi.UNKNOWN_CLIENT)
+
+    all_rows: list[dict[str, Any]] = []
+    for record in records:
+        all_rows.append(
+            build_worker_snapshot_row(
+                record,
+                now=stamp,
+                pid_alive_probe=pid_alive_probe,
+                canonical_repository=canonical_repository,
+                expected_live_revision=expected_live_revision,
+            )
+        )
+
+    live_rows = [r for r in all_rows if r["live"]]
+    historical_rows = [r for r in all_rows if r["historical"]]
+    stale_rows = [r for r in all_rows if r["stale"]]
+
+    # --- identity collisions among live workers ---
+    findings: list[dict[str, Any]] = []
+
+    def _finding(
+        classification: str,
+        *,
+        severity: str,
+        workers: list[dict[str, Any]] | None = None,
+        instance_ids: list[str] | None = None,
+        detail: str,
+        active_blocker: bool,
+    ) -> None:
+        findings.append(
+            {
+                "classification": classification,
+                "severity": severity,
+                "active_blocker": active_blocker,
+                "detail": detail,
+                "client_instance_ids": instance_ids or sorted(
+                    {
+                        str(w.get("client_instance_id"))
+                        for w in (workers or [])
+                        if w.get("client_instance_id")
+                    }
+                ),
+                "worker_identities": [
+                    w.get("worker_identity") for w in (workers or [])
+                ],
+            }
+        )
+
+    # Duplicate worker identity (should not happen with PK, still detect)
+    for wid, group in _collision_groups(
+        live_rows, lambda r: r.get("worker_identity")
+    ).items():
+        _finding(
+            CLASS_WORKER_ID_COLLISION,
+            severity="blocker",
+            workers=group,
+            detail=f"worker identity {wid!r} is claimed by {len(group)} live workers",
+            active_blocker=True,
+        )
+
+    # Reused session identity across live workers
+    for sid, group in _collision_groups(live_rows, lambda r: r.get("session_id")).items():
+        # Same session may appear once; collision only when multiple workers share it
+        # across different worker identities (always true for group size > 1).
+        _finding(
+            CLASS_SESSION_COLLISION,
+            severity="blocker",
+            workers=group,
+            detail=f"session identity {sid!r} is reused by {len(group)} live workers",
+            active_blocker=True,
+        )
+
+    # Generation claimed by multiple live sessions/workers is a conflict when
+    # the workers are not the five sanctioned namespaces of one instance.
+    for gen, group in _collision_groups(
+        live_rows, lambda r: r.get("generation_id")
+    ).items():
+        namespaces = {g.get("namespace") for g in group if g.get("namespace")}
+        instance_ids = {g.get("client_instance_id") for g in group}
+        # Multiple workers under one generation is only valid if they share one
+        # instance and distinct namespaces. Same generation + same namespace = bad.
+        by_ns: dict[str, list] = defaultdict(list)
+        for g in group:
+            by_ns[str(g.get("namespace") or "")].append(g)
+        ns_dups = {ns: rows for ns, rows in by_ns.items() if ns and len(rows) > 1}
+        if ns_dups or len(instance_ids) > 1:
+            _finding(
+                CLASS_GENERATION_COLLISION,
+                severity="blocker",
+                workers=group,
+                detail=(
+                    f"generation {gen!r} is contested across namespaces/instances "
+                    f"(namespaces={sorted(namespaces)}, "
+                    f"instances={sorted(str(i) for i in instance_ids if i)})"
+                ),
+                active_blocker=True,
+            )
+
+    # Process identity / PID collisions across distinct workers
+    for proc, group in _collision_groups(
+        live_rows, lambda r: r.get("process_identity")
+    ).items():
+        if len({r.get("worker_identity") for r in group}) > 1:
+            _finding(
+                CLASS_PROCESS_COLLISION,
+                severity="blocker",
+                workers=group,
+                detail=f"process identity {proc!r} is shared by distinct live workers",
+                active_blocker=True,
+            )
+
+    for pid, group in _collision_groups(live_rows, lambda r: r.get("pid")).items():
+        if len({r.get("worker_identity") for r in group}) > 1:
+            _finding(
+                CLASS_PID_COLLISION,
+                severity="blocker",
+                workers=group,
+                detail=f"PID {pid} is shared by distinct live workers",
+                active_blocker=True,
+            )
+
+    # Fencing/ownership: same fencing epoch on different workers of different instances
+    for epoch, group in _collision_groups(
+        live_rows,
+        lambda r: (
+            f"{r.get('generation_id')}:{r.get('fencing_epoch')}"
+            if r.get("generation_id") is not None and r.get("fencing_epoch") is not None
+            else None
+        ),
+    ).items():
+        if len({r.get("client_instance_id") for r in group}) > 1:
+            _finding(
+                CLASS_OWNERSHIP_COLLISION,
+                severity="blocker",
+                workers=group,
+                detail=(
+                    f"fencing token {epoch!r} spans more than one client_instance_id"
+                ),
+                active_blocker=True,
+            )
+
+    # Per-instance grouping
+    by_instance: dict[str, list[dict[str, Any]]] = defaultdict(list)
+    unkeyed_live: list[dict[str, Any]] = []
+    for row in live_rows:
+        iid = row.get("client_instance_id")
+        if not iid:
+            unkeyed_live.append(row)
+            continue
+        by_instance[str(iid)].append(row)
+
+    instances: list[dict[str, Any]] = []
+    for iid, workers in sorted(by_instance.items()):
+        client_types = sorted({w.get("client_type") for w in workers if w.get("client_type")})
+        trusted = all(w.get("instance_identity_trusted") for w in workers)
+        namespaces = [w.get("namespace") for w in workers]
+        ns_counts: dict[str, int] = defaultdict(int)
+        for ns in namespaces:
+            if ns:
+                ns_counts[str(ns)] += 1
+        dup_ns = sorted(ns for ns, n in ns_counts.items() if n > 1)
+        if dup_ns:
+            _finding(
+                CLASS_DUPLICATE_NAMESPACE,
+                severity="blocker",
+                workers=[w for w in workers if w.get("namespace") in dup_ns],
+                instance_ids=[iid],
+                detail=(
+                    f"instance {iid!r} has more than one live worker for "
+                    f"namespace(s) {dup_ns}"
+                ),
+                active_blocker=True,
+            )
+
+        # Live reuse of one instance ID with incompatible client types
+        if len(client_types) > 1:
+            _finding(
+                CLASS_INSTANCE_ID_COLLISION,
+                severity="blocker",
+                workers=workers,
+                instance_ids=[iid],
+                detail=(
+                    f"client_instance_id {iid!r} is live under multiple client "
+                    f"types {client_types}"
+                ),
+                active_blocker=True,
+            )
+
+        if not trusted:
+            _finding(
+                CLASS_LEGACY_INCOMPLETE,
+                severity="blocker",
+                workers=workers,
+                instance_ids=[iid],
+                detail=(
+                    f"instance {iid!r} lacks trusted launcher-issued instance "
+                    "identity; diagnostic reads remain available"
+                ),
+                active_blocker=True,
+            )
+
+        unknown = [w for w in workers if w.get("client_type") == mwi.UNKNOWN_CLIENT]
+        if unknown:
+            _finding(
+                CLASS_UNKNOWN_CLIENT,
+                severity="blocker",
+                workers=unknown,
+                instance_ids=[iid],
+                detail=f"instance {iid!r} has worker(s) with unknown client_type",
+                active_blocker=True,
+            )
+
+        foreign = [w for w in workers if w.get("foreign_repository")]
+        if foreign:
+            _finding(
+                CLASS_FOREIGN_REPOSITORY,
+                severity="blocker",
+                workers=foreign,
+                instance_ids=[iid],
+                detail=f"instance {iid!r} has foreign-repository workers",
+                active_blocker=True,
+            )
+
+        old = [w for w in workers if w.get("old_revision")]
+        if old:
+            _finding(
+                CLASS_OLD_REVISION,
+                severity="blocker",
+                workers=old,
+                instance_ids=[iid],
+                detail=f"instance {iid!r} has old-revision workers",
+                active_blocker=True,
+            )
+
+        orphans = [w for w in workers if w.get("ownership_state") == "orphaned"]
+        if orphans:
+            _finding(
+                CLASS_ORPHANED,
+                severity="blocker",
+                workers=orphans,
+                instance_ids=[iid],
+                detail=f"instance {iid!r} has orphaned/unowned workers",
+                active_blocker=True,
+            )
+
+        instances.append(
+            {
+                "client_instance_id": iid,
+                "client_types": client_types,
+                "client_type": client_types[0] if len(client_types) == 1 else None,
+                "fleet_run_ids": sorted(
+                    {w.get("fleet_run_id") for w in workers if w.get("fleet_run_id")}
+                ),
+                "worker_count": len(workers),
+                "namespaces": sorted({n for n in namespaces if n}),
+                "namespace_counts": dict(ns_counts),
+                "duplicate_namespaces": dup_ns,
+                "trusted_instance_identity": trusted,
+                "workers": workers,
+                "mutation_safe": all(w.get("mutation_safe") for w in workers)
+                and not dup_ns
+                and trusted,
+            }
+        )
+
+    for row in unkeyed_live:
+        _finding(
+            CLASS_LEGACY_INCOMPLETE,
+            severity="blocker",
+            workers=[row],
+            detail="live worker has no client_instance_id",
+            active_blocker=True,
+        )
+        if row.get("client_type") == mwi.UNKNOWN_CLIENT:
+            _finding(
+                CLASS_UNKNOWN_CLIENT,
+                severity="blocker",
+                workers=[row],
+                detail="live worker has unknown client_type and no instance id",
+                active_blocker=True,
+            )
+
+    for row in stale_rows:
+        _finding(
+            CLASS_STALE_WORKER,
+            severity="warning",
+            workers=[row],
+            detail=(
+                f"worker {row.get('worker_identity')!r} is active in the registry "
+                "but not live (stale heartbeat or dead pid)"
+            ),
+            active_blocker=True,
+        )
+
+    for row in historical_rows:
+        _finding(
+            CLASS_HISTORICAL,
+            severity="info",
+            workers=[row],
+            detail=(
+                f"historical registration {row.get('worker_identity')!r} "
+                f"(status={row.get('status')!r}) is not an active blocker"
+            ),
+            active_blocker=False,
+        )
+
+    # Manifest comparison
+    expected = list(expected_manifest or [])
+    expected_ids = {
+        str(item.get("client_instance_id")).strip()
+        for item in expected
+        if (item.get("client_instance_id") or "").strip()
+    }
+    live_ids = set(by_instance.keys())
+    missing_ids = sorted(expected_ids - live_ids)
+    unmanifested_ids = sorted(live_ids - expected_ids) if expected_ids else []
+
+    for iid in missing_ids:
+        _finding(
+            CLASS_MISSING,
+            severity="blocker",
+            instance_ids=[iid],
+            detail=f"expected enrolled instance {iid!r} is missing from the live fleet",
+            active_blocker=True,
+        )
+    for iid in unmanifested_ids:
+        _finding(
+            CLASS_UNMANIFESTED,
+            severity="blocker",
+            instance_ids=[iid],
+            workers=by_instance.get(iid, []),
+            detail=(
+                f"live instance {iid!r} is not on the approved fleet manifest "
+                "(unmanifested)"
+            ),
+            active_blocker=True,
+        )
+
+    # Same client_type multi-instance is healthy when each has distinct instance IDs
+    by_type: dict[str, list[str]] = defaultdict(list)
+    for inst in instances:
+        for ct in inst.get("client_types") or []:
+            by_type[str(ct)].append(inst["client_instance_id"])
+    multi_instance_same_type = {
+        ct: ids for ct, ids in by_type.items() if len(ids) > 1
+    }
+
+    active_blockers = [f for f in findings if f.get("active_blocker")]
+    historical_only = [f for f in findings if f.get("classification") == CLASS_HISTORICAL]
+    live_safe = not active_blockers
+
+    consistency = registry_revision or _consistency_token(all_rows, snapshot_at)
+
+    return {
+        "success": True,
+        "read_only": True,
+        "snapshot_at": snapshot_at,
+        "consistency_token": consistency,
+        "registry_revision": consistency,
+        "live_worker_count": len(live_rows),
+        "historical_worker_count": len(historical_rows),
+        "stale_worker_count": len(stale_rows),
+        "instance_count": len(instances),
+        "workers": all_rows,
+        "live_workers": live_rows,
+        "historical_workers": historical_rows,
+        "stale_workers": stale_rows,
+        "instances": instances,
+        "multi_instance_same_client_type": multi_instance_same_type,
+        "same_client_type_not_duplicate": True,
+        "expected_manifest": [
+            {
+                "client_instance_id": item.get("client_instance_id"),
+                "client_type": item.get("client_type"),
+                "fleet_run_id": item.get("fleet_run_id"),
+                "namespaces": item.get("namespaces"),
+            }
+            for item in expected
+        ],
+        "missing_expected_instance_ids": missing_ids,
+        "unmanifested_instance_ids": unmanifested_ids,
+        "findings": findings,
+        "active_blockers": active_blockers,
+        "historical_findings": historical_only,
+        "live_fleet_safe": live_safe,
+        "mutation_safe": live_safe and all(
+            inst.get("mutation_safe") for inst in instances
+        )
+        if instances
+        else live_safe,
+        "classification_model": {
+            "exactly_one_process_per_profile": False,
+            "exactly_one_instance_per_client_type": False,
+            "multiple_instances_per_client_type": True,
+            "duplicate_requires": [
+                "live client_instance_id collision",
+                "duplicate namespace worker within one instance",
+                "reused worker/session/generation/process/pid/fencing identity",
+                "cross-instance ownership collision",
+                "unmanifested instance when a manifest is required",
+            ],
+        },
+        "reasons": [f["detail"] for f in active_blockers],
+    }
+
+
+def compare_snapshot_heartbeats(
+    earlier: Mapping[str, Any],
+    later: Mapping[str, Any],
+) -> dict[str, Any]:
+    """Prove heartbeat continuity and stable ownership across two snapshots."""
+    earlier_live = {
+        w.get("worker_identity"): w for w in earlier.get("live_workers") or []
+    }
+    later_live = {
+        w.get("worker_identity"): w for w in later.get("live_workers") or []
+    }
+    shared = sorted(set(earlier_live) & set(later_live))
+    continuity: list[dict[str, Any]] = []
+    stable_ownership = True
+    for wid in shared:
+        a = earlier_live[wid]
+        b = later_live[wid]
+        same_instance = a.get("client_instance_id") == b.get("client_instance_id")
+        same_session = a.get("session_id") == b.get("session_id")
+        same_generation = a.get("generation_id") == b.get("generation_id")
+        hb_advanced_or_equal = True
+        if a.get("last_heartbeat_at") and b.get("last_heartbeat_at"):
+            hb_advanced_or_equal = b["last_heartbeat_at"] >= a["last_heartbeat_at"]
+        if not (same_instance and same_session and same_generation):
+            stable_ownership = False
+        continuity.append(
+            {
+                "worker_identity": wid,
+                "same_client_instance_id": same_instance,
+                "same_session_id": same_session,
+                "same_generation_id": same_generation,
+                "heartbeat_non_decreasing": hb_advanced_or_equal,
+                "earlier_heartbeat": a.get("last_heartbeat_at"),
+                "later_heartbeat": b.get("last_heartbeat_at"),
+            }
+        )
+    return {
+        "shared_live_workers": shared,
+        "continuity": continuity,
+        "stable_ownership": stable_ownership
+        and all(c["heartbeat_non_decreasing"] for c in continuity),
+        "dropped_workers": sorted(set(earlier_live) - set(later_live)),
+        "new_workers": sorted(set(later_live) - set(earlier_live)),
+    }

mcp_worker_identity.py

@@ -29,6 +29,7 @@ implements:
 
 from __future__ import annotations
 
+import atexit
 import hashlib
 import os
 import re
@@ -123,6 +124,103 @@ DEFAULT_REGISTRY_PATH = os.path.expanduser(
 #: worker within a single operator coffee break.
 DEFAULT_HEARTBEAT_TTL_SECONDS = 900.0
 
+#: Optional operator override for the beat interval, in seconds. Clamped by
+#: :func:`heartbeat_interval_for` so it can never be set at or above the TTL.
+HEARTBEAT_INTERVAL_ENV = "GITEA_WORKER_HEARTBEAT_INTERVAL_SECONDS"
+
+#: Never beat faster than this, so a misconfigured interval cannot turn the
+#: supervisor into a busy sqlite writer.
+MIN_HEARTBEAT_INTERVAL_SECONDS = 1.0
+
+
+def heartbeat_interval_for(
+    ttl_seconds: float = DEFAULT_HEARTBEAT_TTL_SECONDS,
+    override: str | float | None = None,
+) -> float:
+    """Beat interval for *ttl_seconds*: one third of the TTL, capped at a half.
+
+    #975. A third means two consecutive beats can be lost before the row is
+    allowed to look stale, and the half-TTL cap is a hard ceiling so no
+    override can produce an interval that expires the registration it is
+    supposed to renew. That is the whole liveness contract: a healthy worker
+    stays owned, and a worker that has genuinely stopped beating still expires
+    on the configured TTL — this function never touches the TTL itself.
+    """
+    try:
+        ttl = float(ttl_seconds)
+    except (TypeError, ValueError):
+        ttl = DEFAULT_HEARTBEAT_TTL_SECONDS
+    if not ttl > 0:
+        ttl = DEFAULT_HEARTBEAT_TTL_SECONDS
+
+    interval = ttl / 3.0
+    if override is not None:
+        candidate = str(override).strip()
+        if candidate:
+            try:
+                parsed = float(candidate)
+            except (TypeError, ValueError):
+                parsed = None
+            if parsed is not None and parsed > 0:
+                interval = parsed
+
+    ceiling = ttl / 2.0
+    if interval > ceiling:
+        interval = ceiling
+    if interval < MIN_HEARTBEAT_INTERVAL_SECONDS:
+        interval = min(MIN_HEARTBEAT_INTERVAL_SECONDS, ceiling)
+    return interval
+
+
+#: Recorded-vs-presented pairs compared by :func:`_heartbeat_expectation_drift`.
+_HEARTBEAT_EXPECTATION_FIELDS = (
+    "session_id",
+    "generation_id",
+    "client_name",
+    "pid",
+)
+
+
+def _heartbeat_expectation_drift(
+    record: dict[str, Any],
+    *,
+    expected_session_id: str | None = None,
+    expected_generation_id: str | None = None,
+    expected_client_name: str | None = None,
+    expected_pid: int | None = None,
+) -> list[tuple[str, Any, Any]]:
+    """Return ``(field, recorded, presented)`` for every mismatched expectation.
+
+    Only supplied expectations are compared, so a caller that presents nothing
+    gets the pre-#975 behaviour. ``client_name`` is compared through
+    :func:`normalize_client_name` so one application's namespaces stay one
+    client while genuinely different clients stay distinct.
+    """
+    presented = {
+        "session_id": expected_session_id,
+        "generation_id": expected_generation_id,
+        "client_name": (
+            normalize_client_name(expected_client_name)
+            if expected_client_name is not None
+            else None
+        ),
+        "pid": expected_pid,
+    }
+    drift: list[tuple[str, Any, Any]] = []
+    for field in _HEARTBEAT_EXPECTATION_FIELDS:
+        want = presented[field]
+        if want is None:
+            continue
+        got = record.get(field)
+        if field == "pid":
+            match = got is not None and int(got) == int(want)
+        else:
+            match = got == want
+        if not match:
+            drift.append((field, got, want))
+    return drift
+
+
 STATUS_ACTIVE = "active"
 STATUS_SUPERSEDED = "superseded"
 STATUS_RELEASED = "released"
@@ -203,8 +301,22 @@ CREATE INDEX IF NOT EXISTS idx_worker_session
     ON worker_registrations(session_id, status);
 CREATE INDEX IF NOT EXISTS idx_worker_profile
     ON worker_registrations(profile, status);
+CREATE INDEX IF NOT EXISTS idx_worker_instance
+    ON worker_registrations(client_instance_id, status);
 """
 
+#: #978 optional columns added without rewriting historical rows.
+_SCHEMA_OPTIONAL_COLUMNS: tuple[tuple[str, str], ...] = (
+    ("fleet_run_id", "TEXT"),
+    ("authenticated_account", "TEXT"),
+    ("process_identity", "TEXT"),
+    ("startup_revision", "TEXT"),
+    ("loaded_revision", "TEXT"),
+    ("parity_revision", "TEXT"),
+    ("live_revision", "TEXT"),
+    ("instance_id_provenance", "TEXT"),
+)
+
 
 class WorkerRegistryError(RuntimeError):
     """Raised for registry misuse that is a programming error, not a refusal."""
@@ -536,6 +648,17 @@ class WorkerRegistry:
             conn = self._connect()
             try:
                 conn.executescript(_SCHEMA_SQL)
+                existing = {
+                    row[1]
+                    for row in conn.execute(
+                        "PRAGMA table_info(worker_registrations)"
+                    ).fetchall()
+                }
+                for name, decl in _SCHEMA_OPTIONAL_COLUMNS:
+                    if name not in existing:
+                        conn.execute(
+                            f"ALTER TABLE worker_registrations ADD COLUMN {name} {decl}"
+                        )
                 conn.commit()
             finally:
                 conn.close()
@@ -660,6 +783,14 @@ class WorkerRegistry:
         heartbeat_ttl_seconds: float = DEFAULT_HEARTBEAT_TTL_SECONDS,
         now: datetime | None = None,
         pid_alive_probe=None,
+        fleet_run_id: str | None = None,
+        authenticated_account: str | None = None,
+        process_identity: str | None = None,
+        startup_revision: str | None = None,
+        loaded_revision: str | None = None,
+        parity_revision: str | None = None,
+        live_revision: str | None = None,
+        instance_id_provenance: str | None = None,
     ) -> dict[str, Any]:
         """Atomically register one worker identity.
 
@@ -668,6 +799,10 @@ class WorkerRegistry:
         to mint a different identity and register that instead (AC32). The
         existing registration is returned untouched so the caller can see what
         it collided with.
+
+        #978 also fails closed when a *live* registration already holds the same
+        ``client_instance_id`` for the same namespace: two workers of one
+        namespace cannot share one instance.
         """
         parsed = parse_worker_identity(worker_identity)
         if not parsed["valid"]:
@@ -685,6 +820,9 @@ class WorkerRegistry:
             }
 
         stamp = _ts(now or _utc_now())
+        proc_id = process_identity or (
+            f"pid-{int(pid)}" if pid is not None else None
+        )
         with self._tx() as conn:
             existing = conn.execute(
                 "SELECT * FROM worker_registrations WHERE worker_identity = ?",
@@ -721,6 +859,41 @@ class WorkerRegistry:
                     ),
                 }
 
+            # #978: refuse a second live worker for the same (instance, namespace).
+            if namespace and client_instance_id:
+                peers = conn.execute(
+                    "SELECT * FROM worker_registrations "
+                    "WHERE client_instance_id = ? AND namespace = ? AND status = ?",
+                    (client_instance_id, namespace, STATUS_ACTIVE),
+                ).fetchall()
+                for peer_row in peers:
+                    peer = self._row_to_record(peer_row)
+                    pid_alive = (
+                        pid_alive_probe(peer.get("pid"))
+                        if pid_alive_probe is not None
+                        else None
+                    )
+                    if self.is_live(peer, now=now, pid_alive=pid_alive)["live"]:
+                        return {
+                            "success": False,
+                            "registered": False,
+                            "mutation_performed": False,
+                            "blocker_kind": BLOCKER_CONFLICTING_SESSIONS,
+                            "collision": True,
+                            "collision_kind": "duplicate_namespace_worker",
+                            "existing_registration": _public_record(peer),
+                            "reasons": [
+                                f"client_instance_id {client_instance_id!r} already "
+                                f"has a live worker for namespace {namespace!r} "
+                                f"({peer.get('worker_identity')!r}); #978 fails closed "
+                                "rather than registering a second worker"
+                            ],
+                            "exact_next_action": (
+                                "Stop the extra namespace worker, or use a distinct "
+                                "client_instance_id for a separate application launch."
+                            ),
+                        }
+
             epoch = self._next_epoch(conn, generation_id)
             conn.execute(
                 """
@@ -729,8 +902,11 @@ class WorkerRegistry:
                     generation_id, role, profile, namespace, remote,
                     repository_binding, pid, transport, token_fingerprint,
                     started_at, last_heartbeat_at, heartbeat_ttl_seconds,
-                    fencing_epoch, status
+                    fencing_epoch, status,
-                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
+                    fleet_run_id, authenticated_account, process_identity,
+                    startup_revision, loaded_revision, parity_revision,
+                    live_revision, instance_id_provenance
+                ) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)
                 """,
                 (
                     worker_identity,
@@ -751,6 +927,14 @@ class WorkerRegistry:
                     float(heartbeat_ttl_seconds),
                     epoch,
                     STATUS_ACTIVE,
+                    (fleet_run_id or "").strip() or None,
+                    (authenticated_account or "").strip() or None,
+                    proc_id,
+                    (startup_revision or "").strip() or None,
+                    (loaded_revision or "").strip() or None,
+                    (parity_revision or "").strip() or None,
+                    (live_revision or "").strip() or None,
+                    (instance_id_provenance or "").strip() or None,
                 ),
             )
             row = conn.execute(
@@ -786,11 +970,25 @@ class WorkerRegistry:
         worker_identity: str,
         fencing_epoch: int,
         now: datetime | None = None,
+        expected_session_id: str | None = None,
+        expected_generation_id: str | None = None,
+        expected_client_name: str | None = None,
+        expected_pid: int | None = None,
     ) -> dict[str, Any]:
         """Renew only the owning registration (#948 AC11).
 
         A stale epoch is refused rather than silently renewed, so a superseded
         session that resumes cannot heartbeat its way back into ownership.
+
+        #975 adds optional keyword-only *expectations*. Every one supplied must
+        match the recorded row or the renewal is refused. They exist because
+        identity plus epoch cannot express "renew the row I registered, and only
+        that row": a recycled PID, or a second session of the same client, would
+        otherwise be renewable by the wrong beater. They are fencing tokens,
+        never assertions — supplying one can only cause a refusal, never grant
+        anything, and omitting them preserves the pre-#975 behaviour exactly.
+        Drift reports the existing ``BLOCKER_FENCED`` rather than a new
+        ``blocker_kind``, because consumers switch on that value.
         """
         stamp = _ts(now or _utc_now())
         with self._tx() as conn:
@@ -807,6 +1005,30 @@ class WorkerRegistry:
                     "reasons": [f"no registration for {worker_identity!r}"],
                 }
             record = self._row_to_record(row)
+            drift = _heartbeat_expectation_drift(
+                record,
+                expected_session_id=expected_session_id,
+                expected_generation_id=expected_generation_id,
+                expected_client_name=expected_client_name,
+                expected_pid=expected_pid,
+            )
+            if drift:
+                return {
+                    "success": False,
+                    "renewed": False,
+                    "mutation_performed": False,
+                    "blocker_kind": BLOCKER_FENCED,
+                    "expectation_drift": drift,
+                    "reasons": [
+                        "presented worker expectations do not match the recorded "
+                        "registration, so this beater does not own the row: "
+                        + "; ".join(
+                            f"{field} recorded {recorded!r}, presented {presented!r}"
+                            for field, recorded, presented in drift
+                        )
+                        + " (#975)"
+                    ],
+                }
             if record["status"] != STATUS_ACTIVE:
                 return {
                     "success": False,
@@ -1013,8 +1235,17 @@ def _public_record(record: dict[str, Any]) -> dict[str, Any]:
         "transport": record.get("transport"),
         "started_at": record.get("started_at"),
         "last_heartbeat_at": record.get("last_heartbeat_at"),
+        "heartbeat_ttl_seconds": record.get("heartbeat_ttl_seconds"),
         "fencing_epoch": record.get("fencing_epoch"),
         "status": record.get("status"),
+        "fleet_run_id": record.get("fleet_run_id"),
+        "authenticated_account": record.get("authenticated_account"),
+        "process_identity": record.get("process_identity"),
+        "startup_revision": record.get("startup_revision"),
+        "loaded_revision": record.get("loaded_revision"),
+        "parity_revision": record.get("parity_revision"),
+        "live_revision": record.get("live_revision"),
+        "instance_id_provenance": record.get("instance_id_provenance"),
     }
 
 
@@ -1430,3 +1661,253 @@ def resolve_bound_remote(
             "explicitly to avoid host drift (#948)."
         ],
     }
+
+
+# --- Production heartbeat lifecycle (#975) --------------------------------
+#
+# #948 delivered ``WorkerRegistry.heartbeat()`` and nothing ever called it, so
+# ``last_heartbeat_at`` stayed pinned to ``started_at`` for every production
+# worker and ``heartbeat_ttl_seconds`` stopped being a liveness window at all —
+# it became a hard cap on how long any client could stay attached. This
+# supervisor is the missing caller.
+#
+# It is a daemon thread rather than an asyncio task or a per-request refresh
+# because the renewal has to survive an *idle* session: a client waiting on a
+# lease with no tool call in flight must not lose ownership, which rules out
+# request-driven refresh. Registration itself happens lazily inside a tool call
+# on the server's event loop, and the registry performs blocking
+# ``BEGIN IMMEDIATE`` sqlite writes, which must not run on that loop.
+# ``daemon=True`` is deliberate: a hard kill takes the thread down with the
+# process, so a dead worker still goes stale on the normal TTL.
+
+#: Refusals that mean this worker no longer owns its row. Beating again could
+#: only ever be an attempt to renew ownership it has already lost, so the
+#: supervisor stops permanently instead of retrying.
+TERMINAL_HEARTBEAT_BLOCKERS = frozenset({BLOCKER_FENCED, BLOCKER_NO_ATTACHMENT})
+
+
+class WorkerHeartbeatSupervisor:
+    """Periodically renew exactly one worker registration.
+
+    Contract:
+
+    * It never registers. A supervisor exists only for an already-registered
+      identity, so it cannot create a second registration or a second identity
+      system.
+    * Every beat presents the full expectation set, so it can renew only the row
+      matching this exact client name, worker identity, session, generation and
+      pid.
+    * A terminal refusal (fenced or missing) stops it permanently and records
+      why. A fenced session must never beat its way back into ownership.
+    * A transient failure (a locked database, say) is counted and the loop
+      continues, so one contended write does not silently end the heartbeat.
+    * Nothing here raises into a caller. ``beat_once`` returns its outcome and
+      the thread body swallows everything, because a heartbeat failure must
+      degrade to "not renewed" and never crash a tool call.
+    """
+
+    def __init__(
+        self,
+        registry: WorkerRegistry,
+        *,
+        worker_identity: str,
+        fencing_epoch: int,
+        session_id: str | None = None,
+        generation_id: str | None = None,
+        client_name: str | None = None,
+        pid: int | None = None,
+        ttl_seconds: float = DEFAULT_HEARTBEAT_TTL_SECONDS,
+        interval_seconds: float | None = None,
+        clock=None,
+    ) -> None:
+        self._registry = registry
+        self.worker_identity = worker_identity
+        self.fencing_epoch = int(fencing_epoch)
+        self.session_id = session_id
+        self.generation_id = generation_id
+        self.client_name = (
+            normalize_client_name(client_name) if client_name is not None else None
+        )
+        self.pid = int(pid) if pid is not None else None
+        self.ttl_seconds = float(ttl_seconds)
+        self.interval_seconds = (
+            heartbeat_interval_for(self.ttl_seconds)
+            if interval_seconds is None
+            else heartbeat_interval_for(self.ttl_seconds, interval_seconds)
+        )
+        #: Injectable so every TTL test uses controlled time and no test waits
+        #: for a real interval or a real TTL to elapse.
+        self._clock = clock or _utc_now
+
+        self._stop_event = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._lock = threading.Lock()
+        self._atexit_registered = False
+
+        self.started = False
+        self.stopped_reason: str | None = None
+        self.beats_attempted = 0
+        self.beats_renewed = 0
+        self.transient_failures = 0
+        self.last_beat_at: str | None = None
+        self.last_result: dict[str, Any] | None = None
+
+    # -- one beat --
+
+    def beat_once(self, now: datetime | None = None) -> dict[str, Any]:
+        """Renew once. Never raises; returns the registry outcome or a failure."""
+        if self.stopped_reason is not None:
+            return {
+                "success": False,
+                "renewed": False,
+                "beat_attempted": False,
+                "reasons": [
+                    f"supervisor already stopped: {self.stopped_reason}"
+                ],
+            }
+
+        stamp = now or self._clock()
+        self.beats_attempted += 1
+        try:
+            result = self._registry.heartbeat(
+                worker_identity=self.worker_identity,
+                fencing_epoch=self.fencing_epoch,
+                now=stamp,
+                expected_session_id=self.session_id,
+                expected_generation_id=self.generation_id,
+                expected_client_name=self.client_name,
+                expected_pid=self.pid,
+            )
+        except Exception as exc:
+            # Transient by assumption: an unexpected error is not proof this
+            # worker lost ownership, so it must not silently end the heartbeat.
+            self.transient_failures += 1
+            result = {
+                "success": False,
+                "renewed": False,
+                "transient": True,
+                "blocker_kind": None,
+                "reasons": [f"heartbeat raised {type(exc).__name__}: {exc}"],
+            }
+            self.last_result = result
+            return result
+
+        self.last_result = result
+        if result.get("renewed"):
+            self.beats_renewed += 1
+            self.last_beat_at = result.get("last_heartbeat_at") or _ts(stamp)
+            return result
+
+        blocker = result.get("blocker_kind")
+        if blocker in TERMINAL_HEARTBEAT_BLOCKERS:
+            self._stop_internal(
+                reason=(
+                    f"refused with blocker_kind={blocker!r}: "
+                    + "; ".join(result.get("reasons") or [])
+                )
+            )
+        else:
+            self.transient_failures += 1
+        return result
+
+    # -- lifecycle --
+
+    def start(self) -> dict[str, Any]:
+        """Start the beat thread. Idempotent; safe to call from a tool call."""
+        with self._lock:
+            if self.stopped_reason is not None:
+                return {
+                    "started": False,
+                    "reasons": [f"supervisor stopped: {self.stopped_reason}"],
+                }
+            if self._thread is not None and self._thread.is_alive():
+                return {"started": True, "already_running": True, "reasons": []}
+
+            self._stop_event.clear()
+            thread = threading.Thread(
+                target=self._run,
+                name=f"gitea-worker-heartbeat-{self.worker_identity}",
+                daemon=True,
+            )
+            self._thread = thread
+            self.started = True
+            if not self._atexit_registered:
+                # Orderly shutdown stops the heartbeat. A hard kill does not
+                # run this, which is correct: the row must then go stale.
+                atexit.register(self._atexit_stop)
+                self._atexit_registered = True
+            thread.start()
+            return {"started": True, "already_running": False, "reasons": []}
+
+    def _run(self) -> None:
+        while not self._stop_event.is_set():
+            # Wait first: registration already stamped a fresh heartbeat, so an
+            # immediate beat would be a redundant write on every server launch.
+            if self._stop_event.wait(self.interval_seconds):
+                return
+            try:
+                self.beat_once()
+            except Exception:
+                # beat_once is already total; this is the last-resort guard that
+                # keeps a supervisor thread from dying silently.
+                self.transient_failures += 1
+            if self.stopped_reason is not None:
+                return
+
+    def stop(self, reason: str = "stopped") -> dict[str, Any]:
+        """Stop beating. Idempotent, and prompt because the loop waits on an Event."""
+        self._stop_internal(reason=reason)
+        thread = self._thread
+        if thread is not None and thread.is_alive():
+            thread.join(timeout=max(1.0, min(5.0, self.interval_seconds)))
+        return {
+            "stopped": True,
+            "reason": self.stopped_reason,
+            "thread_alive": bool(thread is not None and thread.is_alive()),
+        }
+
+    def _stop_internal(self, *, reason: str) -> None:
+        if self.stopped_reason is None:
+            self.stopped_reason = reason
+        self._stop_event.set()
+        if self._atexit_registered:
+            try:
+                atexit.unregister(self._atexit_stop)
+            except Exception:
+                pass
+            self._atexit_registered = False
+
+    def _atexit_stop(self) -> None:
+        try:
+            self.stop(reason="process exit")
+        except Exception:
+            pass
+
+    # -- observability --
+
+    def status(self) -> dict[str, Any]:
+        """Read-only observability payload; safe to embed in a tool result."""
+        thread = self._thread
+        return {
+            "supervised": True,
+            "worker_identity": self.worker_identity,
+            "fencing_epoch": self.fencing_epoch,
+            "session_id": self.session_id,
+            "generation_id": self.generation_id,
+            "client_name": self.client_name,
+            "pid": self.pid,
+            "heartbeat_ttl_seconds": self.ttl_seconds,
+            "heartbeat_interval_seconds": self.interval_seconds,
+            "started": self.started,
+            "running": bool(
+                thread is not None
+                and thread.is_alive()
+                and self.stopped_reason is None
+            ),
+            "stopped_reason": self.stopped_reason,
+            "beats_attempted": self.beats_attempted,
+            "beats_renewed": self.beats_renewed,
+            "transient_failures": self.transient_failures,
+            "last_heartbeat_at": self.last_beat_at,
+            "last_blocker_kind": (self.last_result or {}).get("blocker_kind"),
+        }

namespace_workspace_binding.py

@@ -8,8 +8,10 @@ poison workspace purity checks in another namespace.
 from __future__ import annotations
 
 import os
+import subprocess
 
 import author_mutation_worktree as amw
+import canonical_repository_root as crr
 
 ACTIVE_WORKTREE_ENV = amw.ACTIVE_WORKTREE_ENV
 AUTHOR_WORKTREE_ENV = amw.AUTHOR_WORKTREE_ENV
@@ -152,6 +154,60 @@ def resolve_namespace_workspace(
     return os.path.realpath(process_project_root), "MCP server process root (default)"
 
 
+def verify_git_common_directory_membership(
+    workspace_path: str,
+    canonical_repo_root: str,
+) -> tuple[bool, str | None]:
+    """Verify that workspace_path belongs to canonical_repo_root via git common-dir or branches containment."""
+    ws = (workspace_path or "").strip()
+    root = (canonical_repo_root or "").strip()
+    if not ws or not root:
+        return False, "empty workspace or canonical root path"
+
+    try:
+        real_ws = os.path.realpath(os.path.abspath(ws))
+        real_root = os.path.realpath(os.path.abspath(root))
+    except Exception as exc:
+        return False, f"invalid workspace or root path: {exc}"
+
+    if real_ws == real_root:
+        return True, None
+
+    if not os.path.isdir(real_ws):
+        return False, f"workspace directory '{real_ws}' does not exist"
+
+    try:
+        res = subprocess.run(
+            ["git", "-C", real_ws, "rev-parse", "--git-common-dir"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if res.returncode == 0:
+            common_raw = (res.stdout or "").strip()
+            common_dir = amw._realpath_git_common_dir(real_ws, common_raw)
+            real_common = os.path.realpath(common_dir)
+            canonical_git = os.path.realpath(os.path.join(real_root, ".git"))
+
+            if real_common in (canonical_git, real_root):
+                return True, None
+            return (
+                False,
+                f"workspace '{real_ws}' git common directory '{real_common}' does not match "
+                f"canonical repository root '{real_root}' (.git at '{canonical_git}')"
+            )
+        else:
+            return (
+                False,
+                f"workspace '{real_ws}' is not a valid git repository or git rev-parse failed"
+            )
+    except Exception as exc:
+        return (
+            False,
+            f"failed to inspect git common directory for workspace '{real_ws}': {exc}"
+        )
+
+
 def resolve_namespace_mutation_context(
     *,
     role_kind: str,
@@ -163,6 +219,8 @@ def resolve_namespace_mutation_context(
     worktree: str | None = None,
     profile_name: str | None = None,
     configured_canonical_root: str | None = None,
+    expected_slug: str | None = None,
+    remote: str | None = None,
 ) -> dict:
     """Shared workspace resolution for runtime_context and mutation guards.
 
@@ -180,11 +238,41 @@ def resolve_namespace_mutation_context(
     env_map = env if env is not None else os.environ
     process_root = os.path.realpath(process_project_root)
     role = normalize_role_kind(role_kind, profile_name=profile_name)
-    configured = (configured_canonical_root or "").strip()
+
-    if configured:
+    configured_val = (configured_canonical_root or "").strip()
-        canonical_root = os.path.realpath(configured)
+    if configured_val:
+        crr_assessment = crr.assess_canonical_repository_root(
+            configured_value=configured_val,
+            source="configured_canonical_root",
+            expected_slug=expected_slug,
+            process_project_root=process_root,
+            remote=remote,
+            require_binding=True,
+        )
+        # #973 B10: a refused repository-authority mode deliberately resolves no
+        # canonical root, so downstream guards keep evaluating the install
+        # checkout rather than an identity derived through an undefined mode.
+        # ``roots_aligned`` still follows ``proven`` and stays False, and the
+        # refusal (with its reason_code) rides along in
+        # ``canonical_root_assessment`` — this is a fail-closed fallback, never a
+        # normalisation of the mode.
+        canonical_root = crr_assessment["canonical_repo_root"] or amw.resolve_canonical_repo_root(
+            process_root, process_root
+        )
+        roots_aligned = crr_assessment["proven"]
     else:
-        canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
+        crr_assessment = {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "configured": False,
+            "canonical_repo_root": amw.resolve_canonical_repo_root(process_root, process_root),
+            "resolved_slug": None,
+            "source": None,
+            "reason_code": None,
+        }
+        canonical_root = crr_assessment["canonical_repo_root"]
+        roots_aligned = (canonical_root == process_root)
 
     durable: dict | None = None
     if role == "author":
@@ -234,7 +322,10 @@ def resolve_namespace_mutation_context(
         "ignored_bindings": demotions + (pollution.get("ignored_bindings") or []),
         "process_project_root": process_root,
         "canonical_repo_root": canonical_root,
-        "roots_aligned": canonical_root == process_root,
+        "roots_aligned": roots_aligned,
+        "canonical_root_assessment": crr_assessment,
+        "expected_slug": expected_slug,
+        "remote": remote,
     }
     if durable is not None:
         result["author_worktree_resolution"] = durable
@@ -246,6 +337,15 @@ def resolve_namespace_mutation_context(
         result["author_worktree_reasons"] = list(durable.get("reasons") or [])
         result["author_worktree_blocker_kind"] = durable.get("blocker_kind")
         result["operator_recovery"] = durable.get("operator_recovery")
+    else:
+        path_exists = os.path.exists(workspace)
+        result["path_exists"] = path_exists
+        result["in_git_worktree_list"] = (
+            amw.path_in_git_worktree_list(workspace, canonical_root)
+            if path_exists
+            else False
+        )
+        result["bound_worktree_missing"] = not path_exists
     return result
 
 
@@ -378,8 +478,8 @@ def format_namespace_workspace_binding_error(
 def assess_namespace_mutation_workspace(
     *,
     role_kind: str,
-    worktree_path: str | None,
+    worktree_path: str | None = None,
-    worktree: str | None,
+    worktree: str | None = None,
     process_project_root: str,
     env: dict[str, str] | os._Environ | None = None,
     session_lease_worktree: str | None = None,
@@ -387,6 +487,8 @@ def assess_namespace_mutation_workspace(
     profile_name: str | None = None,
     current_branch: str | None = None,
     configured_canonical_root: str | None = None,
+    expected_slug: str | None = None,
+    remote: str | None = None,
 ) -> dict:
     """Evaluate namespace workspace binding before preflight/mutation."""
     ctx = resolve_namespace_mutation_context(
@@ -399,6 +501,8 @@ def assess_namespace_mutation_workspace(
         session_lock_worktree=session_lock_worktree,
         profile_name=profile_name,
         configured_canonical_root=configured_canonical_root,
+        expected_slug=expected_slug,
+        remote=remote,
     )
     mutation_workspace = ctx["workspace_path"]
     binding_source = ctx["workspace_binding_source"]
@@ -422,6 +526,27 @@ def assess_namespace_mutation_workspace(
 
     reasons = list(metadata.get("reasons") or [])
     operator_recovery = ctx.get("operator_recovery")
+
+    crr_reasons = list(ctx.get("canonical_root_assessment", {}).get("reasons") or [])
+    if crr_reasons:
+        reasons.extend(crr_reasons)
+
+    path_exists = ctx.get("path_exists")
+    if path_exists is None:
+        path_exists = os.path.exists(mutation_workspace)
+
+    if not path_exists:
+        if role != "author":
+            reasons.append(
+                f"{role} mutation blocked: configured workspace directory '{mutation_workspace}' does not exist (nonexistent worktree)"
+            )
+    else:
+        valid_common, common_err = verify_git_common_directory_membership(
+            mutation_workspace, ctx["canonical_repo_root"]
+        )
+        if not valid_common and common_err:
+            reasons.append(common_err)
+
     if role == "author":
         # #618 durable resolution already validated existence, membership,
         # branches/, lock ownership, and traversal safety when present.
@@ -438,27 +563,28 @@ def assess_namespace_mutation_workspace(
             )
             if branches["block"]:
                 reasons.extend(branches["reasons"])
-    elif (
+    elif role in {"reviewer", "merger"}:
-        role == "reviewer"
+        if mutation_workspace == process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
-        and mutation_workspace == process_root
-        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
-    ):
             reasons.append(
                 f"{role} mutation blocked: workspace is the stable control checkout; "
                 f"create or reconnect to a session-owned worktree under branches/ "
                 f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / "
                 f"{ACTIVE_WORKTREE_ENV}"
             )
-    elif (
+        elif mutation_workspace != process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
-        role in {"reviewer", "merger"}
-        and mutation_workspace != process_root
-        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
-    ):
             reasons.append(
                 f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
                 f"'{ctx['canonical_repo_root']}/branches/'"
             )
 
+        if path_exists and amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
+            in_list = ctx.get("in_git_worktree_list")
+            if in_list is False:
+                reasons.append(
+                    f"{role} mutation blocked: workspace '{mutation_workspace}' is under branches/ "
+                    f"but is not registered in git worktree list for '{ctx['canonical_repo_root']}'"
+                )
+
     block = bool(reasons)
     return {
         "block": block,

post_restart_reconcile.py

@@ -475,37 +475,8 @@ 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 = [
-        L for L in (inventory.get("leases") or []) if isinstance(L, Mapping)
-    ]
-    fleet_report = inventory.get("session_fleet")
-    if isinstance(fleet_report, Mapping) and "retireable_session_ids" in fleet_report:
-        fleet_details = dict(fleet_report)
-        retireable_ids = list(fleet_details.get("retireable_session_ids") or [])
-        resolved = bool(fleet_details.get("sessions_dimension_resolved", not retireable_ids))
-    else:
-        try:
-            import session_lifecycle as _sl
-
-            client_managed = inventory.get("client_managed_session_ids")
-            cm_set = None
-            if isinstance(client_managed, (list, tuple, set, frozenset)):
-                cm_set = {str(x) for x in client_managed}
-            fleet = _sl.classify_sessions(
-                sessions,
-                leases=leases_for_sessions,
-                now=started,
-                client_managed_sessions=cm_set,
-            )
-            fleet_details = fleet.as_dict()
-            retireable_ids = list(fleet_details.get("retireable_session_ids") or [])
-            resolved = bool(fleet_details.get("sessions_dimension_resolved"))
-        except Exception as exc:  # noqa: BLE001 — fail closed to legacy signal
-            # Legacy fallback: dead-pid active rows only (pre-#969 behaviour).
     orphan_sessions = [
         s
         for s in sessions
@@ -513,27 +484,15 @@ def reconcile_after_restart(
         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]
+    if 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,14 +163,17 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
-    # #969: explicit plan/apply path for dead-owner session retirement.
+    # #978: instance-level fleet identity/health snapshot. gitea.read is the
-    "retire_stale_workflow_sessions": {
+    # operation gate; the tool additionally restricts role_kind to
+    # controller|reconciler so author/reviewer/merger cannot use it as a
+    # mutation surface and no unrelated write permission is introduced.
+    "snapshot_instance_fleet": {
         "permission": "gitea.read",
-        "role": "author",
+        "role": "controller",
     },
-    "gitea_retire_stale_workflow_sessions": {
+    "gitea_snapshot_instance_fleet": {
         "permission": "gitea.read",
-        "role": "author",
+        "role": "controller",
     },
     # #644: Phase 2 Web Console recovery tasks.
     "clear_stale_binding": {
@@ -396,6 +398,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 +684,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_cross_repo_canonical_consumption.py

@@ -313,6 +313,7 @@ class TestCanonicalRootGuardBinding(_ServerHarness):
             process_project_root=self.install_root,
             remote="prgs",
             require_binding=True,
+            mode="derivation",
         )
         self.assertFalse(got.get("block"), got.get("reasons"))
         self.assertEqual(got["resolved_slug"], TARGET_SLUG)

tests/test_issue_706_canonical_repository_root.py

@@ -241,9 +241,10 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase):
             process_project_root=self.install,
             env={},
             configured_canonical_root=self.target,
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
         )
         self.assertEqual(ctx["canonical_repo_root"], self.target)
-        self.assertFalse(ctx["roots_aligned"])
+        self.assertTrue(ctx["roots_aligned"])
 
     def test_target_worktree_is_member_of_target_root(self):
         got = nwb.amw.assess_workspace_repo_membership(
@@ -268,6 +269,7 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase):
             env={},
             current_branch="feat/issue-1",
             configured_canonical_root=self.target,
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
         )
         self.assertFalse(assessment["block"], assessment.get("reasons"))
         self.assertEqual(assessment["canonical_repo_root"], self.target)

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_issue_973_b10_mode_contract.py

@@ -0,0 +1,739 @@
+"""Regression tests for Issue #973 blocker B10: repository-authority mode contract.
+
+Before this repair ``assess_canonical_repository_root`` never checked its
+``mode`` argument against an allowlist. Both dispatch points were permissive:
+
+* the configured-root path tested ``mode == "derivation"`` and sent every other
+  value into a catch-all ``else``, so an unsupported mode silently received
+  *validation* semantics, and
+* the single-repository default path tested ``mode == "validation"``, so an
+  unsupported mode skipped the identity comparison entirely and was strictly
+  *weaker* than validation.
+
+The measured consequence was that ``mode="invalid_mode"`` with matching expected
+and observed identities returned ``proven: True`` / ``block: False`` with no
+reasons, and that an unsupported mode passed on the default path where
+``"validation"`` correctly blocked.
+
+These tests exercise the production module directly with real git repositories —
+no patched stand-in for the function under test — and drive the production
+enforcement and mutation-context consumers rather than only the intermediate
+assessment.
+"""
+
+from __future__ import annotations
+
+import ast
+import inspect
+import os
+import subprocess
+import tempfile
+import unittest
+from unittest.mock import patch
+
+import canonical_repository_root as crr
+import gitea_config
+import gitea_mcp_server as mcp_server
+import namespace_workspace_binding as nwb
+import stable_control_runtime
+
+
+INSTALL_SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
+TARGET_SLUG = "Scaled-Tech-Consulting/mcp-control-plane"
+FOREIGN_SLUG = "Someone-Else/Evil-Repo"
+
+# Explicitly supplied values that must all be refused. Omission is *not* in this
+# list: omitting the argument keeps the documented ``"validation"`` default.
+UNSUPPORTED_STRING_MODES = (
+    "invalid_mode",
+    "",
+    "validaton",          # misspelling
+    "derivaton",          # misspelling
+    "Validation",         # case variant
+    "DERIVATION",         # case variant
+    " validation",        # leading whitespace
+    "validation ",        # trailing whitespace
+    "derivation\n",       # trailing newline
+    "validation,derivation",
+)
+
+UNSUPPORTED_NON_STRING_MODES = (
+    None,
+    0,
+    1,
+    True,
+    False,
+    3.14,
+    [],
+    ["validation"],
+    {},
+    {"mode": "validation"},
+    ("validation",),
+    object(),
+)
+
+
+def _init_repo(path: str, remote_url: str, *, user: str = "Test User") -> None:
+    os.makedirs(path, exist_ok=True)
+    subprocess.run(["git", "init", "-b", "master"], cwd=path, check=True,
+                   capture_output=True)
+    subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=path,
+                   check=True, capture_output=True)
+    subprocess.run(["git", "config", "user.name", user], cwd=path, check=True,
+                   capture_output=True)
+    with open(os.path.join(path, "README.md"), "w") as handle:
+        handle.write(f"{os.path.basename(path)}\n")
+    subprocess.run(["git", "add", "README.md"], cwd=path, check=True,
+                   capture_output=True)
+    subprocess.run(["git", "commit", "-m", "initial"], cwd=path, check=True,
+                   capture_output=True)
+    subprocess.run(["git", "remote", "add", "prgs", remote_url], cwd=path,
+                   check=True, capture_output=True)
+
+
+class _CanonicalRootFixture(unittest.TestCase):
+    """Real install / target / foreign git repositories, as in the #973 suite."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp_dir = os.path.realpath(self._tmp.name)
+
+        self.install_root = os.path.join(self.tmp_dir, "Gitea-Tools")
+        _init_repo(
+            self.install_root,
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+
+        self.target_root = os.path.join(self.tmp_dir, "mcp-control-plane")
+        _init_repo(
+            self.target_root,
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+
+        self.evil_root = os.path.join(self.tmp_dir, "Evil-Repo")
+        _init_repo(
+            self.evil_root,
+            "https://gitea.prgs.cc/Someone-Else/Evil-Repo.git",
+            user="Evil User",
+        )
+
+        self.target_branches = os.path.join(self.target_root, "branches")
+        self.target_worktree = os.path.join(self.target_branches, "rev-pr-99")
+        subprocess.run(
+            ["git", "worktree", "add", "-b", "rev-pr-99", self.target_worktree],
+            cwd=self.target_root, check=True, capture_output=True,
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def assertRefusedForMode(self, assessment: dict, mode) -> None:
+        """Assert a fail-closed refusal attributable to *mode* and nothing else."""
+        self.assertFalse(assessment["proven"], assessment)
+        self.assertTrue(assessment["block"], assessment)
+        self.assertEqual(assessment["reason_code"], crr.DENY_UNKNOWN_MODE, assessment)
+        self.assertEqual(len(assessment["reasons"]), 1, assessment)
+        reason = assessment["reasons"][0]
+        self.assertIn("unsupported repository-authority mode", reason)
+        self.assertIn(repr(mode), reason)
+        # No trusted repository identity may be derived through an invalid mode.
+        self.assertIsNone(assessment["resolved_slug"], assessment)
+        self.assertIsNone(assessment["canonical_repo_root"], assessment)
+
+
+class TestB10UnsupportedModeIsRejected(_CanonicalRootFixture):
+    """Direct assessment tests for invalid-mode parsing."""
+
+    def test_invalid_string_with_missing_expected_identity(self):
+        """G1: blocks for the mode, not incidentally for a missing identity."""
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="env",
+            expected_slug=None,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode="invalid_mode",
+        )
+        self.assertRefusedForMode(assessment, "invalid_mode")
+        self.assertFalse(
+            any("unprovable or missing" in r for r in assessment["reasons"]),
+            "must block because the mode is unsupported, not because the expected "
+            "identity happened to be missing",
+        )
+
+    def test_invalid_string_with_matching_identities(self):
+        """G2: the contract violation — matching identities used to return proven."""
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="env",
+            expected_slug=TARGET_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode="invalid_mode",
+        )
+        self.assertRefusedForMode(assessment, "invalid_mode")
+
+    def test_invalid_string_with_conflicting_identities(self):
+        """G3: refused for the mode, not for the incidental identity mismatch."""
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.evil_root,
+            source="env",
+            expected_slug=INSTALL_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode="invalid_mode",
+        )
+        self.assertRefusedForMode(assessment, "invalid_mode")
+        self.assertFalse(
+            any("identity mismatch" in r for r in assessment["reasons"]),
+            "the identity comparison must not have run at all",
+        )
+
+    def test_empty_string_mode(self):
+        """G4a: an explicitly supplied empty string is an unsupported value."""
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="env",
+            expected_slug=TARGET_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode="",
+        )
+        self.assertRefusedForMode(assessment, "")
+
+    def test_explicit_none_mode(self):
+        """G4b: explicit None is refused; it is not treated as omission."""
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="env",
+            expected_slug=TARGET_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode=None,
+        )
+        self.assertRefusedForMode(assessment, None)
+        self.assertIn("of type NoneType", assessment["reasons"][0])
+
+    def test_representative_non_string_modes(self):
+        for mode in UNSUPPORTED_NON_STRING_MODES:
+            with self.subTest(mode=repr(mode)):
+                assessment = crr.assess_canonical_repository_root(
+                    configured_value=self.target_root,
+                    source="env",
+                    expected_slug=TARGET_SLUG,
+                    process_project_root=self.install_root,
+                    remote="prgs",
+                    require_binding=True,
+                    mode=mode,
+                )
+                self.assertRefusedForMode(assessment, mode)
+                self.assertIn(
+                    f"of type {type(mode).__name__}", assessment["reasons"][0]
+                )
+
+    def test_unknown_strings_misspellings_and_whitespace_variants(self):
+        for mode in UNSUPPORTED_STRING_MODES:
+            with self.subTest(mode=repr(mode)):
+                assessment = crr.assess_canonical_repository_root(
+                    configured_value=self.target_root,
+                    source="env",
+                    expected_slug=TARGET_SLUG,
+                    process_project_root=self.install_root,
+                    remote="prgs",
+                    require_binding=True,
+                    mode=mode,
+                )
+                self.assertRefusedForMode(assessment, mode)
+
+    def test_supported_modes_are_exactly_two(self):
+        self.assertEqual(
+            crr.SUPPORTED_MODES, ("validation", "derivation")
+        )
+        self.assertIsNone(crr.unsupported_mode_reason("validation"))
+        self.assertIsNone(crr.unsupported_mode_reason("derivation"))
+        self.assertIsNotNone(crr.unsupported_mode_reason("invalid_mode"))
+
+
+class TestB10SupportedModesUnchanged(_CanonicalRootFixture):
+    """The repair must not disturb the two documented modes."""
+
+    def test_omitted_mode_defaults_to_validation(self):
+        """G4c/G4d: omission still selects validation, proven by both outcomes."""
+        matching = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="env",
+            expected_slug=TARGET_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+        )
+        self.assertTrue(matching["proven"], matching)
+        self.assertFalse(matching["block"], matching)
+        self.assertIsNone(matching["reason_code"], matching)
+        self.assertEqual(matching["resolved_slug"], TARGET_SLUG)
+
+        conflicting = crr.assess_canonical_repository_root(
+            configured_value=self.evil_root,
+            source="env",
+            expected_slug=INSTALL_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+        )
+        self.assertFalse(conflicting["proven"], conflicting)
+        self.assertTrue(conflicting["block"], conflicting)
+        self.assertIsNone(conflicting["reason_code"], conflicting)
+        self.assertTrue(
+            any("identity mismatch" in r for r in conflicting["reasons"]),
+            "omission must behave exactly like explicit validation",
+        )
+
+    def test_explicit_validation_retains_strict_behavior(self):
+        """G5a/G5b/G5c."""
+        ok = crr.assess_canonical_repository_root(
+            configured_value=self.target_root, source="env",
+            expected_slug=TARGET_SLUG, process_project_root=self.install_root,
+            remote="prgs", require_binding=True, mode="validation",
+        )
+        self.assertTrue(ok["proven"], ok)
+
+        mismatch = crr.assess_canonical_repository_root(
+            configured_value=self.evil_root, source="env",
+            expected_slug=INSTALL_SLUG, process_project_root=self.install_root,
+            remote="prgs", require_binding=True, mode="validation",
+        )
+        self.assertTrue(mismatch["block"], mismatch)
+        self.assertTrue(any("identity mismatch" in r for r in mismatch["reasons"]))
+
+        unprovable = crr.assess_canonical_repository_root(
+            configured_value=self.target_root, source="env",
+            expected_slug=None, process_project_root=self.install_root,
+            remote="prgs", require_binding=True, mode="validation",
+        )
+        self.assertTrue(unprovable["block"], unprovable)
+        self.assertTrue(
+            any("unprovable or missing" in r for r in unprovable["reasons"])
+        )
+
+    def test_explicit_derivation_retains_trusted_derivation(self):
+        """G5d: derivation still resolves identity with no expected slug."""
+        derived = crr.assess_canonical_repository_root(
+            configured_value=self.target_root, source="env",
+            expected_slug=None, process_project_root=self.install_root,
+            remote="prgs", require_binding=True, mode="derivation",
+        )
+        self.assertTrue(derived["proven"], derived)
+        self.assertFalse(derived["block"], derived)
+        self.assertIsNone(derived["reason_code"], derived)
+        self.assertEqual(derived["resolved_slug"], TARGET_SLUG)
+
+    def test_derivation_without_resolvable_remote_still_fails_closed(self):
+        no_remote = os.path.join(self.tmp_dir, "no-remote-target")
+        os.makedirs(no_remote)
+        subprocess.run(["git", "init", "-b", "master"], cwd=no_remote, check=True,
+                       capture_output=True)
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=no_remote, source="env", expected_slug=None,
+            process_project_root=self.install_root, remote="prgs",
+            require_binding=True, mode="derivation",
+        )
+        self.assertTrue(assessment["block"], assessment)
+        self.assertTrue(
+            any("no resolvable" in r for r in assessment["reasons"]), assessment
+        )
+
+
+class TestB10SingleRepositoryDefaultPath(_CanonicalRootFixture):
+    """G6: on the unconfigured path an invalid mode used to be weaker than validation."""
+
+    def _assess(self, mode_kwargs: dict) -> dict:
+        return crr.assess_canonical_repository_root(
+            configured_value=None,
+            source=None,
+            expected_slug=FOREIGN_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=False,
+            **mode_kwargs,
+        )
+
+    def test_validation_blocks_a_foreign_expected_identity(self):
+        """G6b: the reference behaviour the invalid mode must not undercut."""
+        got = self._assess({"mode": "validation"})
+        self.assertFalse(got["proven"], got)
+        self.assertTrue(got["block"], got)
+        self.assertTrue(any("identity mismatch" in r for r in got["reasons"]))
+
+    def test_invalid_mode_no_longer_passes_where_validation_blocks(self):
+        """G6a: identical inputs, only the mode differs — must not fail open."""
+        invalid = self._assess({"mode": "invalid_mode"})
+        self.assertRefusedForMode(invalid, "invalid_mode")
+
+        validation = self._assess({"mode": "validation"})
+        self.assertEqual(
+            invalid["proven"], validation["proven"],
+            "an unsupported mode must never be more permissive than validation",
+        )
+        self.assertTrue(invalid["block"] and validation["block"])
+
+    def test_empty_string_mode_on_default_path(self):
+        """G6c."""
+        self.assertRefusedForMode(self._assess({"mode": ""}), "")
+
+    def test_omitted_mode_on_default_path_still_validates(self):
+        got = self._assess({})
+        self.assertFalse(got["proven"], got)
+        self.assertTrue(any("identity mismatch" in r for r in got["reasons"]), got)
+
+
+class TestB10RejectionOrdering(_CanonicalRootFixture):
+    """Refusal must precede every form of candidate-root or Git inspection."""
+
+    def test_no_git_or_identity_discovery_runs_for_an_unsupported_mode(self):
+        with patch.object(crr, "resolve_repo_toplevel") as toplevel, \
+             patch.object(crr, "repository_identity_slug") as identity:
+            assessment = crr.assess_canonical_repository_root(
+                configured_value=self.target_root,
+                source="env",
+                expected_slug=TARGET_SLUG,
+                process_project_root=self.install_root,
+                remote="prgs",
+                require_binding=True,
+                mode="invalid_mode",
+            )
+        self.assertRefusedForMode(assessment, "invalid_mode")
+        toplevel.assert_not_called()
+        identity.assert_not_called()
+
+    def test_the_same_spies_do_fire_for_a_supported_mode(self):
+        """Control: proves the previous test's assertions are not vacuous."""
+        with patch.object(crr, "resolve_repo_toplevel",
+                          wraps=crr.resolve_repo_toplevel) as toplevel, \
+             patch.object(crr, "repository_identity_slug",
+                          wraps=crr.repository_identity_slug) as identity:
+            crr.assess_canonical_repository_root(
+                configured_value=self.target_root,
+                source="env",
+                expected_slug=TARGET_SLUG,
+                process_project_root=self.install_root,
+                remote="prgs",
+                require_binding=True,
+                mode="validation",
+            )
+        toplevel.assert_called()
+        identity.assert_called()
+
+    def test_nonexistent_candidate_root_still_reports_the_mode_refusal(self):
+        """No mocks: the existence check cannot have run before the refusal."""
+        nonexistent = os.path.join(self.tmp_dir, "no-such-repository")
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=nonexistent,
+            source="env",
+            expected_slug=TARGET_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode="invalid_mode",
+        )
+        self.assertRefusedForMode(assessment, "invalid_mode")
+        self.assertFalse(
+            any("does not exist" in r for r in assessment["reasons"]), assessment
+        )
+
+    def test_symlinked_candidate_root_is_not_resolved_for_an_unsupported_mode(self):
+        link = os.path.join(self.tmp_dir, "target-alias")
+        os.symlink(self.target_root, link)
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=link,
+            source="env",
+            expected_slug=TARGET_SLUG,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode="invalid_mode",
+        )
+        self.assertRefusedForMode(assessment, "invalid_mode")
+        # The refusal payload must not leak a resolved path for the candidate.
+        self.assertIsNone(assessment["canonical_repo_root"], assessment)
+
+
+class TestB10NoModeInjectionSurface(unittest.TestCase):
+    """``mode`` must not be reachable from requests, environment, or config."""
+
+    PRODUCTION_MODULES = ("gitea_mcp_server.py", "namespace_workspace_binding.py")
+
+    def _repo_root(self) -> str:
+        return os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+    def test_production_call_sites_only_pass_allowlisted_literal_modes(self):
+        """Valid hardcoded modes reach the correct path; nothing else is passed."""
+        seen: list[tuple[str, str | None]] = []
+        for name in self.PRODUCTION_MODULES:
+            path = os.path.join(self._repo_root(), name)
+            with open(path) as handle:
+                tree = ast.parse(handle.read())
+            for node in ast.walk(tree):
+                if not isinstance(node, ast.Call):
+                    continue
+                func = node.func
+                called = (
+                    func.attr if isinstance(func, ast.Attribute)
+                    else getattr(func, "id", None)
+                )
+                if called != "assess_canonical_repository_root":
+                    continue
+                supplied = [k for k in node.keywords if k.arg == "mode"]
+                if not supplied:
+                    seen.append((name, None))
+                    continue
+                value = supplied[0].value
+                self.assertIsInstance(
+                    value, ast.Constant,
+                    f"{name}: mode must be a literal, never a variable or expression",
+                )
+                self.assertIn(
+                    value.value, crr.SUPPORTED_MODES,
+                    f"{name}: unsupported mode literal {value.value!r}",
+                )
+                seen.append((name, value.value))
+        self.assertTrue(seen, "expected production call sites to be found")
+        # Both documented modes are exercised by production, and omission is used.
+        self.assertIn(None, [mode for _, mode in seen])
+        self.assertIn("derivation", [mode for _, mode in seen])
+
+    def test_no_public_entry_point_exposes_a_mode_parameter(self):
+        for func in (
+            nwb.resolve_namespace_mutation_context,
+            nwb.assess_namespace_mutation_workspace,
+            mcp_server._resolve_namespace_mutation_context,
+            mcp_server._enforce_canonical_repository_root,
+            mcp_server._canonical_repository_slug,
+            mcp_server._trusted_session_repository,
+            mcp_server._resolve_expected_repository_slug,
+        ):
+            with self.subTest(func=func.__name__):
+                self.assertNotIn("mode", inspect.signature(func).parameters)
+
+    def test_no_environment_key_selects_a_repository_authority_mode(self):
+        for key in gitea_config.RECOGNIZED_GITEA_ENV_KEYS:
+            self.assertNotIn(
+                "CANONICAL_REPOSITORY_MODE", key.upper(),
+                f"{key} would expose a repository-authority mode selector",
+            )
+        # An invented mode-ish variable is simply not consumed by crr.
+        env = {
+            "GITEA_CANONICAL_REPOSITORY_ROOT": "/some/path",
+            "GITEA_CANONICAL_REPOSITORY_MODE": "invalid_mode",
+        }
+        value, source = crr.configured_canonical_root(None, env)
+        self.assertEqual(value, "/some/path")
+        self.assertNotIn("mode", (source or "").lower())
+        self.assertIn(
+            "GITEA_CANONICAL_REPOSITORY_MODE",
+            gitea_config.get_unconsumed_gitea_env_overrides(env),
+            "an unknown GITEA_* key must still be rejected as unrecognised",
+        )
+
+    def test_repository_configuration_carries_no_mode_field(self):
+        profile = {
+            "canonical_repository_root": "/some/path",
+            "mode": "invalid_mode",
+        }
+        value, source = crr.configured_canonical_root(profile, {})
+        self.assertEqual(value, "/some/path")
+        self.assertEqual(source, "profile canonical_repository_root")
+
+
+class TestB10ProductionEnforcementPaths(_CanonicalRootFixture):
+    """Production enforcement, mutation-context, reviewer and merger consumers."""
+
+    def _force_invalid_mode(self):
+        """Simulate a future call site threading an unsupported mode.
+
+        The real ``assess_canonical_repository_root`` still executes — only the
+        caller-side argument is substituted — so the refusal under test is
+        produced by production code, not by a stand-in. No caller-controlled
+        ``mode`` parameter is added to any production signature to achieve this.
+        """
+        real = crr.assess_canonical_repository_root
+
+        def _wrapper(**kwargs):
+            kwargs["mode"] = "invalid_mode"
+            return real(**kwargs)
+
+        return patch.object(crr, "assess_canonical_repository_root", _wrapper)
+
+    def test_mutation_context_fails_closed_under_a_refused_mode(self):
+        with self._force_invalid_mode():
+            ctx = nwb.resolve_namespace_mutation_context(
+                role_kind="reviewer",
+                worktree_path=self.target_worktree,
+                process_project_root=self.install_root,
+                env={},
+                configured_canonical_root=self.target_root,
+                expected_slug=TARGET_SLUG,
+                remote="prgs",
+            )
+        assessment = ctx["canonical_root_assessment"]
+        self.assertFalse(ctx["roots_aligned"], ctx)
+        self.assertTrue(assessment["block"], assessment)
+        self.assertEqual(assessment["reason_code"], crr.DENY_UNKNOWN_MODE)
+        self.assertIsNone(assessment["resolved_slug"])
+        # The refused mode must not yield the candidate root as canonical.
+        self.assertNotEqual(ctx["canonical_repo_root"], self.target_root)
+        self.assertEqual(ctx["canonical_repo_root"], self.install_root)
+
+    def test_mutation_context_unchanged_for_the_supported_default(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="reviewer",
+            worktree_path=self.target_worktree,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=self.target_root,
+            expected_slug=TARGET_SLUG,
+            remote="prgs",
+        )
+        self.assertTrue(ctx["roots_aligned"], ctx)
+        self.assertEqual(ctx["canonical_repo_root"], self.target_root)
+        self.assertIsNone(ctx["canonical_root_assessment"]["reason_code"])
+
+    def test_reviewer_and_merger_authorization_fails_closed_under_a_refused_mode(self):
+        for role in ("reviewer", "merger"):
+            with self.subTest(role=role), self._force_invalid_mode():
+                assessment = nwb.assess_namespace_mutation_workspace(
+                    role_kind=role,
+                    worktree_path=self.target_worktree,
+                    worktree=None,
+                    process_project_root=self.install_root,
+                    env={},
+                    configured_canonical_root=self.target_root,
+                    expected_slug=TARGET_SLUG,
+                    remote="prgs",
+                )
+                self.assertTrue(assessment["block"], assessment)
+                self.assertTrue(
+                    any("unsupported repository-authority mode" in r
+                        for r in assessment["reasons"]),
+                    assessment,
+                )
+
+    def test_reviewer_and_merger_authorization_unchanged_for_supported_modes(self):
+        for role in ("reviewer", "merger"):
+            with self.subTest(role=role):
+                assessment = nwb.assess_namespace_mutation_workspace(
+                    role_kind=role,
+                    worktree_path=self.target_worktree,
+                    worktree=None,
+                    process_project_root=self.install_root,
+                    env={},
+                    configured_canonical_root=self.target_root,
+                    expected_slug=TARGET_SLUG,
+                    remote="prgs",
+                )
+                self.assertFalse(assessment["block"], assessment)
+
+    def test_final_mutation_gate_blocks_when_the_mode_refusal_unaligns_roots(self):
+        with self._force_invalid_mode():
+            ctx = nwb.resolve_namespace_mutation_context(
+                role_kind="reviewer",
+                worktree_path=self.target_worktree,
+                process_project_root=self.install_root,
+                env={},
+                configured_canonical_root=self.target_root,
+                expected_slug=TARGET_SLUG,
+                remote="prgs",
+            )
+        report = stable_control_runtime.build_runtime_report(
+            process_root=self.install_root,
+            checkout_branch="master",
+            runtime_head="abcdef123456",
+            active_task_workspace=ctx["workspace_path"],
+            canonical_repository_root=ctx["canonical_repo_root"],
+            workspace_roots_aligned=ctx["roots_aligned"],
+        )
+        gate = stable_control_runtime.assess_runtime_mutation_gate(report)
+        self.assertTrue(gate["block"], gate)
+
+    def test_enforce_canonical_repository_root_uses_the_validation_default(self):
+        """B8 boundary intact: a foreign configured root raises through production."""
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server, "_configured_canonical_root",
+                          return_value=(self.evil_root, "env")), \
+             patch.object(mcp_server.session_ctx, "get_session_context",
+                          return_value=None):
+            with self.assertRaises(RuntimeError) as raised:
+                mcp_server._enforce_canonical_repository_root(remote="prgs")
+        self.assertIn("identity mismatch", str(raised.exception))
+
+    def test_enforce_canonical_repository_root_refuses_a_threaded_invalid_mode(self):
+        bound = {"org": "Scaled-Tech-Consulting",
+                 "repository": "mcp-control-plane", "remote": "prgs"}
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server, "_configured_canonical_root",
+                          return_value=(self.target_root, "env")), \
+             patch.object(mcp_server.session_ctx, "get_session_context",
+                          return_value=bound), \
+             self._force_invalid_mode():
+            with self.assertRaises(RuntimeError) as raised:
+                mcp_server._enforce_canonical_repository_root(remote="prgs")
+        self.assertIn("unsupported repository-authority mode", str(raised.exception))
+
+    def test_enforce_canonical_repository_root_passes_for_a_valid_binding(self):
+        bound = {"org": "Scaled-Tech-Consulting",
+                 "repository": "mcp-control-plane", "remote": "prgs"}
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server, "_configured_canonical_root",
+                          return_value=(self.target_root, "env")), \
+             patch.object(mcp_server.session_ctx, "get_session_context",
+                          return_value=bound), \
+             patch.object(mcp_server.session_ctx, "assess_session_context",
+                          return_value={"block": False, "reasons": []}), \
+             patch.object(mcp_server, "get_profile",
+                          return_value={"profile_name": "prgs-reviewer"}):
+            mcp_server._enforce_canonical_repository_root(remote="prgs")
+
+    def test_canonical_repository_slug_derivation_unbroken(self):
+        """B9 boundary intact: legitimate cross-repository derivation still works."""
+        profile = {
+            "profile_name": "prgs-author",
+            "allowed_repositories": [TARGET_SLUG],
+            "canonical_repository_root": self.target_root,
+        }
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server.session_ctx, "get_session_context",
+                          return_value=None):
+            slug, reasons = mcp_server._canonical_repository_slug(profile, "prgs")
+            self.assertEqual(slug, TARGET_SLUG, reasons)
+            self.assertEqual(reasons, [])
+
+    def test_canonical_repository_slug_fails_closed_under_a_refused_mode(self):
+        profile = {
+            "profile_name": "prgs-author",
+            "allowed_repositories": [TARGET_SLUG],
+            "canonical_repository_root": self.target_root,
+        }
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server.session_ctx, "get_session_context",
+                          return_value=None), \
+             self._force_invalid_mode():
+            slug, reasons = mcp_server._canonical_repository_slug(profile, "prgs")
+            self.assertIsNone(slug)
+            self.assertTrue(
+                any("unsupported repository-authority mode" in r for r in reasons),
+                reasons,
+            )
+            result = mcp_server._trusted_session_repository(
+                profile, "prgs", for_mutation=True
+            )
+            self.assertIsNone(result["org"])
+            self.assertIsNone(result["repository"])
+            self.assertTrue(result["reasons"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_973_cross_repo_canonical_roots.py

@@ -0,0 +1,546 @@
+"""Regression tests for Issue #973: validated cross-repository canonical roots."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import tempfile
+import unittest
+from unittest.mock import patch, MagicMock
+from pathlib import Path
+import subprocess
+
+import gitea_config
+import namespace_workspace_binding as nwb
+import canonical_repository_root as crr
+import stable_control_runtime
+import gitea_mcp_server as mcp_server
+
+
+class TestIssue973RecognizedEnvKeys(unittest.TestCase):
+    """Test recognized environment variable keys under #973."""
+
+    def test_recognized_gitea_env_keys(self):
+        for key in (
+            "GITEA_CANONICAL_REPOSITORY_ROOT",
+            "GITEA_REVIEWER_WORKTREE",
+            "GITEA_MERGER_WORKTREE",
+            "GITEA_MCP_SESSION_STATE_TTL_HOURS",
+        ):
+            self.assertIn(key, gitea_config.RECOGNIZED_GITEA_ENV_KEYS)
+
+    def test_get_unconsumed_gitea_env_overrides_ignores_recognized(self):
+        env = {
+            "GITEA_CANONICAL_REPOSITORY_ROOT": "/some/path",
+            "GITEA_REVIEWER_WORKTREE": "/some/reviewer/path",
+            "GITEA_MERGER_WORKTREE": "/some/merger/path",
+            "GITEA_MCP_SESSION_STATE_TTL_HOURS": "24",
+            "GITEA_UNRECOGNIZED_FOO_VAR": "bar",
+        }
+        unconsumed = gitea_config.get_unconsumed_gitea_env_overrides(env)
+        self.assertNotIn("GITEA_CANONICAL_REPOSITORY_ROOT", unconsumed)
+        self.assertNotIn("GITEA_REVIEWER_WORKTREE", unconsumed)
+        self.assertNotIn("GITEA_MERGER_WORKTREE", unconsumed)
+        self.assertNotIn("GITEA_MCP_SESSION_STATE_TTL_HOURS", unconsumed)
+        self.assertIn("GITEA_UNRECOGNIZED_FOO_VAR", unconsumed)
+
+
+class TestIssue973CrossRepoCanonicalRoots(unittest.TestCase):
+    """Test workspace binding and canonical root validation for cross-repo namespaces."""
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp_dir = os.path.realpath(self._tmp.name)
+
+        # Create simulated installation root
+        self.install_root = os.path.join(self.tmp_dir, "Gitea-Tools")
+        os.makedirs(self.install_root)
+        subprocess.run(["git", "init", "-b", "master"], cwd=self.install_root, check=True)
+        subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.install_root, check=True)
+        subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.install_root, check=True)
+        with open(os.path.join(self.install_root, "README.md"), "w") as f:
+            f.write("install\n")
+        subprocess.run(["git", "add", "README.md"], cwd=self.install_root, check=True)
+        subprocess.run(["git", "commit", "-m", "initial"], cwd=self.install_root, check=True)
+
+        # Create simulated target repository root
+        self.target_root = os.path.join(self.tmp_dir, "mcp-control-plane")
+        os.makedirs(self.target_root)
+        subprocess.run(["git", "init", "-b", "master"], cwd=self.target_root, check=True)
+        subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.target_root, check=True)
+        subprocess.run(["git", "config", "user.name", "Test User"], cwd=self.target_root, check=True)
+        with open(os.path.join(self.target_root, "README.md"), "w") as f:
+            f.write("target\n")
+        subprocess.run(["git", "add", "README.md"], cwd=self.target_root, check=True)
+        subprocess.run(["git", "commit", "-m", "initial"], cwd=self.target_root, check=True)
+
+        # Add remotes to simulate real git repositories with identities
+        subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"], cwd=self.install_root, check=True)
+        subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git"], cwd=self.target_root, check=True)
+
+        # Create simulated foreign repository root
+        self.evil_root = os.path.join(self.tmp_dir, "Evil-Repo")
+        os.makedirs(self.evil_root)
+        subprocess.run(["git", "init", "-b", "master"], cwd=self.evil_root, check=True)
+        subprocess.run(["git", "config", "user.email", "[email protected]"], cwd=self.evil_root, check=True)
+        subprocess.run(["git", "config", "user.name", "Evil User"], cwd=self.evil_root, check=True)
+        with open(os.path.join(self.evil_root, "README.md"), "w") as f:
+            f.write("evil\n")
+        subprocess.run(["git", "add", "README.md"], cwd=self.evil_root, check=True)
+        subprocess.run(["git", "commit", "-m", "initial"], cwd=self.evil_root, check=True)
+        subprocess.run(["git", "remote", "add", "prgs", "https://gitea.prgs.cc/Someone-Else/Evil-Repo.git"], cwd=self.evil_root, check=True)
+
+        # Create branches/ directory and a valid registered worktree in target repository
+        self.target_branches = os.path.join(self.target_root, "branches")
+        self.target_worktree = os.path.join(self.target_branches, "rev-pr-99")
+        subprocess.run(["git", "worktree", "add", "-b", "rev-pr-99", self.target_worktree], cwd=self.target_root, check=True)
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_valid_same_repository_configuration(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="reviewer",
+            worktree_path=None,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=None,
+        )
+        self.assertEqual(ctx["canonical_repo_root"], self.install_root)
+        self.assertTrue(ctx["roots_aligned"])
+        self.assertTrue(ctx["canonical_root_assessment"]["proven"])
+
+    def test_valid_cross_repo_canonical_root(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="reviewer",
+            worktree_path=self.target_worktree,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=self.target_root,
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+        )
+        self.assertEqual(ctx["canonical_repo_root"], self.target_root)
+        self.assertTrue(ctx["roots_aligned"])
+        self.assertTrue(ctx["canonical_root_assessment"]["proven"])
+
+    def test_expected_repository_identity_match(self):
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="test",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install_root,
+            remote="prgs",
+        )
+        self.assertTrue(assessment["proven"])
+        self.assertFalse(assessment["block"])
+
+    def test_foreign_repository_identity_mismatch(self):
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.evil_root,
+            source="test",
+            expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
+            process_project_root=self.install_root,
+            remote="prgs",
+        )
+        self.assertFalse(assessment["proven"])
+        self.assertTrue(assessment["block"])
+        self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"]))
+
+    def test_native_repository_binding_mismatch(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="reviewer",
+            worktree_path=self.target_worktree,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=self.evil_root,
+            expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
+            remote="prgs",
+        )
+        self.assertFalse(ctx["roots_aligned"])
+        self.assertFalse(ctx["canonical_root_assessment"]["proven"])
+        self.assertTrue(any("identity mismatch" in r for r in ctx["canonical_root_assessment"]["reasons"]))
+
+    def test_unpatched_foreign_configured_root_derives_expected_from_process_root_and_blocks(self):
+        """B8: Production path test where foreign configured root cannot self-authorize."""
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server, "_configured_canonical_root", return_value=(self.evil_root, "env")):
+            expected_slug = mcp_server._resolve_expected_repository_slug("prgs")
+            self.assertEqual(expected_slug, "Scaled-Tech-Consulting/Gitea-Tools")
+            assessment = crr.assess_canonical_repository_root(
+                configured_value=self.evil_root,
+                source="env",
+                expected_slug=expected_slug,
+                process_project_root=self.install_root,
+                remote="prgs",
+                require_binding=True,
+            )
+            self.assertFalse(assessment["proven"])
+            self.assertTrue(assessment["block"])
+            self.assertEqual(assessment["resolved_slug"], "Someone-Else/Evil-Repo")
+            self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"]))
+
+    def test_unpatched_valid_cross_repo_matching_session_context(self):
+        """B8: Valid cross-repo namespace matches when session context is bound to target repo."""
+        bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"}
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx):
+            expected_slug = mcp_server._resolve_expected_repository_slug("prgs")
+            self.assertEqual(expected_slug, "Scaled-Tech-Consulting/mcp-control-plane")
+            assessment = crr.assess_canonical_repository_root(
+                configured_value=self.target_root,
+                source="env",
+                expected_slug=expected_slug,
+                process_project_root=self.install_root,
+                remote="prgs",
+                require_binding=True,
+            )
+            self.assertTrue(assessment["proven"])
+            self.assertFalse(assessment["block"])
+            self.assertEqual(assessment["resolved_slug"], "Scaled-Tech-Consulting/mcp-control-plane")
+
+    def test_unprovable_expected_identity_fails_closed(self):
+        """B8: If expected repository identity is unprovable for a configured root, fail closed."""
+        no_remote_root = os.path.join(self.tmp_dir, "no-remote-process-root")
+        os.makedirs(no_remote_root)
+        subprocess.run(["git", "init", "-b", "master"], cwd=no_remote_root, check=True)
+        with patch.object(mcp_server, "PROJECT_ROOT", no_remote_root), \
+             patch.object(mcp_server.session_ctx, "get_session_context", return_value=None):
+            expected_slug = mcp_server._resolve_expected_repository_slug("prgs")
+            self.assertIsNone(expected_slug)
+            assessment = crr.assess_canonical_repository_root(
+                configured_value=self.target_root,
+                source="env",
+                expected_slug=expected_slug,
+                process_project_root=no_remote_root,
+                remote="prgs",
+                require_binding=True,
+            )
+            self.assertFalse(assessment["proven"])
+            self.assertTrue(assessment["block"])
+            self.assertTrue(any("unprovable or missing" in r for r in assessment["reasons"]))
+
+    def test_missing_canonical_root(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="author",
+            worktree_path=None,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root="",
+        )
+        self.assertEqual(ctx["canonical_repo_root"], self.install_root)
+        self.assertTrue(ctx["roots_aligned"])
+
+    def test_nonexistent_configured_canonical_root(self):
+        nonexistent = os.path.join(self.tmp_dir, "nonexistent-repo")
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="reviewer",
+            worktree_path=self.target_worktree,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=nonexistent,
+        )
+        self.assertFalse(ctx["roots_aligned"])
+        self.assertFalse(ctx["canonical_root_assessment"]["proven"])
+        self.assertTrue(any("does not exist" in r for r in ctx["canonical_root_assessment"]["reasons"]))
+
+    def test_non_git_configured_canonical_root(self):
+        non_git = os.path.join(self.tmp_dir, "non-git-dir")
+        os.makedirs(non_git)
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="reviewer",
+            worktree_path=self.target_worktree,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=non_git,
+        )
+        self.assertFalse(ctx["roots_aligned"])
+        self.assertFalse(ctx["canonical_root_assessment"]["proven"])
+        self.assertTrue(any("not a git repository" in r for r in ctx["canonical_root_assessment"]["reasons"]))
+
+    def test_git_common_directory_membership_matching(self):
+        valid, err = nwb.verify_git_common_directory_membership(
+            self.target_worktree, self.target_root
+        )
+        self.assertTrue(valid, err)
+        self.assertIsNone(err)
+
+    def test_foreign_git_common_directory(self):
+        # Foreign worktree created under install_root
+        install_branches = os.path.join(self.install_root, "branches")
+        foreign_wt = os.path.join(install_branches, "foreign-wt")
+        subprocess.run(["git", "worktree", "add", "-b", "foreign-wt", foreign_wt], cwd=self.install_root, check=True)
+
+        valid, err = nwb.verify_git_common_directory_membership(
+            foreign_wt, self.target_root
+        )
+        self.assertFalse(valid)
+        self.assertIn("does not match", err)
+
+    def test_normalized_path_aliases(self):
+        alias_path = self.target_worktree + "/../rev-pr-99/./"
+        valid, err = nwb.verify_git_common_directory_membership(
+            alias_path, self.target_root
+        )
+        self.assertTrue(valid, err)
+
+    def test_safe_symlink_identity(self):
+        link_path = os.path.join(self.target_branches, "symlink-rev-99")
+        try:
+            os.symlink(self.target_worktree, link_path)
+            valid, err = nwb.verify_git_common_directory_membership(
+                link_path, self.target_root
+            )
+            self.assertTrue(valid, err)
+        finally:
+            if os.path.exists(link_path):
+                os.unlink(link_path)
+
+    def test_symlink_escape_or_foreign_alias(self):
+        outside_dir = os.path.join(self.tmp_dir, "outside-target")
+        os.makedirs(outside_dir)
+        link_escape = os.path.join(self.target_branches, "escape-link")
+        try:
+            os.symlink(outside_dir, link_escape)
+            assessment = nwb.assess_namespace_mutation_workspace(
+                role_kind="reviewer",
+                worktree_path=link_escape,
+                worktree=None,
+                process_project_root=self.install_root,
+                env={},
+                configured_canonical_root=self.target_root,
+                expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            )
+            self.assertTrue(assessment["block"])
+        finally:
+            if os.path.exists(link_escape):
+                os.unlink(link_escape)
+
+    def test_reviewer_worktree_registered_and_valid(self):
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="reviewer",
+            worktree_path=self.target_worktree,
+            worktree=None,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=self.target_root,
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+        )
+        self.assertFalse(assessment["block"])
+
+    def test_reviewer_worktree_unregistered_blocks(self):
+        unreg_wt = os.path.join(self.target_branches, "unregistered-reviewer")
+        os.makedirs(unreg_wt)
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="reviewer",
+            worktree_path=unreg_wt,
+            worktree=None,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=self.target_root,
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+        )
+        self.assertTrue(assessment["block"])
+        self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"]))
+
+    def test_merger_worktree_registered_and_valid(self):
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="merger",
+            worktree_path=self.target_worktree,
+            worktree=None,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=self.target_root,
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+        )
+        self.assertFalse(assessment["block"])
+
+    def test_merger_worktree_unregistered_blocks(self):
+        unreg_wt = os.path.join(self.target_branches, "unregistered-merger")
+        os.makedirs(unreg_wt)
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="merger",
+            worktree_path=unreg_wt,
+            worktree=None,
+            process_project_root=self.install_root,
+            env={},
+            configured_canonical_root=self.target_root,
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+        )
+        self.assertTrue(assessment["block"])
+        self.assertTrue(any("is not registered in git worktree list" in r for r in assessment["reasons"]))
+
+    def test_reviewer_or_merger_worktree_outside_branches_blocks(self):
+        outside_wt = os.path.join(self.target_root, "outside_branches_wt")
+        os.makedirs(outside_wt)
+        for r_kind in ("reviewer", "merger"):
+            assessment = nwb.assess_namespace_mutation_workspace(
+                role_kind=r_kind,
+                worktree_path=outside_wt,
+                worktree=None,
+                process_project_root=self.install_root,
+                env={},
+                configured_canonical_root=self.target_root,
+                expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            )
+            self.assertTrue(assessment["block"])
+            self.assertTrue(any("is not under" in r for r in assessment["reasons"]))
+
+    def test_nonexistent_reviewer_or_merger_worktree_blocks(self):
+        nonexistent_wt = os.path.join(self.target_branches, "nonexistent-wt")
+        for r_kind in ("reviewer", "merger"):
+            assessment = nwb.assess_namespace_mutation_workspace(
+                role_kind=r_kind,
+                worktree_path=nonexistent_wt,
+                worktree=None,
+                process_project_root=self.install_root,
+                env={},
+                configured_canonical_root=self.target_root,
+                expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            )
+            self.assertTrue(assessment["block"])
+            self.assertTrue(any("does not exist" in r for r in assessment["reasons"]))
+
+    def test_safe_and_unsafe_mutation_alignment_outcomes(self):
+        # Safe alignment (same repo)
+        report_safe = stable_control_runtime.build_runtime_report(
+            process_root=self.install_root,
+            checkout_branch="master",
+            runtime_head="abcdef123456",
+            active_task_workspace=self.install_root,
+            canonical_repository_root=self.install_root,
+            workspace_roots_aligned=True,
+        )
+        gate_safe = stable_control_runtime.assess_runtime_mutation_gate(report_safe)
+        self.assertFalse(gate_safe["block"])
+
+        # Unsafe alignment
+        report_unsafe = stable_control_runtime.build_runtime_report(
+            process_root=self.install_root,
+            checkout_branch="master",
+            runtime_head="abcdef123456",
+            active_task_workspace=self.target_worktree,
+            canonical_repository_root=self.target_root,
+            workspace_roots_aligned=False,
+        )
+        gate_unsafe = stable_control_runtime.assess_runtime_mutation_gate(report_unsafe)
+        self.assertTrue(gate_unsafe["block"])
+
+    def test_reviewer_lease_lifecycle_production_path(self):
+        """B5: Automated regression for reviewer lease acquire and release through production path."""
+        mock_whoami = {
+            "authenticated": True,
+            "username": "sysadmin",
+            "remote": "prgs",
+            "profile": {
+                "profile_name": "prgs-reviewer",
+                "role": "reviewer",
+                "role_kind": "reviewer",
+                "allowed_operations": ["gitea.read", "gitea.pr.comment", "gitea.pr.approve", "gitea.pr.request_changes"],
+                "forbidden_operations": [],
+            },
+        }
+
+        def mock_api_request(method, url, auth=None, json_data=None):
+            if method == "GET":
+                return {"number": 99, "head": {"sha": "abc1234"}, "state": "open", "merged": False, "merged_at": None}
+            elif method == "POST":
+                return {"id": 9999, "body": (json_data or {}).get("body", "")}
+            return {}
+
+        with patch.object(mcp_server, "gitea_whoami", return_value=mock_whoami), \
+             patch.object(mcp_server, "get_profile", return_value=mock_whoami["profile"]), \
+             patch.object(mcp_server, "_effective_workspace_role", return_value="reviewer"), \
+             patch.object(mcp_server, "_configured_canonical_root", return_value=(self.target_root, "env")), \
+             patch.object(mcp_server, "_reviewer_session_worktree", return_value=self.target_worktree), \
+             patch.object(mcp_server, "_auth", return_value="token mock-token"), \
+             patch.object(mcp_server, "_fetch_pr_comments", return_value=[]), \
+             patch.object(mcp_server, "api_request", side_effect=mock_api_request), \
+             patch("reviewer_pr_lease.assess_acquire_lease", return_value={"acquire_allowed": True, "reasons": [], "lease_body": "<!-- LEASE -->"}), \
+             patch("reviewer_pr_lease.find_active_reviewer_lease", return_value={"session_id": "sid-123", "reviewer": "sysadmin"}), \
+             patch("reviewer_pr_lease.get_session_lease", return_value={"session_id": "sid-123", "reviewer": "sysadmin"}), \
+             patch("reviewer_pr_lease.clear_session_lease") as mock_clear:
+
+            acq_res = mcp_server.gitea_acquire_reviewer_pr_lease(
+                pr_number=99,
+                remote="prgs",
+                worktree=self.target_worktree,
+                org="Scaled-Tech-Consulting",
+                repo="mcp-control-plane",
+            )
+            self.assertTrue(acq_res.get("success"), acq_res)
+
+            rel_res = mcp_server.gitea_release_reviewer_pr_lease(
+                pr_number=99,
+                worktree=self.target_worktree,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="mcp-control-plane",
+            )
+            self.assertTrue(rel_res.get("success"), rel_res)
+            mock_clear.assert_called_once()
+
+    def test_unbound_session_derives_and_retains_configured_target_repository(self):
+        """B9: Unbound session with a legitimate configured target root derives and retains target repository."""
+        prof = {
+            "profile_name": "prgs-author",
+            "allowed_repositories": ["Scaled-Tech-Consulting/mcp-control-plane"],
+            "canonical_repository_root": self.target_root,
+        }
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server.session_ctx, "get_session_context", return_value=None):
+            slug, reasons = mcp_server._canonical_repository_slug(prof, "prgs")
+            self.assertEqual(slug, "Scaled-Tech-Consulting/mcp-control-plane", f"reasons: {reasons}")
+            self.assertEqual(reasons, [])
+            res = mcp_server._trusted_session_repository(prof, "prgs")
+            self.assertEqual(res["org"], "Scaled-Tech-Consulting")
+            self.assertEqual(res["repository"], "mcp-control-plane")
+
+    def test_process_root_a_configured_target_b_retains_b(self):
+        """B9: Process root A (Gitea-Tools) and configured target B (mcp-control-plane) keep B as expected identity when bound."""
+        bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"}
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx):
+            expected = mcp_server._resolve_expected_repository_slug("prgs")
+            self.assertEqual(expected, "Scaled-Tech-Consulting/mcp-control-plane")
+
+    def test_request_supplied_repository_c_cannot_replace_derived_or_bound_repository_b(self):
+        """B9: Request-supplied repository C (Timesheet) cannot replace bound repository B (mcp-control-plane)."""
+        bound_ctx = {"org": "Scaled-Tech-Consulting", "repository": "mcp-control-plane", "remote": "prgs"}
+        with patch.object(mcp_server, "PROJECT_ROOT", self.install_root), \
+             patch.object(mcp_server.session_ctx, "get_session_context", return_value=bound_ctx):
+            expected = mcp_server._resolve_expected_repository_slug("prgs", org="Scaled-Tech-Consulting", repo="Timesheet")
+            self.assertEqual(expected, "Scaled-Tech-Consulting/mcp-control-plane")
+
+    def test_known_expected_repo_a_plus_candidate_root_b_fails_closed(self):
+        """B9: Known expected repository A plus candidate root B fails closed in validation mode."""
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="env",
+            expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode="validation",
+        )
+        self.assertFalse(assessment["proven"])
+        self.assertTrue(assessment["block"])
+        self.assertTrue(any("identity mismatch" in r for r in assessment["reasons"]))
+
+    def test_validation_mode_with_unprovable_expected_identity_fails_closed(self):
+        """B9: Validation mode with expected_slug=None and require_binding=True fails closed."""
+        assessment = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="env",
+            expected_slug=None,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+            mode="validation",
+        )
+        self.assertFalse(assessment["proven"])
+        self.assertTrue(assessment["block"])
+        self.assertTrue(any("unprovable or missing" in r for r in assessment["reasons"]))
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_namespace_workspace_binding.py

@@ -245,10 +245,15 @@ class TestNamespaceWorkspaceIntegration(unittest.TestCase):
     def test_pr487_style_merge_binds_clean_merger_workspace(
         self, _exists, _isdir, mock_run
     ):
-        mock_run.return_value = MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+        def mock_git(cmd, *args, **kwargs):
+            if "rev-parse" in cmd:
+                return MagicMock(returncode=0, stdout=f"{CONTROL_ROOT}/.git\n")
+            return MagicMock(returncode=0, stdout=f"worktree {CONTROL_ROOT}\nworktree {MERGER_CLEAN}\n")
+        mock_run.side_effect = mock_git
         os.environ[nwb.AUTHOR_WORKTREE_ENV] = AUTHOR_DIRTY
+        os.environ[nwb.MERGER_WORKTREE_ENV] = MERGER_CLEAN
         srv._preflight_resolved_role = "reviewer"
         with mock.patch.object(srv, "PROJECT_ROOT", MCP_PROCESS_ROOT):
             with mock.patch("gitea_mcp_server.get_profile", return_value=self._merger_profile()):
                 resolved = srv._verify_role_mutation_workspace("prgs")
-        self.assertEqual(resolved, os.path.realpath(MCP_PROCESS_ROOT))
+        self.assertEqual(resolved, os.path.realpath(MERGER_CLEAN))

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",
     }