fix(runtime): recognize heartbeat interval env and wire production lifecycle tests

Address REQUEST_CHANGES review #652 on PR #976 (issue #975).

Blocker 1: name GITEA_WORKER_HEARTBEAT_INTERVAL_SECONDS individually in
RECOGNIZED_GITEA_ENV_KEYS so the production-consumed operator override is no
longer classified as unsupported-env / runtime_reconnect_required. No prefix
broadening; unknown overrides remain fail-closed.

Blocker 2: add focused production-path tests that call _active_worker_identity
and _start_worker_heartbeat, proving supervisor attachment, identity/session/
generation/pid/epoch agreement, no duplicates, failed/fenced paths, orderly
shutdown, configured interval, and off-loop sqlite heartbeats.

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,19 +295,36 @@ def assess_canonical_repository_root(
 
     resolved_slug = repository_identity_slug(toplevel, remote=remote)
     reasons: list[str] = []
-    expected = (expected_slug or "").strip() or None
+
-    if expected:
+    if mode == MODE_DERIVATION:
-        if resolved_slug and resolved_slug.lower() != expected.lower():
+        if not resolved_slug:
             reasons.append(
-                f"canonical repository root identity mismatch: '{toplevel}' resolves "
+                f"configured canonical repository root '{toplevel}' has no resolvable "
-                f"to repository '{resolved_slug}' but the session is authorized for "
+                "git remote identity (fail closed)"
-                f"'{expected}' (forged or conflicting binding, fail closed)"
             )
-        elif not resolved_slug and require_binding:
+    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():
+                reasons.append(
+                    f"canonical repository root identity mismatch: '{toplevel}' resolves "
+                    f"to repository '{resolved_slug}' but the session is authorized for "
+                    f"'{expected}' (forged or conflicting binding, fail closed)"
+                )
+            elif not resolved_slug and require_binding:
+                reasons.append(
+                    f"canonical repository root '{toplevel}' has no resolvable git "
+                    f"remote identity to confirm authorization for '{expected}' "
+                    "(fail closed)"
+                )
+        elif require_binding:
             reasons.append(
-                f"canonical repository root '{toplevel}' has no resolvable git "
+                f"canonical repository root '{toplevel}' has configured value "
-                f"remote identity to confirm authorization for '{expected}' "
+                f"'{configured_value}' but authoritative expected repository identity "
-                "(fail closed)"
+                "is unprovable or missing (fail closed)"
             )
 
     return _assessment(
@@ -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

@@ -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
@@ -1913,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,
         *,
@@ -3051,3 +3229,123 @@ class ControlPlaneDB:
             "live_lease_id": None if live_lease_id is None else str(live_lease_id),
             "reconcile_action": "reconcile_required" if stale else "safe_to_resume",
         }
+
+    def retire_session_checkpoint_worktree_path(
+        self,
+        session_id: str,
+        *,
+        checkpoint_id: str | None = None,
+        expected_path: str | None = None,
+        expected_status: str | None = None,
+        reason: str = "missing_worktree_path_retired",
+    ) -> dict[str, Any]:
+        """Retire a missing worktree_path from session_checkpoints (#970).
+
+        Compare-and-swap, mirroring :meth:`retire_lease_worktree_path` (#970
+        review 644 B3). ``expected_path`` names the exact stored path the caller
+        audited; the guarded ``UPDATE`` is keyed on that stored value, so a
+        checkpoint whose path was moved, replaced, or concurrently rewritten
+        after the audit is refused without mutation rather than blindly cleared.
+
+        Exactly one checkpoint row is targeted: by ``checkpoint_id`` when given,
+        otherwise by ``session_id``, which must identify a single row.
+        """
+        now_s = _ts()
+        expected_norm = _realpath_or_raw(expected_path)
+        if not expected_norm:
+            raise ControlPlaneError(
+                "cannot retire session checkpoint worktree_path: expected_path "
+                "is required for compare-and-swap retirement (fail closed)"
+            )
+        if not checkpoint_id and not (session_id or "").strip():
+            raise ControlPlaneError(
+                "cannot retire session checkpoint worktree_path: checkpoint_id "
+                "or session_id is required (fail closed)"
+            )
+
+        with self._tx() as conn:
+            if checkpoint_id:
+                selector_sql = "SELECT * FROM session_checkpoints WHERE checkpoint_id = ?"
+                selector_params: tuple[Any, ...] = (checkpoint_id,)
+                selector_desc = f"checkpoint_id '{checkpoint_id}'"
+            else:
+                selector_sql = "SELECT * FROM session_checkpoints WHERE session_id = ?"
+                selector_params = (session_id,)
+                selector_desc = f"session_id '{session_id}'"
+
+            rows = [dict(r) for r in conn.execute(selector_sql, selector_params).fetchall()]
+            if not rows:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path: no "
+                    f"checkpoint matches {selector_desc} (fail closed)"
+                )
+            if len(rows) > 1:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path: "
+                    f"{selector_desc} matches {len(rows)} checkpoints; supply an "
+                    "exact checkpoint_id (fail closed)"
+                )
+
+            record = rows[0]
+            target_checkpoint_id = record.get("checkpoint_id")
+            current_wt = (record.get("worktree_path") or "").strip()
+
+            if not current_wt:
+                # Idempotent: the binding this caller audited is already gone.
+                return {
+                    "session_id": session_id,
+                    "checkpoint_id": target_checkpoint_id,
+                    "retired": False,
+                    "already_retired": True,
+                    "prior_worktree_path": "",
+                    "expected_worktree_path": expected_path,
+                    "reason": reason,
+                    "compare_and_swap": {
+                        "matched": True,
+                        "outcome": "already_retired",
+                    },
+                }
+
+            if _realpath_or_raw(current_wt) != expected_norm:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path for "
+                    f"{selector_desc}: expected '{expected_path}' does not match "
+                    f"current '{current_wt}' (fail closed)"
+                )
+
+            if expected_status is not None:
+                current_status = record.get("status")
+                if str(current_status or "").strip() != str(expected_status).strip():
+                    raise ControlPlaneError(
+                        f"cannot retire session checkpoint worktree_path for "
+                        f"{selector_desc}: status changed since audit (expected "
+                        f"'{expected_status}', found '{current_status}'); fail closed"
+                    )
+
+            cur = conn.execute(
+                "UPDATE session_checkpoints SET worktree_path = '', updated_at = ? "
+                "WHERE checkpoint_id = ? AND worktree_path = ?",
+                (now_s, target_checkpoint_id, record.get("worktree_path")),
+            )
+            if cur.rowcount != 1:
+                raise ControlPlaneError(
+                    f"cannot retire session checkpoint worktree_path for "
+                    f"{selector_desc}: compare-and-swap matched no row "
+                    "(concurrent change); fail closed"
+                )
+
+            return {
+                "session_id": session_id,
+                "checkpoint_id": target_checkpoint_id,
+                "retired": True,
+                "already_retired": False,
+                "prior_worktree_path": current_wt,
+                "expected_worktree_path": expected_path,
+                "retired_at": now_s,
+                "reason": reason,
+                "compare_and_swap": {
+                    "matched": True,
+                    "outcome": "retired",
+                    "expected_status": expected_status,
+                },
+            }

docs/mcp-tool-inventory.md

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

docs/remote-mcp/threat-model-anchors.json

@@ -8,10 +8,10 @@
     "document. #930's inventory had no such guard and its gitea_mcp_server.py",
     "anchors drifted between 7bf4f125 and aad5c8b4."
   ],
-  "generated_against_commit": "ca5f078d8a575ea3e2991771f8b4ea85e3dcaaa0",
+  "generated_against_commit": "1dd30ecb1508b559868c2d5d94367bc055d5138e",
   "anchors": [
     {
-      "anchor": "gitea_mcp_server.py:24864",
+      "anchor": "gitea_mcp_server.py:25087",
       "expect": "mcp_daemon_guard.bind_native_mcp_transport()"
     },
     {
@@ -27,11 +27,11 @@
       "expect": "def assess_transport_for_auth_mint"
     },
     {
-      "anchor": "gitea_mcp_server.py:9191",
+      "anchor": "gitea_mcp_server.py:9192",
       "expect": "assess_transport_for_auth_mint()"
     },
     {
-      "anchor": "gitea_mcp_server.py:9440",
+      "anchor": "gitea_mcp_server.py:9441",
       "expect": "assess_transport_for_auth_mint()"
     },
     {
@@ -39,31 +39,31 @@
       "expect": "The transport is selected by deployment configuration"
     },
     {
-      "anchor": "gitea_mcp_server.py:15481",
+      "anchor": "gitea_mcp_server.py:15630",
       "expect": "def _is_client_managed_process"
     },
     {
-      "anchor": "gitea_mcp_server.py:15511",
+      "anchor": "gitea_mcp_server.py:15644",
       "expect": "def _provenance_mutation_block"
     },
     {
-      "anchor": "gitea_mcp_server.py:15519",
+      "anchor": "gitea_mcp_server.py:15652",
       "expect": "unsupported_manual_launch"
     },
     {
-      "anchor": "gitea_mcp_server.py:19070",
+      "anchor": "gitea_mcp_server.py:19217",
       "expect": "server_provenance"
     },
     {
-      "anchor": "gitea_mcp_server.py:21581",
+      "anchor": "gitea_mcp_server.py:21741",
       "expect": "def _check_mcp_runtimes_diagnostics"
     },
     {
-      "anchor": "gitea_mcp_server.py:21601",
+      "anchor": "gitea_mcp_server.py:21761",
       "expect": "\"ps\", \"-o\", \"pid,lstart,command\""
     },
     {
-      "anchor": "gitea_mcp_server.py:21645",
+      "anchor": "gitea_mcp_server.py:21805",
       "expect": "\"ps\", \"eww\""
     },
     {
@@ -107,19 +107,19 @@
       "expect": "def assert_keychain_access_allowed"
     },
     {
-      "anchor": "gitea_mcp_server.py:19327",
+      "anchor": "gitea_mcp_server.py:19487",
       "expect": "def gitea_list_profiles"
     },
     {
-      "anchor": "gitea_mcp_server.py:19378",
+      "anchor": "gitea_mcp_server.py:19538",
       "expect": "gitea_config.resolve_token(p)"
     },
     {
-      "anchor": "gitea_mcp_server.py:19691",
+      "anchor": "gitea_mcp_server.py:19851",
       "expect": "def gitea_audit_config"
     },
     {
-      "anchor": "gitea_mcp_server.py:19713",
+      "anchor": "gitea_mcp_server.py:19873",
       "expect": "service_summaries(config)"
     },
     {
@@ -135,19 +135,19 @@
       "expect": "_keychain_token(auth.get(\"id\"))"
     },
     {
-      "anchor": "gitea_mcp_server.py:17776",
+      "anchor": "gitea_mcp_server.py:17909",
       "expect": "\"jenkins-mcp\""
     },
     {
-      "anchor": "gitea_mcp_server.py:17782",
+      "anchor": "gitea_mcp_server.py:17915",
       "expect": "external-mcp"
     },
     {
-      "anchor": "gitea_mcp_server.py:17803",
+      "anchor": "gitea_mcp_server.py:17936",
       "expect": "\"glitchtip-mcp\""
     },
     {
-      "anchor": "gitea_mcp_server.py:17808",
+      "anchor": "gitea_mcp_server.py:17941",
       "expect": "external-mcp"
     },
     {
@@ -183,7 +183,7 @@
       "expect": "mutation_safe"
     },
     {
-      "anchor": "gitea_mcp_server.py:19171",
+      "anchor": "gitea_mcp_server.py:19331",
       "expect": "def gitea_assess_master_parity"
     },
     {
@@ -195,11 +195,11 @@
       "expect": "AUTHOR_WORKTREE_ENV"
     },
     {
-      "anchor": "gitea_mcp_server.py:2351",
+      "anchor": "gitea_mcp_server.py:2352",
       "expect": "/tmp/gitea_issue_lock.json"
     },
     {
-      "anchor": "gitea_mcp_server.py:10956",
+      "anchor": "gitea_mcp_server.py:10957",
       "expect": "def gitea_bootstrap_author_issue_worktree"
     },
     {
@@ -239,7 +239,7 @@
       "expect": "os.getpid()"
     },
     {
-      "anchor": "gitea_mcp_server.py:12870",
+      "anchor": "gitea_mcp_server.py:12871",
       "expect": "owner_pid_alive"
     }
   ]

