Fix review 646 blockers B1-B7 for cross-repo canonical roots (#973)

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

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

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

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

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

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

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

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

Closes #970

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

control_plane_db.py

@@ -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`

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

gitea_mcp_server.py

@@ -500,10 +500,38 @@ 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 _resolve_expected_repository_slug(
+    remote: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> str | None:
+    """Resolve expected repository slug from parameters, session context, or remote URL."""
+    if org and repo:
+        return session_ctx.format_repository_slug(org, repo)
+    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)
+    eff_remote = remote or bound.get("remote") or _effective_remote()
+    parsed = remote_repo_guard.parse_org_repo_from_remote_url(
+        _local_git_remote_url(eff_remote)
+    )
+    if parsed:
+        return session_ctx.format_repository_slug(parsed[0], parsed[1])
+    return None
+
+
+def _resolve_namespace_mutation_context(
+    worktree_path: str | None = None,
+    remote: str | None = None,
+) -> dict:
     """Canonical namespace workspace + repository root for guards (#460/#510/#706/#618)."""
     role = _effective_workspace_role()
     configured_root, _source = _configured_canonical_root()
+    bound = session_ctx.get_session_context() or {}
+    eff_remote = remote or bound.get("remote") or _effective_remote()
+    expected_slug = _resolve_expected_repository_slug(eff_remote)
     return nwb.resolve_namespace_mutation_context(
         role_kind=role,
         worktree_path=worktree_path,
@@ -516,9 +544,12 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic
         ),
         profile_name=get_profile().get("profile_name"),
         configured_canonical_root=configured_root,
+        expected_slug=expected_slug,
+        remote=eff_remote,
     )
 
 
+
 def _resolve_author_mutation_context(worktree_path: str | None = None) -> dict:
     """Backward-compatible alias for namespace workspace context."""
     return _resolve_namespace_mutation_context(worktree_path)
@@ -620,6 +651,9 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
         inspected_root = _get_git_root(workspace)
     process_root = ctx["process_project_root"]
     canonical_root = ctx["canonical_repo_root"]
+    crr_assessment = ctx.get("canonical_root_assessment") or {}
+    resolved_slug = crr_assessment.get("resolved_slug")
+    expected_slug = ctx.get("expected_slug")
     active_root = os.path.realpath(inspected_root or workspace)
     if active_root == canonical_root:
         dirty_scope = "control checkout"
@@ -649,6 +683,10 @@ def _preflight_workspace_details(worktree_path: str | None, dirty_files: list[st
         "workspace_healthy": not bool(
             ctx.get("bound_worktree_missing") or ctx.get("author_worktree_block")
         ),
+        "canonical_root_assessment": crr_assessment,
+        "expected_repository_slug": expected_slug,
+        "observed_repository_identity": resolved_slug,
+        "worktree_registration_result": ctx.get("in_git_worktree_list"),
     }
     if not ctx["roots_aligned"]:
         details["workspace_root_mismatch"] = (
@@ -658,6 +696,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')}",
@@ -1866,6 +1905,9 @@ def _verify_role_mutation_workspace(
     # back to the install checkout and validated Gitea-Tools/branches/ instead
     # of the target repository the namespace is actually bound to.
     _configured_root, _configured_source = _configured_canonical_root()
+    bound = session_ctx.get_session_context() or {}
+    eff_remote = remote or bound.get("remote") or _effective_remote()
+    expected_slug = _resolve_expected_repository_slug(eff_remote, org=org, repo=repo)
     assessment = nwb.assess_namespace_mutation_workspace(
         role_kind=role,
         worktree_path=worktree_path,
@@ -1880,6 +1922,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(
@@ -3414,7 +3458,7 @@ def cleanup_in_progress_for_pr(
 
 # ── Helpers ───────────────────────────────────────────────────────────────────
 
-def _effective_remote(remote: str) -> str:
+def _effective_remote(remote: str = "dadeschools") -> str:
     """If remote is the default ('dadeschools') but the active profile base_url maps to a known remote, use that remote instead."""
     try:
         profile = get_profile()
@@ -13547,6 +13591,162 @@ def gitea_scan_already_landed_open_prs(
     }
 
 
+@mcp.tool()
+def gitea_audit_missing_worktree_bindings(
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: audit and classify worktree bindings whose paths are missing on disk (#970).
+
+    Correlates missing-path bindings from control-plane leases, session checkpoints,
+    and issue locks with repository, host, branch, issue/PR, session, and lease state.
+    Distinguishes deleted worktrees from moved paths, host/mount failures, and live
+    leases/sessions.
+
+    Args:
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with audit counts, missing binding classifications, and resolution status.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    import missing_worktree_reconcile
+    db, _ = _control_plane_db_or_error()
+    root = _canonical_local_git_root()
+    h, o, r = _resolve(remote, host, org, repo)
+
+    return missing_worktree_reconcile.audit_missing_worktree_bindings(
+        db,
+        project_root=root,
+        remote=remote,
+        org=o,
+        repo=r,
+        host=h,
+    )
+
+
+@mcp.tool()
+def gitea_reconcile_missing_worktree_bindings(
+    dry_run: bool = True,
+    # Deprecated: retained so callers that still pass it get an explicit deny.
+    operator_authorized: bool = False,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Audit and safely resolve (retire) missing worktree path bindings (#970).
+
+    Identifies confirmed stale deleted worktree bindings, re-validates them
+    against a fresh authoritative read immediately before mutation to prevent
+    recreation and ownership races, and retires only the exact stale bindings
+    while preserving unrelated worktrees and Git metadata.
+
+    Apply mode (``dry_run=False``) is authorized **server-side** (#970 review
+    644 B2): it requires the reconciler-only ``gitea.branch.delete``
+    capability, the resolved ``reconcile_missing_worktree_bindings`` task
+    capability, and an active cleanup phase minted through
+    ``gitea_authorize_reconciliation_cleanup_phase``. A caller-supplied
+    ``operator_authorized`` is never authorization evidence (#709 F1 /
+    review 434) and is rejected outright. Dry-run stays available to any
+    ``gitea.read`` profile and never mutates.
+
+    Args:
+        dry_run: If True (default), reports planned mutations without modifying state.
+        operator_authorized: Rejected. Authorization is a server-side artifact.
+        remote: Known instance — 'dadeschools' or 'prgs'.
+        host: Override the Gitea host.
+        org: Override the owner/organization.
+        repo: Override the repository name.
+
+    Returns:
+        dict with before/after audit state, resolutions, and dimension status.
+    """
+    import missing_worktree_reconcile
+
+    # Explicitly reject the self-assertable Boolean before anything else, so it
+    # can never combine with a legitimate gate to authorize a mutation.
+    if operator_authorized:
+        return {
+            "success": False,
+            "performed": False,
+            "reason": "operator_authorized_rejected",
+            "reasons": [missing_worktree_reconcile.OPERATOR_AUTHORIZED_REJECTION],
+            "exact_next_action": (
+                "authorize cleanup via gitea_authorize_reconciliation_cleanup_phase "
+                "from a reconciler profile, then re-run with dry_run=False"
+            ),
+        }
+
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "performed": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    db, db_errs = _control_plane_db_or_error()
+    root = _canonical_local_git_root()
+    h, o, r = _resolve(remote, host, org, repo)
+
+    profile = get_profile()
+    role = (profile.get("role") or profile.get("role_kind") or "").strip().lower()
+
+    cleanup_authorization = None
+    if not dry_run:
+        cleanup_task = missing_worktree_reconcile.CLEANUP_TASK
+        required_permission = task_capability_map.required_permission(cleanup_task)
+        capability_blockers = _profile_operation_gate(required_permission)
+        role_matches = role == task_capability_map.required_role(cleanup_task)
+        cleanup_authorization = missing_worktree_reconcile.assess_cleanup_authorization(
+            role=role,
+            capability_blockers=capability_blockers,
+            operator_authorized=False,
+            task_capability_resolved=(not capability_blockers) and role_matches,
+        )
+        if not cleanup_authorization.get("authorized"):
+            return {
+                "success": False,
+                "performed": False,
+                "reason": "cleanup_authorization_required",
+                "reasons": cleanup_authorization.get("reasons") or [],
+                "cleanup_authorization": cleanup_authorization,
+                "permission_report": _permission_block_report(required_permission),
+                "exact_next_action": (
+                    "resolve the reconcile_missing_worktree_bindings capability "
+                    "from a reconciler profile and authorize cleanup via "
+                    "gitea_authorize_reconciliation_cleanup_phase"
+                ),
+            }
+
+    return missing_worktree_reconcile.reconcile_missing_worktree_bindings(
+        db,
+        project_root=root,
+        remote=remote,
+        org=o,
+        repo=r,
+        host=h,
+        dry_run=dry_run,
+        operator_authorized=False,
+        cleanup_authorization=cleanup_authorization,
+    )
+
+
 @mcp.tool()
 def gitea_audit_worktree_cleanup(
     remote: str = "dadeschools",
@@ -15121,22 +15321,17 @@ def _current_runtime_mode_report(refresh: bool = False) -> dict:
     workspace_root = None
     aligned = None
     canonical_root = None
+    resolved_slug = None
     try:
         ctx = _resolve_namespace_mutation_context(None)
         workspace_root = ctx.get("workspace_path")
         canonical_root = ctx.get("canonical_repo_root")
-        # Alignment keeps its established repository-level meaning (#615 F1):
-        # does this namespace target the repository the process is installed in,
-        # i.e. canonical_repo_root == process_project_root. It is deliberately
-        # NOT path equality between the task workspace and the process root --
-        # the global worktree rule requires task work to live in a branches/
-        # worktree, so that comparison would classify every correctly bound
-        # author, reviewer, and merger session as unsafe.
         aligned = ctx.get("roots_aligned")
+        crr_assessment = ctx.get("canonical_root_assessment") or {}
+        resolved_slug = crr_assessment.get("resolved_slug")
     except Exception:
-        # An unresolvable binding is reported as unknown alignment, never as
-        # proof of alignment.
         aligned = None
+        resolved_slug = None
     try:
         profile_name = get_profile()["profile_name"]
     except Exception:
@@ -15149,6 +15344,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(),

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,30 @@ 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:
-        canonical_root = os.path.realpath(configured)
+
+    configured_val = (configured_canonical_root or "").strip()
+    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,
+        )
+        canonical_root = crr_assessment["canonical_repo_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,
+        }
+        canonical_root = crr_assessment["canonical_repo_root"]
+        roots_aligned = (canonical_root == process_root)
 
     durable: dict | None = None
     if role == "author":
@@ -234,7 +311,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 +326,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 +467,8 @@ def format_namespace_workspace_binding_error(
 def assess_namespace_mutation_workspace(
     *,
     role_kind: str,
-    worktree_path: str | None,
-    worktree: str | None,
+    worktree_path: str | None = 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 +476,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 +490,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 +515,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 +552,27 @@ def assess_namespace_mutation_workspace(
             )
             if branches["block"]:
                 reasons.extend(branches["reasons"])
-    elif (
-        role == "reviewer"
-        and mutation_workspace == process_root
-        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
-    ):
-        reasons.append(
-            f"{role} mutation blocked: workspace is the stable control checkout; "
-            f"create or reconnect to a session-owned worktree under branches/ "
-            f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / "
-            f"{ACTIVE_WORKTREE_ENV}"
-        )
-    elif (
-        role in {"reviewer", "merger"}
-        and mutation_workspace != process_root
-        and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"])
-    ):
-        reasons.append(
-            f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
-            f"'{ctx['canonical_repo_root']}/branches/'"
-        )
+    elif role in {"reviewer", "merger"}:
+        if mutation_workspace == process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
+            reasons.append(
+                f"{role} mutation blocked: workspace is the stable control checkout; "
+                f"create or reconnect to a session-owned worktree under branches/ "
+                f"or set {ROLE_WORKTREE_ENVS.get(role, ACTIVE_WORKTREE_ENV)} / "
+                f"{ACTIVE_WORKTREE_ENV}"
+            )
+        elif mutation_workspace != process_root and not amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
+            reasons.append(
+                f"{role} mutation blocked: workspace '{mutation_workspace}' is not under "
+                f"'{ctx['canonical_repo_root']}/branches/'"
+            )
+
+        if path_exists and amw.is_path_under_branches(mutation_workspace, ctx["canonical_repo_root"]):
+            in_list = ctx.get("in_git_worktree_list")
+            if in_list is False:
+                reasons.append(
+                    f"{role} mutation blocked: workspace '{mutation_workspace}' is under branches/ "
+                    f"but is not registered in git worktree list for '{ctx['canonical_repo_root']}'"
+                )
 
     block = bool(reasons)
     return {

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

@@ -243,7 +243,7 @@ class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase):
             configured_canonical_root=self.target,
         )
         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(

tests/test_issue_973_cross_repo_canonical_roots.py

@@ -0,0 +1,401 @@
+"""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)
+
+        # 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,
+        )
+        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):
+        with patch("canonical_repository_root.repository_identity_slug", return_value="Scaled-Tech-Consulting/Gitea-Tools"):
+            assessment = crr.assess_canonical_repository_root(
+                configured_value=self.target_root,
+                source="test",
+                expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
+                process_project_root=self.install_root,
+            )
+            self.assertTrue(assessment["proven"])
+            self.assertFalse(assessment["block"])
+
+    def test_foreign_repository_identity_mismatch(self):
+        with patch("canonical_repository_root.repository_identity_slug", return_value="Someone-Else/Evil-Repo"):
+            assessment = crr.assess_canonical_repository_root(
+                configured_value=self.target_root,
+                source="test",
+                expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
+                process_project_root=self.install_root,
+            )
+            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):
+        with patch("canonical_repository_root.repository_identity_slug", return_value="Foreign/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=self.target_root,
+                expected_slug="Scaled-Tech-Consulting/Gitea-Tools",
+            )
+            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_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,
+            )
+            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,
+        )
+        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,
+        )
+        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,
+        )
+        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,
+        )
+        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,
+            )
+            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,
+            )
+            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="Gitea-Tools",
+            )
+            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="Gitea-Tools",
+            )
+            self.assertTrue(rel_res.get("success"), rel_res)
+            mock_clear.assert_called_once()
+
+
+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",
     }