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

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()
@@ -15277,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:
@@ -15305,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 {

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