docs/remote-mcp/threat-model.md

@@ -5,7 +5,7 @@ What the adversary is, what each boundary protects, and which services may share
 - **Issue:** #956 (Remote-MCP threat model), child of epic #929, cross-linked to #955.
 - **Depends on:** #930 (closed) — `docs/remote-mcp/coupling-inventory.md`.
 - **Blocks:** #932, #933, #934, #938.
-- **Generated against commit:** `ca5f078d8a575ea3e2991771f8b4ea85e3dcaaa0` (#708's
+- **Generated against commit:** `1dd30ecb1508b559868c2d5d94367bc055d5138e` (#708's
   namespace-attachment gate). Originally generated against
   `aad5c8b42361d380a8eeb07b94b90815e594c2c5` (`master`), re-anchored at
   `a143cd065ba06e1a2bdc5143a19ec156e53650ef` when #931's transport bind seam shifted the
@@ -31,7 +31,7 @@ document cites an anchor the fixture does not cover.
 
 This guard exists because #930 did not have one. Its inventory was generated at
 `7bf4f125`; by `aad5c8b4` its `gitea_mcp_server.py` anchors had drifted — the transport
-bind it cited at line 23750 now lives at `gitea_mcp_server.py:24864`, and its
+bind it cited at line 23750 now lives at `gitea_mcp_server.py:25087`, and its
 client-managed provenance anchor at 14588 now lands in an unrelated function. Nothing
 failed, because nothing checked. Anchors into a ~24,700-line module rot silently, and a
 security document that cannot prove its own citations is worse than none, because it is
@@ -75,19 +75,19 @@ authenticate the *caller*, not the *intent*.
 ## 3. Trust boundaries
 
 "Crossing requires today" is what the code actually enforces at
-`ca5f078d8a575ea3e2991771f8b4ea85e3dcaaa0`, not what the design intends.
+`1dd30ecb1508b559868c2d5d94367bc055d5138e`, not what the design intends.
 
 | ID | Boundary | Protects | Crossing requires today | Crossing must require remotely |
 | -- | -------- | -------- | ----------------------- | ------------------------------ |
-| B1 | LLM client ↔ MCP server session | A1, A3, A10 — that a mutating session was established through the sanctioned client path | A single configured bind (`gitea_mcp_server.py:24864`) validated against one closed allowlist (`mcp_daemon_guard.py:49`, `mcp_daemon_guard.py:195`) — since #931 the identifier comes from deployment configuration and defaults to the local transport, so the boundary no longer rests on a literal, but it still rests on the *bind* rather than on an authenticated caller; client-managed provenance (`gitea_mcp_server.py:15481`) or a refusal (`gitea_mcp_server.py:15519`); production transport before recovery-authorization mint (`irrecoverable_provenance.py:497`, consumed at `gitea_mcp_server.py:9191` and `gitea_mcp_server.py:9440`) | An authenticated handshake issuing a server-side session identity bound to a principal, with the transport recorded in provenance. The physical proof (a pipe) must become a cryptographic one. |
+| B1 | LLM client ↔ MCP server session | A1, A3, A10 — that a mutating session was established through the sanctioned client path | A single configured bind (`gitea_mcp_server.py:25087`) validated against one closed allowlist (`mcp_daemon_guard.py:49`, `mcp_daemon_guard.py:195`) — since #931 the identifier comes from deployment configuration and defaults to the local transport, so the boundary no longer rests on a literal, but it still rests on the *bind* rather than on an authenticated caller; client-managed provenance (`gitea_mcp_server.py:15630`) or a refusal (`gitea_mcp_server.py:15652`); production transport before recovery-authorization mint (`irrecoverable_provenance.py:497`, consumed at `gitea_mcp_server.py:9192` and `gitea_mcp_server.py:9441`) | An authenticated handshake issuing a server-side session identity bound to a principal, with the transport recorded in provenance. The physical proof (a pipe) must become a cryptographic one. |
 | B2 | Role ↔ role | A9 — that author, reviewer, merger, and reconciler are distinct authorities | **The process boundary only.** The role is a property of the process, read once from `GITEA_MCP_PROFILE` (`gitea_config.py:54`). A caller gets author permissions by connecting to the author process. Review and merge are the operations singled out for extra care (`gitea_config.py:97`) | A per-request principal, so the role follows from the credential presented and cannot be selected by reaching a different endpoint. |
 | B3 | MCP server ↔ credential store | A3, A8 — that only sanctioned code turns a profile into a token | `_keychain_token` shelling out to the login keychain (`gitea_config.py:956`), dispatched by `resolve_token` (`gitea_config.py:974`) with the reference type built at `gitea_config.py:1015`, gated by `assert_keychain_access_allowed` (`mcp_daemon_guard.py:583`). Inline secrets are rejected at config load (`gitea_config.py:294`) | A credential provider keyed by the *request* principal, returning only that principal's credential, with the source recorded and the value never returned. |
 | B4 | MCP server ↔ Gitea | A1, A2 — that only authorized calls reach the forge | A bearer token over TLS. Server-side, nothing distinguishes one role's token from another beyond the account it belongs to | Unchanged at the forge; the endpoint in front of it must refuse unauthenticated and plaintext connections before tool dispatch. |
-| B5 | MCP server ↔ caller's filesystem | A7 — that a tool acts on the *caller's* disk or refuses | Nothing. The server's disk *is* the caller's disk. Worktree bootstrap writes directly (`gitea_mcp_server.py:10956`); the active workspace is process-global (`gitea_mcp_server.py:193`, `gitea_mcp_server.py:194`) | An explicit per-tool classification, enforced at dispatch, refusing filesystem tools over a transport that cannot reach the caller's disk. A green verdict about the wrong disk is the failure to prevent. |
+| B5 | MCP server ↔ caller's filesystem | A7 — that a tool acts on the *caller's* disk or refuses | Nothing. The server's disk *is* the caller's disk. Worktree bootstrap writes directly (`gitea_mcp_server.py:10957`); the active workspace is process-global (`gitea_mcp_server.py:193`, `gitea_mcp_server.py:194`) | An explicit per-tool classification, enforced at dispatch, refusing filesystem tools over a transport that cannot reach the caller's disk. A green verdict about the wrong disk is the failure to prevent. |
-| B6 | MCP server ↔ coordination state | A6, A9 — mutual exclusion | Local files and a local SQLite database, with liveness judged from the local process table (`issue_lock_store.py:98`), keyed on paths under one user's home (`issue_lock_store.py:26`, `mcp_session_state.py:27`, `control_plane_db.py:47`) and on `os.getpid()` (`control_plane_db.py:1145`, `gitea_mcp_server.py:12870`). A legacy global slot still exists at `gitea_mcp_server.py:2351`, and the session-pointer file is named per PID (`issue_lock_store.py:83`) | One authority per ownership question, with liveness from session identity and expiry, and atomic acquire, renew, and release across hosts. |
+| B6 | MCP server ↔ coordination state | A6, A9 — mutual exclusion | Local files and a local SQLite database, with liveness judged from the local process table (`issue_lock_store.py:98`), keyed on paths under one user's home (`issue_lock_store.py:26`, `mcp_session_state.py:27`, `control_plane_db.py:47`) and on `os.getpid()` (`control_plane_db.py:1145`, `gitea_mcp_server.py:12871`). A legacy global slot still exists at `gitea_mcp_server.py:2352`, and the session-pointer file is named per PID (`issue_lock_store.py:83`) | One authority per ownership question, with liveness from session identity and expiry, and atomic acquire, renew, and release across hosts. |
 | B7 | Gitea integration ↔ unrelated integrations | A4, A5 — that a Gitea compromise is not a CI and observability compromise | **Nothing.** See §5. The Gitea server reads Jenkins and GlitchTip secrets (`gitea_config.py:851`, reached from `gitea_config.py:837`) and holds the Sentry token (`sentry_incident_bridge.py:190`) | A hard process boundary. This is the boundary #956 exists to create. |
 | B8 | Tenant ↔ tenant (`prgs` / `mdcps` / `local-lab`) | A2 — that one organization's compromise is not another's | Convention. One configuration declares all three contexts; `resolve_service` fails closed on a *disabled* context (`gitea_config.py:704`) but the credentials of enabled ones remain reachable in-process. A per-profile repository scope exists (`gitea_config.py:499`) | Separate deployments, or at minimum per-tenant credential scopes with no process able to resolve both. |
-| B9 | Deployed code ↔ merged policy | A1, A10 — that the running server enforces the rules that were actually merged | Comparing this process's startup commit against this disk (`master_parity_gate.py:168`), conjoined into a single verdict (`master_parity_gate.py:255`) published by `gitea_mcp_server.py:19171` | Freshness defined against the deployed build identity, with an explicit fail-closed verdict when undeterminable. |
+| B9 | Deployed code ↔ merged policy | A1, A10 — that the running server enforces the rules that were actually merged | Comparing this process's startup commit against this disk (`master_parity_gate.py:168`), conjoined into a single verdict (`master_parity_gate.py:255`) published by `gitea_mcp_server.py:19331` | Freshness defined against the deployed build identity, with an explicit fail-closed verdict when undeterminable. |
 
 ### What no boundary constrains
 
@@ -132,10 +132,10 @@ Two flows deserve attention because neither is obvious from the code:
 
 1. **The keychain flow fans out.** B3 is drawn once but resolves credentials for *every*
    configured profile and service, not only the active one. `gitea_list_profiles`
-   (`gitea_mcp_server.py:19327`) reports each profile's credential status by calling
+   (`gitea_mcp_server.py:19487`) reports each profile's credential status by calling
-   `resolve_token` on it (`gitea_mcp_server.py:19378`), and `gitea_audit_config`
+   `resolve_token` on it (`gitea_mcp_server.py:19538`), and `gitea_audit_config`
-   (`gitea_mcp_server.py:19691`) reports service credential status through
+   (`gitea_mcp_server.py:19851`) reports service credential status through
-   `service_summaries` (`gitea_mcp_server.py:19713`).
+   `service_summaries` (`gitea_mcp_server.py:19873`).
 2. **The return path is a flow too.** Content read from Gitea travels back into the model
    and is treated as instruction. This is the ADV2 edge, and it is the only edge in the
    diagram with no authentication on it, because it is not a request.
@@ -180,16 +180,16 @@ the credential, and an attacker holding the token does not call our tools.
 
 **Finding 3 — Any one role process can resolve every other role's credential.** This is not
 inferred; it is demonstrated by tool output. `gitea_list_profiles`
-(`gitea_mcp_server.py:19327`) called from the **author** session reports
+(`gitea_mcp_server.py:19487`) called from the **author** session reports
 `identity_status: "credentials present"` for `prgs-merger`, `prgs-reviewer`,
 `prgs-reconciler`, and every `mdcps` profile, because it calls `resolve_token` on each one
-(`gitea_mcp_server.py:19378`). The author process does not merely *have access to* the
+(`gitea_mcp_server.py:19538`). The author process does not merely *have access to* the
 merger's credential — it reads it to answer a status query. B2 is not a credential boundary
 in either direction.
 
 **Finding 4 — The Gitea server reads CI and observability secrets.** `gitea_audit_config`
-(`gitea_mcp_server.py:19691`) reports `MDCPS Jenkins: enabled, read-only, authenticated`.
+(`gitea_mcp_server.py:19851`) reports `MDCPS Jenkins: enabled, read-only, authenticated`.
-That word `authenticated` is produced by `service_summaries` (`gitea_mcp_server.py:19713`,
+That word `authenticated` is produced by `service_summaries` (`gitea_mcp_server.py:19873`,
 defined at `gitea_config.py:837`), whose default check calls `_keychain_token` on the
 service's own keychain reference (`gitea_config.py:851`). Producing that one line requires
 the Gitea MCP server to read the Jenkins secret and the GlitchTip secret out of the
@@ -197,8 +197,8 @@ keychain. B7 does not exist.
 
 **Finding 5 — Jenkins and GlitchTip are already decomposed; the reach is residual.** Their
 tools live in separately registered servers, marked `external-mcp`
-(`gitea_mcp_server.py:17776`, `gitea_mcp_server.py:17782`, `gitea_mcp_server.py:17803`,
+(`gitea_mcp_server.py:17909`, `gitea_mcp_server.py:17915`, `gitea_mcp_server.py:17936`,
-`gitea_mcp_server.py:17808`) with their own expected tool sets (`mcp_discoverability.py:9`,
+`gitea_mcp_server.py:17941`) with their own expected tool sets (`mcp_discoverability.py:9`,
 `mcp_discoverability.py:17`). The correct decomposition was already chosen. What remains is
 a leak across it: the credential *references* still live in the Gitea configuration and are
 still resolved by the Gitea process. #75 bundled these services into one control-plane
@@ -209,8 +209,8 @@ GlitchTip, the Sentry bridge runs *inside* the Gitea server, resolving its token
 process environment (`sentry_incident_bridge.py:190`) and sending it as a bearer header
 (`sentry_incident_bridge.py:289`). Being an environment variable rather than a keychain item
 makes it strictly worse: it needs no keychain prompt and is inherited by every subprocess the
-server spawns — including the `ps` invocations at `gitea_mcp_server.py:21601` and
+server spawns — including the `ps` invocations at `gitea_mcp_server.py:21761` and
-`gitea_mcp_server.py:21645`, reached from `gitea_mcp_server.py:21581`.
+`gitea_mcp_server.py:21805`, reached from `gitea_mcp_server.py:21741`.
 
 **Finding 7 — The highest-value coordination asset has the weakest gate.** A6 is protected
 by filesystem permissions alone (CR14). Corrupting a lease requires no Gitea credential,
@@ -219,8 +219,8 @@ assumes. Every other asset costs an attacker a credential; this one costs nothin
 local access, which is exactly ADV5's position.
 
 **Finding 8 — Provenance authenticates the launch, not the caller.** `server_provenance` is
-reported as exactly `client_managed` or `manual_launch` (`gitea_mcp_server.py:19070`),
+reported as exactly `client_managed` or `manual_launch` (`gitea_mcp_server.py:19217`),
-derived from environment inspection (`gitea_mcp_server.py:15481`) with the recognized-key
+derived from environment inspection (`gitea_mcp_server.py:15630`) with the recognized-key
 allowlist at `gitea_config.py:1172` and the generator that emits the marker at
 `gitea_config.py:1233`. Every one of those facts is fixed at process start. A client that is
 trustworthy at launch and compromised a minute later remains `client_managed` for the life
@@ -259,7 +259,7 @@ holds the token and calls the API instead of the tool.
 
 **D3 — Credential resolution is scoped to the request principal.** A session must resolve its
 own credential and must have no path to any other principal's. The resolve-every-profile
-behavior behind `gitea_mcp_server.py:19378` and `gitea_mcp_server.py:19713` must report
+behavior behind `gitea_mcp_server.py:19538` and `gitea_mcp_server.py:19873` must report
 configured-or-not from configuration alone, without resolving the secret.
 
 *Rationale.* Finding 3. An audit surface that proves a credential exists by fetching it is a
@@ -336,11 +336,11 @@ The client is attached to the local fleet over stdio.
 
 | Boundary | What ADV1 reaches | Stopped by |
 | -------- | ----------------- | ---------- |
-| B1 | Everything the fleet serves. The client *is* the sanctioned launcher: it satisfies the client-managed check (`gitea_mcp_server.py:15481`) by construction, and provenance is never re-verified after launch (Finding 8). | Nothing. The guard authenticates the launch, not the caller. |
+| B1 | Everything the fleet serves. The client *is* the sanctioned launcher: it satisfies the client-managed check (`gitea_mcp_server.py:15630`) by construction, and provenance is never re-verified after launch (Finding 8). | Nothing. The guard authenticates the launch, not the caller. |
 | B2 | All five roles — it is attached to all five namespaces. It can author a PR, approve it from the reviewer namespace, and merge it from the merger namespace. | Only the in-process self-review check, which compares `jcwalker3` (author) against `sysadmin` (reviewer) and **passes**, because Finding 1 made them different accounts while leaving reviewer and merger identical. A9 falls in one sequence of legitimate calls. |
 | B3 | Every credential in CR1–CR10 via CR13, with no additional prompt — the daemon is already sanctioned, so `assert_keychain_access_allowed` (`mcp_daemon_guard.py:583`) returns immediately. | Nothing. |
 | B4 | A1 and A2 in full. | Branch protection at the forge, to the extent configured. |
-| B5 | The operator's checkout and every worktree, through the author tools (`gitea_mcp_server.py:10956`), plus the shared stderr path at `mcp_server.py:13`. | Nothing; the server's disk is the target disk. |
+| B5 | The operator's checkout and every worktree, through the author tools (`gitea_mcp_server.py:10957`), plus the shared stderr path at `mcp_server.py:13`. | Nothing; the server's disk is the target disk. |
 | B6 | All coordination state — no credential required (CR14). It can forge lease ownership and clear decision locks. | Filesystem permissions, which it already satisfies. |
 | B7 | Jenkins (A4) and GlitchTip (A5) secrets via Finding 4, and CR11/CR12 from its own environment. | Nothing. |
 | B8 | Both tenants. | Nothing in-process; only the disabled-context check (`gitea_config.py:704`), which does not apply to enabled contexts. |

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,22 @@ 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",
+    # #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

@@ -199,6 +199,7 @@ RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
 import namespace_workspace_binding as nwb  # noqa: E402
 import canonical_repository_root as crr  # noqa: E402  # #706 cross-repo canonical root
 import mcp_namespace_health  # noqa: E402
+import mcp_worker_identity  # noqa: E402  # #948 single client/session provenance authority
 import stale_binding_recovery  # noqa: E402
 
 # Worktree env bindings inherited from the parent environment at daemon boot
@@ -499,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,
@@ -515,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)
@@ -619,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"
@@ -648,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"] = (
@@ -657,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')}",
@@ -929,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,
@@ -1851,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()
@@ -1865,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,
@@ -1879,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(
@@ -2183,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:
@@ -3413,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()
@@ -13546,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",
@@ -14916,7 +15151,7 @@ def _stale_runtime_reconnect_action() -> str:
     return (
         "blocker_kind=runtime_reconnect_required: call "
         "gitea_request_mcp_reconnect(namespace=<active gitea-* namespace>, "
-        "reason='stale-runtime', client='codex') for a typed operator "
+        "reason='stale-runtime') for a typed operator "
         "reconnect blocker with exact UI steps, then reconnect the IDE/client "
         "MCP session so the server reloads at the current master head. Do not "
         "call gitea_activate_profile, pkill, touch configs, or switch MCP role "
@@ -15120,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:
@@ -15148,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(),
@@ -15478,34 +15709,253 @@ def _session_context_mutation_block(
     return blocked
 
 
-def _is_client_managed_process() -> bool:
+# --- #948 client/session runtime ownership -------------------------------
-    """Check whether the current MCP server process has client-managed launch provenance (#686)."""
+#
-    val = (
+# One daemon launch is one *generation*. The session that owns it registers a
-        os.environ.get("GITEA_CLIENT_MANAGED")
+# unique worker identity against it, so a second healthy client attaching its
-        or os.environ.get("GITEA_MCP_CLIENT_MANAGED")
+# own generation is no longer indistinguishable from a duplicate process.
-        or os.environ.get("GITEA_SERVER_PROVENANCE")
+# Everything here is process-local cache plus a local SQLite registry: no Gitea
-        or os.environ.get("GITEA_FORCE_CLIENT_MANAGED")
+# call, no network, no config write, and every failure degrades to "unproven"
-        or ""
+# rather than raising into a tool call.
-    ).strip().lower()
 
-    if val in ("0", "false", "no", "manual", "manual_launch"):
+_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
+#: produced Codex reconnect steps for a Gemini operator.
+CLIENT_NAME_ENV = "GITEA_MCP_CLIENT"
+CLIENT_INSTANCE_ENV = "GITEA_MCP_CLIENT_INSTANCE"
+CLIENT_SESSION_ENV = "GITEA_MCP_CLIENT_SESSION"
+
+
+def _worker_registry():
+    """Local worker registry, or ``None`` when it cannot be opened.
+
+    Under pytest the registry is only opened when a test has pinned a path, so
+    a test run never writes into the operator's real registry.
+    """
+    global _WORKER_REGISTRY
+    if _WORKER_REGISTRY is not None:
+        return _WORKER_REGISTRY
+    if mcp_daemon_guard.is_pytest_runtime() and not (
+        os.environ.get(mcp_worker_identity.REGISTRY_PATH_ENV) or ""
+    ).strip():
+        return None
+    try:
+        _WORKER_REGISTRY = mcp_worker_identity.WorkerRegistry()
+    except Exception:
+        return None
+    return _WORKER_REGISTRY
+
+
+def _client_identity_hints() -> dict:
+    """What the launcher told us about itself. Unset fields stay unset."""
+    return {
+        "client_name": (os.environ.get(CLIENT_NAME_ENV) or "").strip() or None,
+        "client_instance_id": (os.environ.get(CLIENT_INSTANCE_ENV) or "").strip()
+        or f"pid-{os.getpid()}",
+        "session_id": (os.environ.get(CLIENT_SESSION_ENV) or "").strip()
+        or f"proc-{os.getpid()}-{_process_boot_head_sha or 'nohead'}",
+    }
+
+
+def _active_worker_identity() -> str | None:
+    """This runtime's worker identity, registering it once per process.
+
+    A collision does not adopt the existing registration: a fresh identity is
+    minted and registered instead, which is the #948 AC32 requirement and also
+    the only safe response — adopting would silently transfer another worker's
+    leases and fencing tokens.
+    """
+    global _WORKER_IDENTITY, _WORKER_GENERATION, _WORKER_REGISTRATION_ATTEMPTED
+    if _WORKER_IDENTITY is not None:
+        return _WORKER_IDENTITY
+    if _WORKER_REGISTRATION_ATTEMPTED:
+        return None
+    _WORKER_REGISTRATION_ATTEMPTED = True
+
+    registry = _worker_registry()
+    if registry is None:
+        return None
+
+    hints = _client_identity_hints()
+    generation = _WORKER_GENERATION or mcp_worker_identity.new_generation_id(os.getpid())
+    native = mcp_daemon_guard.native_runtime_status()
+
+    try:
+        for _attempt in range(3):
+            identity = mcp_worker_identity.generate_worker_identity(
+                hints["client_name"], hints["session_id"]
+            )
+            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(),
+                profile=(os.environ.get(gitea_config.ENV_PROFILE) or "").strip() or None,
+                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,
+            )
+            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
+            # Collided: loop mints a different identity rather than reusing this
+            # one. The existing registration is left exactly as it was.
+    except Exception:
+        return 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.
+
+    The role is recorded for diagnostics only. It is deliberately not part of
+    the identity: role is a reusable capability definition that many live
+    workers may share (#948 AC37).
+    """
+    try:
+        profile = get_profile() or {}
+        return _role_kind(
+            profile.get("allowed_operations") or [],
+            profile.get("forbidden_operations") or [],
+        )
+    except Exception:
+        return None
+
+
+def _stdin_is_tty() -> bool:
+    """Terminal evidence for launch provenance, tolerant of a detached stdin."""
+    try:
+        return bool(sys.stdin and sys.stdin.isatty())
+    except Exception:
         return False
 
-    if val in ("1", "true", "yes", "client_managed"):
-        return True
 
-    # A terminal launch has an active TTY on stdin
+def _launch_provenance() -> dict:
-    try:
+    """This process's launch provenance, from the one shared authority (#948).
-        if sys.stdin and sys.stdin.isatty():
-            return False
-    except Exception:
-        pass
 
-    # Standard client launch or test runner with stdio pipe and profile env
+    Previously each surface reimplemented this. ``gitea_get_runtime_context``
-    if "GITEA_MCP_CONFIG" in os.environ or "GITEA_MCP_PROFILE" in os.environ or "GITEA_PROFILE_NAME" in os.environ:
+    read the live environment while ``mcp_namespace_health`` read a filtered
-        return True
+    summary that structurally could not see the provenance keys, so the two
+    reported different provenance for the same process. Both now call
+    ``mcp_worker_identity``.
+    """
+    return mcp_worker_identity.assess_launch_provenance(
+        dict(os.environ), stdin_is_tty=_stdin_is_tty()
+    )
 
-    return False
+
+def _is_client_managed_process() -> bool:
+    """Check whether the current MCP server process has client-managed launch provenance (#686).
+
+    #948: the decision logic moved to
+    ``mcp_worker_identity.assess_launch_provenance`` unchanged — same env keys,
+    same precedence, same TTY fallback — so this keeps returning exactly what it
+    always returned while no longer being a second, divergent implementation.
+    Launch provenance answers "was this hand-launched from a terminal", which is
+    the #686 question. It does *not* answer which live session owns this
+    runtime; that is ``session_ownership`` and needs an attachment record.
+    """
+    return bool(_launch_provenance()["client_managed"])
 
 
 def _provenance_mutation_block(**extra_fields) -> dict | None:
@@ -19048,7 +19498,21 @@ def gitea_get_runtime_context(
         source="gitea_get_runtime_context",
     )
 
-    is_client_managed = _is_client_managed_process()
+    # #948: one assessment, shared with mcp_namespace_health. Reporting both
+    # dimensions from a single call is what makes the two surfaces agree — the
+    # contradiction they used to produce came from two implementations, not from
+    # two genuinely different observations.
+    provenance_assessment = mcp_worker_identity.assess_provenance(
+        registry=_worker_registry(),
+        worker_identity=_active_worker_identity(),
+        env=dict(os.environ),
+        native_transport_bound=mcp_daemon_guard.bound_transport() is not None,
+        profile=profile.get("profile_name"),
+        role=_role_kind(allowed, forbidden),
+        stdin_is_tty=_stdin_is_tty(),
+        pid_alive_probe=issue_lock_store.is_process_alive,
+    )
+    is_client_managed = provenance_assessment["is_client_managed"]
     unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides()
 
     result = {
@@ -19067,8 +19531,26 @@ def gitea_get_runtime_context(
         "review_merge_blocked_reasons": blocked_reasons,
         "suggested_fix": suggested_fix,
         "safe_next_action": safe_next_action,
-        "server_provenance": "client_managed" if is_client_managed else "manual_launch",
+        "server_provenance": provenance_assessment["launch_provenance"],
         "is_client_managed": is_client_managed,
+        # #948: the ownership dimension, which the environment cannot establish.
+        # A surface that needs "may this session mutate on behalf of its client"
+        # reads these, not server_provenance.
+        "session_ownership": provenance_assessment["session_ownership"],
+        "session_owned": provenance_assessment["session_owned"],
+        "worker_identity": provenance_assessment["worker_identity"],
+        "session_id": provenance_assessment["session_id"],
+        "client_instance_id": provenance_assessment["client_instance_id"],
+        "generation_id": provenance_assessment["generation_id"],
+        "client_name": provenance_assessment["client_name"],
+        "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"],
@@ -21577,6 +22059,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).
@@ -21654,11 +22147,23 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
         if match:
             profile = match.group(1)
 
+        # #948: provenance for a scanned peer comes from the shared authority,
+        # fed by that peer's own environment, so the fleet scan cannot disagree
+        # with what that peer reports about itself.
+        peer_env = {
+            m.group(1): m.group(2)
+            for m in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out)
+        }
+        # declared_only: this is a peer process. Its stdin is not ours to
+        # inspect, and it inherits GITEA_MCP_PROFILE from any shell that
+        # exported it, so only an explicit declaration counts as evidence here.
         is_client_managed = bool(
-            re.search(r'\bGITEA_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
+            mcp_worker_identity.assess_launch_provenance(
-            or re.search(r'\bGITEA_MCP_CLIENT_MANAGED=(1|true|yes|client_managed)\b', env_out, re.IGNORECASE)
+                peer_env, declared_only=True
-            or re.search(r'\bGITEA_SERVER_PROVENANCE=client_managed\b', env_out, re.IGNORECASE)
+            )["client_managed"]
         )
+        peer_worker_identity = peer_env.get("GITEA_MCP_WORKER_IDENTITY")
+        peer_generation = peer_env.get("GITEA_MCP_GENERATION_ID")
 
         for env_match in re.finditer(r'\b(GITEA_[A-Z0-9_]+)=([^\s]+)', env_out):
             k, v = env_match.group(1), env_match.group(2)
@@ -21674,6 +22179,8 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
             "start_time": start_time,
             "is_stale": is_stale,
             "is_client_managed": is_client_managed,
+            "worker_identity": peer_worker_identity,
+            "generation_id": peer_generation,
         }
         if profile not in all_profile_procs:
             all_profile_procs[profile] = []
@@ -21681,12 +22188,44 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
 
     running_profiles = {}
     for profile, procs in all_profile_procs.items():
-        if len(procs) > 1:
+        # #948 AC40/AC43: sharing a profile is legitimate — profile is a
+        # reusable capability definition, not a worker identity. What is *not*
+        # legitimate is reusing one worker identity, or two live sessions
+        # claiming one generation. Distinctness has to be proven, though:
+        # processes carrying no identity evidence are indistinguishable, so
+        # they stay classified as duplicates and keep the #686 wall intact.
+        identified = [p for p in procs if p.get("worker_identity")]
+        distinct_identities = {p["worker_identity"] for p in identified}
+        all_identified = len(identified) == len(procs)
+        duplicate_identity = len(identified) != len(distinct_identities)
+        contested_generation = any(
+            len({p["worker_identity"] for p in identified if p.get("generation_id") == gen}) > 1
+            for gen in {p.get("generation_id") for p in identified if p.get("generation_id")}
+        )
+
+        if len(procs) > 1 and (
+            not all_identified or duplicate_identity or contested_generation
+        ):
             pids_str = ", ".join(str(p["pid"]) for p in procs)
+            if duplicate_identity or contested_generation:
+                detail = (
+                    "The same worker identity or generation is claimed more than once, "
+                    "so these are genuine duplicates rather than independent workers."
+                )
+            else:
+                detail = (
+                    "Manual or duplicate launches defeat staleness detection and cannot "
+                    "receive client stdio."
+                )
             reasons.append(
                 f"stale-runtime: Duplicate MCP server process(es) detected for profile '{profile}' (PIDs: {pids_str}). "
-                "Manual or duplicate launches defeat staleness detection and cannot receive client stdio."
+                + detail
             )
+        # Otherwise: several independently identified workers share one profile.
+        # No reason is appended, deliberately. Every reason this function
+        # returns is raised as a hard RuntimeError by its callers, so recording
+        # legitimate concurrency here as "informational" would block exactly the
+        # case #948 exists to permit.
         client_procs = [p for p in procs if p["is_client_managed"]]
         if client_procs:
             client_procs.sort(key=lambda p: p["start_time"], reverse=True)
@@ -22095,7 +22634,7 @@ def gitea_resolve_task_capability(
         next_safe_action = (
             "blocker_kind=runtime_reconnect_required: call "
             "gitea_request_mcp_reconnect(namespace=<active gitea-* namespace>, "
-            "reason='stale-runtime', client='codex') for a typed operator "
+            "reason='stale-runtime') for a typed operator "
             "blocker with exact UI steps, then reconnect/reload the IDE-managed "
             "Gitea MCP server for this profile so it reloads current master. "
             "Do not edit mcp_config.json by hand, pkill, or touch configs; the "
@@ -23663,7 +24202,7 @@ def gitea_workflow_dashboard(
 def gitea_request_mcp_reconnect(
     namespace: str | None = None,
     reason: str | None = None,
-    client: str = "codex",
+    client: str | None = None,
     remote: str = "dadeschools",
     host: str | None = None,
     session_id: str | None = None,
@@ -23693,8 +24232,12 @@ def gitea_request_mcp_reconnect(
             to the active profile's inferred namespace.
         reason: Why reconnect is requested: ``stale-runtime``, ``transport_eof``,
             ``missing_namespace``, ``not_required``, or free-form (normalized).
-        client: Operator UI surface — ``codex`` (default), ``claude_code``, or
+        client: Operator UI surface — ``codex``, ``claude_code``, or
-            ``generic``.
+            ``generic``. #948: omitting it resolves the client from this
+            runtime's live attachment record rather than assuming one vendor,
+            and falls back to ``generic`` when nothing identifies the client.
+            Emitting Codex panel steps to a Gemini/Antigravity operator left
+            them with no reachable recovery path.
         remote: Known instance — ``dadeschools`` or ``prgs`` (parity context).
         host: Optional host override for parity context.
         session_id: Optional session id to echo in the report.
@@ -23754,6 +24297,19 @@ def gitea_request_mcp_reconnect(
         else:
             effective_reason = mcp_client_reconnect.REASON_UNSPECIFIED
 
+    # #948: an omitted client is resolved from the live attachment record, so
+    # the steps describe the UI the operator is actually in front of.
+    if not (client or "").strip():
+        client = mcp_worker_identity.reconnect_client_for(
+            mcp_worker_identity.assess_provenance(
+                registry=_worker_registry(),
+                worker_identity=_active_worker_identity(),
+                env=dict(os.environ),
+                stdin_is_tty=_stdin_is_tty(),
+                pid_alive_probe=issue_lock_store.is_process_alive,
+            )
+        )
+
     payload = mcp_client_reconnect.build_reconnect_request(
         namespace=ns,
         profile=profile_name,

mcp_client_reconnect.py

@@ -92,7 +92,15 @@ OPERATOR_UI_STEPS: dict[str, tuple[str, ...]] = {
     ),
 }
 
-DEFAULT_CLIENT = "codex"
+#: What an *unidentified* client gets. #948: this is deliberately the
+#: host-agnostic step set rather than a specific product. Defaulting to one
+#: vendor emitted Codex UI steps to a Gemini/Antigravity operator, who then had
+#: no reachable recovery path — the guidance named a panel they do not have.
+DEFAULT_CLIENT = "generic"
+
+#: The historical default, kept addressable by name so Codex callers still get
+#: Codex steps, without it silently becoming the fallback for unknown clients.
+LEGACY_DEFAULT_CLIENT = "codex"
 
 
 def normalize_reason(reason: str | None) -> str:
@@ -128,14 +136,18 @@ def normalize_reason(reason: str | None) -> str:
 
 
 def normalize_client(client: str | None) -> str:
-    """Return a known client key for operator UI steps."""
+    """Return the UI-step key for a client.
-    text = (client or "").strip().lower().replace(" ", "_").replace("-", "_")
+
-    if text in ("codex", "openai_codex", "openai"):
+    #948: alias resolution is shared with ``mcp_worker_identity`` so a client
-        return "codex"
+    name means the same thing wherever it is read. A name we recognise but have
-    if text in ("claude", "claude_code", "claude_desktop", "anthropic"):
+    no bespoke steps for — Gemini, Antigravity, Grok — resolves to the generic
-        return "claude_code"
+    host-agnostic steps rather than to another vendor's panel.
-    if text in OPERATOR_UI_STEPS:
+    """
-        return text
+    import mcp_worker_identity
+
+    canonical = mcp_worker_identity.normalize_client_name(client)
+    if canonical in OPERATOR_UI_STEPS:
+        return canonical
     return DEFAULT_CLIENT
 
 

mcp_namespace_health.py

@@ -95,6 +95,15 @@ SAFE_ENV_KEYS = (
     "GITEA_MCP_CONFIG",
 )
 
+# #948: provenance used to be derived from the summary this allowlist produces.
+# The allowlist never carried a provenance key, so that derivation could only
+# ever evaluate to ``manual_launch`` — whatever the process actually was — while
+# ``gitea_get_runtime_context`` read the live environment and reported
+# ``client_managed`` for the same process. Provenance is no longer derived here.
+# It comes from ``mcp_worker_identity.assess_provenance``, the single authority
+# every surface shares. This allowlist keeps its original and only job: deciding
+# which env values are safe to echo back in diagnostics.
+
 
 def assess_connected_namespace_attachment(
     *,
@@ -443,12 +452,23 @@ def classify_namespace_probe(
     profile: str | None = None,
     configured: bool = True,
     probe_source: str | None = None,
+    worker_identity: str | None = None,
+    generation_id: str | None = None,
+    registry: Any | None = None,
+    pid_alive_probe: Any | None = None,
 ) -> dict[str, Any]:
     """Classify whether a required tool is callable through a live namespace.
 
     ``registered_tools`` is static/server-side evidence. ``probe_result`` is
     live invocation evidence. Only ``probe_source=client_namespace`` proves the
     IDE-managed path; ``offline_spawn`` is an offline subprocess check only.
+
+    #948: ``worker_identity``/``generation_id``/``registry`` carry the
+    client/session ownership evidence. Provenance is resolved by
+    ``mcp_worker_identity.assess_provenance`` — the same call
+    ``gitea_get_runtime_context`` makes — so the two surfaces cannot report
+    different provenance for one process. Omitting them yields the fail-closed
+    ``unproven`` verdict, never a fabricated ``client_managed``.
     """
     ns = (namespace or "").strip()
     tool = required_tool or REQUIRED_NAMESPACE_TOOLS.get(ns) or "gitea_whoami"
@@ -565,14 +585,31 @@ def classify_namespace_probe(
         blocks = namespace_health_blocks_task("merge_pr", healthy)
 
     import gitea_config
+    import mcp_worker_identity
+
     raw_env = process.get("env") if isinstance(process, dict) else None
     unconsumed_env = gitea_config.get_unconsumed_gitea_env_overrides(raw_env)
-    is_client_managed = bool(
+
-        env_summary.get("GITEA_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
+    # #948: one authority, shared with gitea_get_runtime_context. The env is
-        or env_summary.get("GITEA_MCP_CLIENT_MANAGED") in ("1", "true", "yes", "client_managed")
+    # passed whole rather than through SAFE_ENV_KEYS — the allowlist exists to
-        or env_summary.get("GITEA_SERVER_PROVENANCE") == "client_managed"
+    # decide what may be *echoed*, and using it to decide what may be *believed*
+    # is what made this surface structurally unable to report client_managed.
+    # ``declared_only``: ``process`` describes an observed peer, not this
+    # interpreter. Its stdin is unavailable and its launcher-config env is
+    # inherited from whatever shell started it, so only an explicit declaration
+    # is evidence. Absence of one is ``unproven``, not an asserted manual launch.
+    provenance_verdict = mcp_worker_identity.assess_provenance(
+        registry=registry,
+        worker_identity=worker_identity,
+        generation_id=generation_id,
+        env=raw_env if isinstance(raw_env, dict) else {},
+        namespace=ns,
+        profile=profile_name,
+        pid_alive_probe=pid_alive_probe,
+        declared_only=True,
     )
-    provenance = "client_managed" if is_client_managed else "manual_launch"
+    provenance = provenance_verdict["provenance"]
+    is_client_managed = provenance_verdict["is_client_managed"]
 
     return {
         "success": healthy,
@@ -591,6 +628,15 @@ def classify_namespace_probe(
         "remediation": remediation,
         "provenance": provenance,
         "is_client_managed": is_client_managed,
+        # Every non-client-session verdict fails closed. Consumers that only
+        # need "may this mutate?" read this and stay correct across the #948
+        # vocabulary split between ``manual_launch`` and ``unproven``.
+        "provenance_fail_closed": provenance_verdict["fail_closed"],
+        "provenance_assessment": provenance_verdict,
+        "worker_identity": provenance_verdict["worker_identity"],
+        "session_id": provenance_verdict["session_id"],
+        "generation_id": provenance_verdict["generation_id"],
+        "client_name": provenance_verdict["client_name"],
         "unconsumed_gitea_env": unconsumed_env,
         "diagnostics": {
             "namespace": ns,
@@ -602,6 +648,13 @@ def classify_namespace_probe(
             "probe_source": source,
             "provenance": provenance,
             "is_client_managed": is_client_managed,
+            "provenance_fail_closed": provenance_verdict["fail_closed"],
+            "provenance_blocker_kind": provenance_verdict["blocker_kind"],
+            "provenance_scope": provenance_verdict["scope"],
+            "worker_identity": provenance_verdict["worker_identity"],
+            "session_id": provenance_verdict["session_id"],
+            "generation_id": provenance_verdict["generation_id"],
+            "client_name": provenance_verdict["client_name"],
             "unconsumed_gitea_env": unconsumed_env,
         },
         "blocks_merge_workflow": blocks,

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,26 +563,27 @@ 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
+            reasons.append(
-        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
+                f"{role} mutation blocked: workspace is the stable control checkout; "
-    ):
+                f"create or reconnect to a session-owned worktree under branches/ "
-        reasons.append(
+                f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / "
-            f"{role} mutation blocked: workspace is the stable control checkout; "
+                f"{ACTIVE_WORKTREE_ENV}"
-            f"create or reconnect to a session-owned worktree under branches/ "
+            )
-            f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / "
+        elif mutation_workspace != process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
-            f"{ACTIVE_WORKTREE_ENV}"
+            reasons.append(
-        )
+                f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
-    elif (
+                f"'{ctx['canonical_repo_root']}/branches/'"
-        role in {"reviewer", "merger"}
+            )
-        and mutation_workspace != process_root
+
-        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
+        if path_exists and amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
-    ):
+            in_list = ctx.get("in_git_worktree_list")
-        reasons.append(
+            if in_list is False:
-            f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
+                reasons.append(
-            f"'{ctx['canonical_repo_root']}/branches/'"
+                    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 {

task_capability_map.py

@@ -386,6 +386,25 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.branch.delete",
         "role": "reconciler",
     },
+    # #970: auditing missing worktree bindings is read-only; retiring one is a
+    # control-plane cleanup mutation and carries the same reconciler-only
+    # authority as any other reconciliation cleanup (review 644 B2).
+    "audit_missing_worktree_bindings": {
+        "permission": "gitea.read",
+        "role": "reconciler",
+    },
+    "gitea_audit_missing_worktree_bindings": {
+        "permission": "gitea.read",
+        "role": "reconciler",
+    },
+    "reconcile_missing_worktree_bindings": {
+        "permission": "gitea.branch.delete",
+        "role": "reconciler",
+    },
+    "gitea_reconcile_missing_worktree_bindings": {
+        "permission": "gitea.branch.delete",
+        "role": "reconciler",
+    },
     "work_issue": {
         "permission": "gitea.pr.create",
         "role": "author",
@@ -653,6 +672,8 @@ ROLE_EXCLUSIVE_TASKS: frozenset[str] = frozenset(
         "delete_branch",
         "cleanup_merged_pr_branch",
         "reconciliation_cleanup",
+        "reconcile_missing_worktree_bindings",
+        "gitea_reconcile_missing_worktree_bindings",
         "work_issue",
         "work-issue",
     }

tests/test_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_686_manual_mcp_provenance.py

@@ -112,7 +112,19 @@ class TestIssue686ManualMcpProvenance(unittest.TestCase):
         self.assertTrue(any("All matching profiles for task 'create_issue' (['prgs-author']) are running but stale" in r for r in reasons))
 
     def test_namespace_health_classification_includes_provenance(self):
-        """AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env."""
+        """AC 1 & 4: mcp_namespace_health diagnostics include provenance and unconsumed_gitea_env.
+
+        #948 narrowed the vocabulary here. This process carries no client-managed
+        declaration, so the old code labelled it ``manual_launch`` — asserting a
+        hand-launched terminal process it had no evidence for, and contradicting
+        ``gitea_get_runtime_context``, which read the same process and reported
+        ``client_managed``. Absence of proof is now reported as ``unproven``.
+
+        The #686 wall itself is unchanged and still asserted below:
+        ``is_client_managed`` stays False, so nothing previously refused is now
+        permitted. Only the label on the *reason* changed, so remediation names
+        the proof that is actually missing.
+        """
         process = {
             "pid": 5555,
             "profile": "prgs-author",
@@ -129,10 +141,12 @@ class TestIssue686ManualMcpProvenance(unittest.TestCase):
             process=process,
             probe_source="client_namespace",
         )
-        self.assertEqual(res["provenance"], "manual_launch")
+        self.assertEqual(res["provenance"], "unproven")
         self.assertFalse(res["is_client_managed"])
+        # The wall is intact: no client-managed proof still fails closed.
+        self.assertTrue(res["provenance_fail_closed"])
         self.assertEqual(res["unconsumed_gitea_env"], {"GITEA_DUMMY": "99"})
-        self.assertEqual(res["diagnostics"]["provenance"], "manual_launch")
+        self.assertEqual(res["diagnostics"]["provenance"], "unproven")
 
 
 if __name__ == "__main__":

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

@@ -0,0 +1,662 @@
+"""Client/session-aware runtime ownership and provenance (#948).
+
+Covers the reproduced contradiction that motivated the issue: one surface
+reporting ``client_managed`` while another reported ``manual_launch`` for the
+same process, remediation hardcoded to one vendor, and a profile-wide duplicate
+wall that could not tell two healthy clients apart.
+
+All client and session identifiers here are synthetic.
+"""
+
+from __future__ import annotations
+
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+
+import mcp_client_reconnect
+import mcp_namespace_health
+import mcp_worker_identity as mwi
+
+
+NOW = datetime(2026, 7, 29, 6, 0, 0, tzinfo=timezone.utc)
+
+
+def _registry() -> mwi.WorkerRegistry:
+    """A registry on a throwaway path; never the operator's real one."""
+    handle, path = tempfile.mkstemp(suffix=".sqlite3")
+    os.close(handle)
+    os.unlink(path)
+    return mwi.WorkerRegistry(path)
+
+
+def _attach(
+    registry: mwi.WorkerRegistry,
+    *,
+    client: str,
+    session: str,
+    generation: str,
+    profile: str = "prgs-reviewer",
+    role: str = "reviewer",
+    pid: int = 4242,
+    now: datetime = NOW,
+    ttl: float = 900.0,
+) -> dict:
+    """Register one synthetic worker and return the outcome."""
+    identity = mwi.generate_worker_identity(client, session, now=now)
+    outcome = registry.register(
+        worker_identity=identity,
+        client_name=client,
+        client_instance_id=f"inst-{session}",
+        session_id=session,
+        generation_id=generation,
+        role=role,
+        profile=profile,
+        pid=pid,
+        heartbeat_ttl_seconds=ttl,
+        now=now,
+    )
+    outcome["identity"] = identity
+    return outcome
+
+
+class IdentityFormatTests(unittest.TestCase):
+    """AC27-29: collision-resistant `<llm-name>-<UTC-timestamp>-<short-sha>`."""
+
+    def test_identity_matches_required_format(self):
+        identity = mwi.generate_worker_identity("Gemini", "sess-0001", now=NOW)
+        parsed = mwi.parse_worker_identity(identity)
+        self.assertTrue(parsed["valid"], parsed["reasons"])
+        self.assertEqual(parsed["client_name"], "gemini")
+        self.assertEqual(parsed["minted_at"], "20260729T060000Z")
+        self.assertEqual(len(parsed["digest"]), 12)
+
+    def test_digest_varies_with_session_and_nonce(self):
+        base = dict(timestamp_ns=1, now=NOW)
+        a = mwi.generate_worker_identity("codex", "sess-A", nonce="n", **base)
+        b = mwi.generate_worker_identity("codex", "sess-B", nonce="n", **base)
+        c = mwi.generate_worker_identity("codex", "sess-A", nonce="m", **base)
+        self.assertNotEqual(a, b, "session must feed the digest")
+        self.assertNotEqual(a, c, "nonce must feed the digest")
+
+    def test_identity_is_not_role_or_profile(self):
+        """AC26: identity is independent of role and profile."""
+        args = dict(timestamp_ns=7, nonce="fixed", now=NOW)
+        same = mwi.generate_worker_identity("claude", "sess-1", **args)
+        self.assertEqual(same, mwi.generate_worker_identity("claude", "sess-1", **args))
+        # Nothing role- or profile-derived appears in the identity.
+        self.assertNotIn("reviewer", same)
+        self.assertNotIn("prgs", same)
+
+    def test_malformed_identity_rejected(self):
+        self.assertFalse(mwi.parse_worker_identity("prgs-reviewer")["valid"])
+        self.assertFalse(mwi.parse_worker_identity("")["valid"])
+        self.assertFalse(mwi.parse_worker_identity(None)["valid"])
+
+
+class PerClientAttachmentTests(unittest.TestCase):
+    """Every supported client attaches and is reported as itself."""
+
+    def _assert_attached_as(self, client: str, expected_name: str):
+        registry = _registry()
+        outcome = _attach(
+            registry, client=client, session=f"sess-{client}", generation="gen-1"
+        )
+        self.assertTrue(outcome["registered"], outcome["reasons"])
+
+        verdict = mwi.assess_provenance(
+            registry=registry, worker_identity=outcome["identity"], env={}, now=NOW
+        )
+        self.assertEqual(verdict["session_ownership"], mwi.OWNERSHIP_OWNED)
+        self.assertEqual(verdict["provenance"], mwi.PROVENANCE_CLIENT_SESSION)
+        self.assertEqual(verdict["client_name"], expected_name)
+        self.assertTrue(verdict["session_owned"])
+        self.assertFalse(verdict["fail_closed"])
+        return verdict
+
+    def test_codex_attachment(self):
+        self._assert_attached_as("codex", "codex")
+
+    def test_gemini_attachment(self):
+        self._assert_attached_as("gemini", "gemini")
+
+    def test_antigravity_attachment(self):
+        self._assert_attached_as("antigravity", "antigravity")
+
+    def test_claude_attachment(self):
+        self._assert_attached_as("claude", "claude_code")
+
+    def test_unknown_client_is_named_not_guessed(self):
+        verdict = self._assert_attached_as("some_new_llm", "some_new_llm")
+        self.assertNotEqual(verdict["client_name"], "codex")
+
+
+class SessionLifecycleTests(unittest.TestCase):
+    def test_same_client_new_session_gets_distinct_identity(self):
+        registry = _registry()
+        first = _attach(registry, client="codex", session="sess-1", generation="gen-1")
+        second = _attach(registry, client="codex", session="sess-2", generation="gen-2")
+        self.assertTrue(first["registered"])
+        self.assertTrue(second["registered"])
+        self.assertNotEqual(first["identity"], second["identity"])
+
+        # Both are live and neither blocks the other.
+        cohort = mwi.classify_cohort(registry.list_workers(), now=NOW)
+        self.assertEqual(cohort["live_worker_count"], 2)
+        self.assertFalse(cohort["blocked"], cohort["reasons"])
+
+    def test_different_client_attaches_after_previous_session_ends(self):
+        """AC14: expiry then takeover with a higher fencing epoch."""
+        registry = _registry()
+        gone = _attach(
+            registry, client="codex", session="sess-old", generation="gen-shared", ttl=60
+        )
+        later = NOW + timedelta(hours=1)
+        self.assertFalse(
+            registry.is_live(registry.get(gone["identity"]), now=later)["live"]
+        )
+
+        arriving = _attach(
+            registry,
+            client="gemini",
+            session="sess-new",
+            generation="gen-other",
+            now=later,
+        )
+        claim = registry.claim_generation(
+            worker_identity=arriving["identity"],
+            generation_id="gen-shared",
+            now=later,
+        )
+        self.assertTrue(claim["claimed"], claim["reasons"])
+        self.assertIn(gone["identity"], claim["superseded_workers"])
+        self.assertGreater(claim["fencing_epoch"], gone["fencing_epoch"])
+
+    def test_superseded_session_is_fenced_on_resume(self):
+        """AC15/AC16: the prior session cannot heartbeat its way back."""
+        registry = _registry()
+        old = _attach(
+            registry, client="codex", session="sess-old", generation="gen-shared", ttl=60
+        )
+        later = NOW + timedelta(hours=1)
+        new = _attach(
+            registry, client="gemini", session="sess-new", generation="gen-x", now=later
+        )
+        registry.claim_generation(
+            worker_identity=new["identity"], generation_id="gen-shared", now=later
+        )
+
+        resumed = registry.heartbeat(
+            worker_identity=old["identity"],
+            fencing_epoch=old["fencing_epoch"],
+            now=later,
+        )
+        self.assertFalse(resumed["renewed"])
+        self.assertFalse(resumed["mutation_performed"])
+        self.assertEqual(resumed["blocker_kind"], mwi.BLOCKER_FENCED)
+
+    def test_heartbeat_renews_only_the_owning_lease(self):
+        """AC11: a wrong epoch never renews, and never mutates."""
+        registry = _registry()
+        worker = _attach(registry, client="codex", session="s", generation="g")
+        good = registry.heartbeat(
+            worker_identity=worker["identity"],
+            fencing_epoch=worker["fencing_epoch"],
+            now=NOW + timedelta(minutes=5),
+        )
+        self.assertTrue(good["renewed"])
+
+        bad = registry.heartbeat(
+            worker_identity=worker["identity"],
+            fencing_epoch=worker["fencing_epoch"] + 99,
+            now=NOW + timedelta(minutes=6),
+        )
+        self.assertFalse(bad["renewed"])
+        self.assertFalse(bad["mutation_performed"])
+        self.assertEqual(
+            registry.get(worker["identity"])["last_heartbeat_at"],
+            good["last_heartbeat_at"],
+            "a refused heartbeat must not advance the record",
+        )
+
+
+class ConflictAndCollisionTests(unittest.TestCase):
+    def test_two_live_sessions_cannot_claim_one_generation(self):
+        registry = _registry()
+        first = _attach(registry, client="codex", session="s1", generation="gen-shared")
+        second = _attach(registry, client="gemini", session="s2", generation="gen-other")
+
+        claim = registry.claim_generation(
+            worker_identity=second["identity"],
+            generation_id="gen-shared",
+            now=NOW,
+        )
+        self.assertFalse(claim["claimed"])
+        self.assertFalse(claim["mutation_performed"])
+        self.assertEqual(claim["blocker_kind"], mwi.BLOCKER_CONFLICTING_SESSIONS)
+        self.assertEqual(
+            claim["conflicting_owners"][0]["worker_identity"], first["identity"]
+        )
+        # The sanctioned recovery must never be "kill the other process".
+        self.assertIn("Do not kill", claim["exact_next_action"])
+
+    def test_contested_generation_fails_closed_in_assessment(self):
+        registry = _registry()
+        first = _attach(registry, client="codex", session="s1", generation="gen-shared")
+        _attach(registry, client="gemini", session="s2", generation="gen-shared")
+
+        verdict = mwi.assess_provenance(
+            registry=registry, worker_identity=first["identity"], env={}, now=NOW
+        )
+        self.assertEqual(verdict["session_ownership"], mwi.OWNERSHIP_CONTESTED)
+        self.assertTrue(verdict["fail_closed"])
+        self.assertEqual(verdict["blocker_kind"], mwi.BLOCKER_CONTRADICTORY)
+        self.assertTrue(verdict["conflicting_live_sessions"])
+
+    def test_identity_collision_is_refused_without_corrupting_existing(self):
+        """AC31: never replace, adopt, merge with, or corrupt the incumbent."""
+        registry = _registry()
+        incumbent = _attach(registry, client="codex", session="s1", generation="gen-1")
+        before = registry.get(incumbent["identity"])
+
+        collided = registry.register(
+            worker_identity=incumbent["identity"],
+            client_name="gemini",
+            client_instance_id="inst-other",
+            session_id="s2",
+            generation_id="gen-2",
+            pid=9999,
+            now=NOW,
+        )
+        self.assertFalse(collided["registered"])
+        self.assertTrue(collided["collision"])
+        self.assertFalse(collided["mutation_performed"])
+        self.assertEqual(collided["blocker_kind"], mwi.BLOCKER_IDENTITY_COLLISION)
+        self.assertEqual(collided["collision_kind"], "active_worker")
+        self.assertEqual(
+            registry.get(incumbent["identity"]), before, "incumbent must be untouched"
+        )
+
+    def test_after_collision_a_regenerated_identity_registers(self):
+        """AC32/AC35: forced collision, safe regeneration, successful replacement."""
+        registry = _registry()
+        fixed = dict(timestamp_ns=99, nonce="deterministic", now=NOW)
+        forced = mwi.generate_worker_identity("codex", "sess-collide", **fixed)
+        first = registry.register(
+            worker_identity=forced,
+            client_name="codex",
+            client_instance_id="inst-1",
+            session_id="sess-collide",
+            generation_id="gen-1",
+            now=NOW,
+        )
+        self.assertTrue(first["registered"])
+
+        # A second worker deriving the same inputs collides deterministically.
+        again = mwi.generate_worker_identity("codex", "sess-collide", **fixed)
+        self.assertEqual(again, forced)
+        self.assertTrue(
+            registry.register(
+                worker_identity=again,
+                client_name="codex",
+                client_instance_id="inst-2",
+                session_id="sess-collide",
+                generation_id="gen-2",
+                now=NOW,
+            )["collision"]
+        )
+
+        replacement = mwi.generate_worker_identity(
+            "codex", "sess-collide", timestamp_ns=100, nonce="different", now=NOW
+        )
+        self.assertNotEqual(replacement, forced)
+        self.assertTrue(
+            registry.register(
+                worker_identity=replacement,
+                client_name="codex",
+                client_instance_id="inst-2",
+                session_id="sess-collide",
+                generation_id="gen-2",
+                now=NOW,
+            )["registered"]
+        )
+
+    def test_restarted_worker_inherits_nothing(self):
+        """AC33/AC34: a restart mints a new identity and no prior epoch."""
+        registry = _registry()
+        before = _attach(
+            registry, client="codex", session="sess-before", generation="gen-1", ttl=60
+        )
+        later = NOW + timedelta(hours=2)
+        after = _attach(
+            registry, client="codex", session="sess-after", generation="gen-2", now=later
+        )
+        self.assertNotEqual(before["identity"], after["identity"])
+        self.assertNotEqual(
+            registry.get(after["identity"])["generation_id"],
+            registry.get(before["identity"])["generation_id"],
+        )
+
+
+class LivenessTests(unittest.TestCase):
+    def test_stale_session_record_is_not_live(self):
+        registry = _registry()
+        worker = _attach(registry, client="codex", session="s", generation="g", ttl=300)
+        stale = registry.is_live(
+            registry.get(worker["identity"]), now=NOW + timedelta(hours=1)
+        )
+        self.assertFalse(stale["live"])
+        self.assertFalse(stale["heartbeat_fresh"])
+
+    def test_liveness_is_not_pid_comparison_alone(self):
+        """AC7: a live PID does not resurrect an expired registration."""
+        registry = _registry()
+        worker = _attach(registry, client="codex", session="s", generation="g", ttl=60)
+        verdict = registry.is_live(
+            registry.get(worker["identity"]),
+            now=NOW + timedelta(hours=1),
+            pid_alive=True,
+        )
+        self.assertFalse(
+            verdict["live"], "a live PID must not override a dead heartbeat"
+        )
+
+    def test_dead_pid_withdraws_liveness_from_a_fresh_heartbeat(self):
+        registry = _registry()
+        worker = _attach(registry, client="codex", session="s", generation="g")
+        verdict = registry.is_live(
+            registry.get(worker["identity"]), now=NOW, pid_alive=False
+        )
+        self.assertFalse(verdict["live"])
+
+    def test_stale_ownership_does_not_permanently_strand_a_daemon(self):
+        registry = _registry()
+        stranded = _attach(
+            registry, client="codex", session="s-old", generation="gen-daemon", ttl=60
+        )
+        later = NOW + timedelta(hours=3)
+        rescuer = _attach(
+            registry, client="claude", session="s-new", generation="gen-tmp", now=later
+        )
+        claim = registry.claim_generation(
+            worker_identity=rescuer["identity"],
+            generation_id="gen-daemon",
+            now=later,
+        )
+        self.assertTrue(claim["claimed"], claim["reasons"])
+        self.assertIn(stranded["identity"], claim["superseded_workers"])
+
+
+class EvidenceTests(unittest.TestCase):
+    def test_env_flag_alone_does_not_prove_session_ownership(self):
+        verdict = mwi.assess_provenance(
+            registry=None,
+            worker_identity=None,
+            env={"GITEA_CLIENT_MANAGED": "1", "GITEA_MCP_SANCTIONED_DAEMON": "1"},
+            now=NOW,
+        )
+        self.assertFalse(verdict["session_owned"])
+        self.assertEqual(verdict["session_ownership"], mwi.OWNERSHIP_UNOWNED)
+        self.assertTrue(verdict["env_flag_only"])
+        self.assertTrue(verdict["fail_closed"])
+        self.assertNotIn(mwi.EVIDENCE_ATTACHMENT_RECORD, verdict["evidence"])
+        self.assertFalse(verdict["env_signal"]["proves_session_ownership"])
+
+    def test_env_flag_still_answers_the_launch_question(self):
+        """The #686 wall is preserved: env decides launch, not ownership."""
+        self.assertTrue(
+            mwi.assess_launch_provenance({"GITEA_CLIENT_MANAGED": "1"})["client_managed"]
+        )
+        self.assertFalse(
+            mwi.assess_launch_provenance({"GITEA_CLIENT_MANAGED": "0"})["client_managed"]
+        )
+        self.assertFalse(
+            mwi.assess_launch_provenance({}, stdin_is_tty=True)["client_managed"]
+        )
+        self.assertTrue(
+            mwi.assess_launch_provenance({"GITEA_MCP_PROFILE": "prgs-author"})[
+                "client_managed"
+            ]
+        )
+
+    def test_missing_evidence_is_unproven_not_manual(self):
+        """A missing proof must not be reported as a hand-launched process."""
+        verdict = mwi.assess_provenance(
+            registry=None, worker_identity=None, env={}, now=NOW
+        )
+        self.assertEqual(verdict["provenance"], mwi.PROVENANCE_UNPROVEN)
+        self.assertNotEqual(verdict["provenance"], mwi.PROVENANCE_MANUAL)
+        self.assertTrue(verdict["fail_closed"])
+
+    def test_declared_manual_launch_is_reported_as_manual(self):
+        verdict = mwi.assess_provenance(
+            registry=None,
+            worker_identity=None,
+            env={"GITEA_CLIENT_MANAGED": "0"},
+            now=NOW,
+        )
+        self.assertEqual(verdict["provenance"], mwi.PROVENANCE_MANUAL)
+
+    def test_fail_closed_refusal_names_its_scope_not_the_profile(self):
+        """AC17/AC41: no refusal is profile-wide."""
+        verdict = mwi.assess_provenance(
+            registry=None,
+            worker_identity=None,
+            env={},
+            profile="prgs-reviewer",
+            role="reviewer",
+            now=NOW,
+        )
+        self.assertFalse(verdict["scope"]["profile_wide"])
+        self.assertEqual(verdict["blocker_kind"], mwi.BLOCKER_NO_ATTACHMENT)
+
+
+class CohortScopingTests(unittest.TestCase):
+    def test_shared_profile_with_distinct_identities_does_not_block(self):
+        """AC40: profile is not a singleton identity."""
+        registry = _registry()
+        _attach(
+            registry,
+            client="codex",
+            session="s1",
+            generation="g1",
+            profile="prgs-reviewer",
+        )
+        _attach(
+            registry,
+            client="gemini",
+            session="s2",
+            generation="g2",
+            profile="prgs-reviewer",
+        )
+
+        cohort = mwi.classify_cohort(registry.list_workers(), now=NOW)
+        self.assertFalse(cohort["blocked"], cohort["reasons"])
+        self.assertEqual(cohort["blocker_kind"], mwi.BLOCKER_NONE)
+        self.assertIn("prgs-reviewer", cohort["shared_profiles"])
+        self.assertTrue(cohort["profile_sharing_permitted"])
+        self.assertEqual(cohort["blocked_worker_identities"], [])
+
+    def test_duplicate_cohort_records_block_only_the_offenders(self):
+        registry = _registry()
+        _attach(registry, client="codex", session="s1", generation="gen-contested")
+        _attach(registry, client="gemini", session="s2", generation="gen-contested")
+        _attach(registry, client="claude", session="s3", generation="gen-fine")
+
+        cohort = mwi.classify_cohort(registry.list_workers(), now=NOW)
+        self.assertTrue(cohort["blocked"])
+        self.assertEqual(cohort["contested_generations"], ["gen-contested"])
+        self.assertEqual(len(cohort["blocked_worker_identities"]), 2)
+
+    def test_mixed_runtime_generations_are_scoped_independently(self):
+        """AC17: one stale generation does not wall unrelated healthy ones."""
+        registry = _registry()
+        stale = _attach(
+            registry, client="codex", session="s1", generation="gen-stale", ttl=60
+        )
+        healthy_a = _attach(registry, client="gemini", session="s2", generation="gen-a")
+        healthy_b = _attach(registry, client="claude", session="s3", generation="gen-b")
+
+        scoped = mwi.scope_runtime_failure(
+            failure_kind="stale-runtime",
+            worker_identity=stale["identity"],
+            profile="prgs-reviewer",
+            all_live_workers=registry.list_workers(),
+        )
+        self.assertFalse(scoped["profile_wide"])
+        self.assertFalse(scoped["fleet_wide"])
+        self.assertEqual(len(scoped["affected_workers"]), 1)
+        self.assertEqual(scoped["unaffected_worker_count"], 2)
+        unaffected = {w["worker_identity"] for w in scoped["unaffected_workers"]}
+        self.assertEqual(unaffected, {healthy_a["identity"], healthy_b["identity"]})
+
+
+class HardcodedClientRegressionTests(unittest.TestCase):
+    def test_unknown_client_does_not_resolve_to_codex(self):
+        for name in ("gemini", "antigravity", "grok", "some_new_llm", "", None):
+            with self.subTest(client=name):
+                self.assertNotEqual(
+                    mcp_client_reconnect.normalize_client(name),
+                    "codex",
+                    "an unidentified client must never be handed Codex UI steps",
+                )
+
+    def test_known_clients_still_get_their_own_steps(self):
+        self.assertEqual(mcp_client_reconnect.normalize_client("codex"), "codex")
+        self.assertEqual(
+            mcp_client_reconnect.normalize_client("claude_code"), "claude_code"
+        )
+
+    def test_generic_steps_do_not_name_a_specific_vendor(self):
+        steps = " ".join(mcp_client_reconnect.operator_ui_steps("gemini"))
+        self.assertNotIn("Codex", steps)
+
+    def test_reconnect_client_is_derived_from_the_attachment_record(self):
+        registry = _registry()
+        worker = _attach(registry, client="antigravity", session="s", generation="g")
+        verdict = mwi.assess_provenance(
+            registry=registry, worker_identity=worker["identity"], env={}, now=NOW
+        )
+        self.assertEqual(mwi.reconnect_client_for(verdict), "antigravity")
+
+    def test_reconnect_client_is_unknown_rather_than_guessed(self):
+        self.assertEqual(mwi.reconnect_client_for({}), mwi.UNKNOWN_CLIENT)
+
+
+class RemoteBindingTests(unittest.TestCase):
+    def test_explicit_prgs_selection_is_honoured(self):
+        resolved = mwi.resolve_bound_remote(
+            requested_remote="prgs", bound_remote="prgs", default_remote="dadeschools"
+        )
+        self.assertEqual(resolved["remote"], "prgs")
+        self.assertFalse(resolved["drifted"])
+
+    def test_omitted_remote_uses_the_binding_not_the_library_default(self):
+        """The reported dadeschools host drift."""
+        resolved = mwi.resolve_bound_remote(
+            requested_remote=None, bound_remote="prgs", default_remote="dadeschools"
+        )
+        self.assertEqual(resolved["remote"], "prgs")
+        self.assertNotEqual(resolved["remote"], "dadeschools")
+        self.assertEqual(resolved["resolved_from"], "session_binding")
+
+    def test_contradicting_the_binding_is_refused(self):
+        resolved = mwi.resolve_bound_remote(
+            requested_remote="dadeschools",
+            bound_remote="prgs",
+            default_remote="dadeschools",
+        )
+        self.assertEqual(resolved["remote"], "prgs")
+        self.assertTrue(resolved["drifted"])
+        self.assertFalse(resolved["honoured_request"])
+
+    def test_unbound_session_falls_back_and_says_so(self):
+        resolved = mwi.resolve_bound_remote(
+            requested_remote=None, bound_remote=None, default_remote="dadeschools"
+        )
+        self.assertEqual(resolved["remote"], "dadeschools")
+        self.assertEqual(resolved["resolved_from"], "library_default")
+        self.assertTrue(resolved["reasons"])
+
+
+class SurfaceAgreementTests(unittest.TestCase):
+    """The reproduced contradiction: two surfaces, one process, two answers."""
+
+    def test_namespace_health_and_direct_assessment_agree(self):
+        registry = _registry()
+        worker = _attach(
+            registry,
+            client="gemini",
+            session="sess-agree",
+            generation="gen-agree",
+            profile="prgs-reviewer",
+        )
+        env = {"GITEA_MCP_PROFILE": "prgs-reviewer", "GITEA_CLIENT_MANAGED": "1"}
+
+        direct = mwi.assess_provenance(
+            registry=registry,
+            worker_identity=worker["identity"],
+            env=env,
+            profile="prgs-reviewer",
+        )
+        health = mcp_namespace_health.classify_namespace_probe(
+            "gitea-reviewer",
+            configured=True,
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True},
+            probe_source="client_namespace",
+            process={"pid": 4242, "profile": "prgs-reviewer", "env": env},
+            registry=registry,
+            worker_identity=worker["identity"],
+        )
+
+        self.assertEqual(health["provenance"], direct["provenance"])
+        self.assertEqual(health["is_client_managed"], direct["is_client_managed"])
+        self.assertEqual(health["worker_identity"], direct["worker_identity"])
+        self.assertEqual(health["session_id"], "sess-agree")
+        self.assertEqual(health["client_name"], "gemini")
+
+    def test_namespace_health_can_report_client_managed_at_all(self):
+        """The old derivation was structurally incapable of this."""
+        env = {"GITEA_CLIENT_MANAGED": "1", "GITEA_MCP_PROFILE": "prgs-author"}
+        health = mcp_namespace_health.classify_namespace_probe(
+            "gitea-author",
+            configured=True,
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True},
+            probe_source="client_namespace",
+            process={"pid": 1234, "profile": "prgs-author", "env": env},
+        )
+        self.assertTrue(
+            health["is_client_managed"],
+            "a client-managed launch must be reportable as client-managed",
+        )
+
+    def test_namespace_health_without_attachment_fails_closed(self):
+        health = mcp_namespace_health.classify_namespace_probe(
+            "gitea-author",
+            configured=True,
+            registered_tools=["gitea_whoami"],
+            probe_result={"success": True},
+            probe_source="client_namespace",
+            process={"pid": 1234, "profile": "prgs-author", "env": {}},
+        )
+        self.assertTrue(health["provenance_fail_closed"])
+        self.assertEqual(health["provenance"], mwi.PROVENANCE_UNPROVEN)
+        self.assertIsNone(health["session_id"])
+
+    def test_no_false_reconnect_loop_for_an_owned_session(self):
+        """A proven owner must not be told to reconnect."""
+        registry = _registry()
+        worker = _attach(registry, client="claude", session="s", generation="g")
+        verdict = mwi.assess_provenance(
+            registry=registry, worker_identity=worker["identity"], env={}, now=NOW
+        )
+        self.assertFalse(verdict["fail_closed"])
+        self.assertEqual(verdict["blocker_kind"], mwi.BLOCKER_NONE)
+        self.assertEqual(verdict["reasons"], [])
+
+
+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",
     }