Merge pull request 'fix(mcp): complete canonical-root consumption for cross-repository namespaces (Closes #739 )' (#740 ) from feat/issue-739-canonical-root-consumption into master

Merge pull request 'fix(mcp): accept acquired merger-lease provenance and add owner finalization (Closes #742 )' (#743 ) from fix/issue-742-merger-lease-provenance into master
fix(mcp): complete canonical-root consumption for cross-repo namespaces (Closes #739 )
2026-07-18 10:34:10 -05:00 · 2026-07-18 10:12:27 -05:00 · 2026-07-18 03:44:28 -04:00
3 changed files with 926 additions and 29 deletions
@@ -1648,6 +1648,45 @@ def _workspace_repository_slug(remote: str | None) -> str | None:
    return session_ctx.format_repository_slug(parsed[0], parsed[1])


+def _canonical_repository_slug(
+    profile: dict,
+    remote: str | None,
+) -> tuple[str | None, list[str]]:
+    """Repository identity of the *configured* canonical root, if any (#739 F2).
+
+    Returns ``(slug, reasons)``. A namespace with no configured
+    ``canonical_repository_root`` yields ``(None, [])`` so callers keep the
+    install-derived single-repository default. A configured root that cannot be
+    resolved to a repository identity yields ``(None, reasons)`` — a fail-closed
+    signal, never a silent fallback to the installation checkout, because that
+    fallback is exactly what lets a cross-repository namespace validate against
+    the wrong repository.
+
+    Env-over-profile precedence and the existence / git-toplevel / resolvable
+    remote-identity validation all live in ``crr``.
+    """
+    configured_value, configured_source = crr.configured_canonical_root(
+        profile, os.environ
+    )
+    if not configured_value:
+        return None, []
+    assessment = crr.assess_canonical_repository_root(
+        configured_value=configured_value,
+        source=configured_source,
+        expected_slug=None,
+        process_project_root=PROJECT_ROOT,
+        remote=remote,
+        require_binding=True,
+    )
+    slug = assessment.get("resolved_slug")
+    if assessment.get("block") or not slug:
+        return None, list(assessment.get("reasons") or [
+            "configured canonical repository root has no resolvable "
+            "repository identity (fail closed)"
+        ])
+    return slug, []
+
+
 def _trusted_session_repository(
    profile: dict,
    remote: str | None,
@@ -1686,32 +1725,10 @@ def _trusted_session_repository(
    # self-authorizing). Env-over-profile precedence and fail-closed validation
    # (existence, git toplevel, resolvable remote identity) are handled by
    # ``crr``; unconfigured single-repo namespaces keep the install-derived slug.
-    configured_value, configured_source = crr.configured_canonical_root(
-        profile, os.environ
-    )
-    if configured_value:
-        assessment = crr.assess_canonical_repository_root(
-            configured_value=configured_value,
-            source=configured_source,
-            expected_slug=None,
-            process_project_root=PROJECT_ROOT,
-            remote=remote,
-            require_binding=True,
-        )
-        slug = assessment.get("resolved_slug")
-        if assessment.get("block") or not slug:
-            return {
-                "org": None,
-                "repository": None,
-                "reasons": assessment.get("reasons")
-                or [
-                    "configured canonical repository root has no resolvable "
-                    "repository identity; session repository cannot be "
-                    "authorized (fail closed)"
-                ],
-            }
-    else:
-        slug = _workspace_repository_slug(remote)
+    canonical_slug, canonical_reasons = _canonical_repository_slug(profile, remote)
+    if canonical_reasons:
+        return {"org": None, "repository": None, "reasons": canonical_reasons}
+    slug = canonical_slug or _workspace_repository_slug(remote)
    scope = session_ctx.assess_repository_scope(
        workspace_slug=slug,
        allowed=allowed,
@@ -8768,8 +8785,28 @@ def _delete_branch_repository_binding_block(
      (unproven target).

    Returns a structured block dict, or ``None`` when the target is authorized.
+
+    #739 F2: the trusted identity is the session's configured
+    ``canonical_repository_root`` when the namespace declares one. Deriving it
+    from ``_workspace_repository_slug`` alone reads the *installation* checkout
+    (``_local_git_remote_url`` runs in ``PROJECT_ROOT``, always Gitea-Tools),
+    which inverts the guard for a cross-repository namespace: the genuinely
+    bound target is rejected and Gitea-Tools is accepted. A configured root that
+    cannot be resolved fails closed rather than falling back to the install
+    identity; unconfigured namespaces keep the install-derived default.
    """
-    workspace_slug = _workspace_repository_slug(remote)
+    canonical_slug, canonical_reasons = _canonical_repository_slug(
+        get_profile(), remote
+    )
+    if canonical_reasons:
+        return {
+            "success": False,
+            "performed": False,
+            "required_permission": "gitea.branch.delete",
+            "reasons": canonical_reasons,
+            "blocker_kind": "repository_binding",
+        }
+    workspace_slug = canonical_slug or _workspace_repository_slug(remote)
    workspace_parts = (
        session_ctx.parse_repository_slug(workspace_slug) if workspace_slug else None
    )
@@ -14226,6 +14263,16 @@ def gitea_get_runtime_context(
        remote: Known instance — 'dadeschools' or 'prgs'.
        host: Override the Gitea host.
    """
+    # #739 F1: normalize the remote *before* any host lookup, identity
+    # resolution, or session seeding. ``remote`` defaults to 'dadeschools'
+    # across the whole tool surface, so a caller that omits it on a prgs-hosted
+    # namespace would otherwise pin the argument default: the identity lookup
+    # below would authenticate against the wrong host, and because first-bind is
+    # first-write-wins a later, correct ``gitea_whoami(remote="prgs")`` could not
+    # repair the binding. ``gitea_whoami`` has always normalized here; this is
+    # the same call, not a new policy. Explicit non-default remotes pass through
+    # untouched, and a dadeschools-hosted profile still resolves to dadeschools.
+    remote = _effective_remote(remote)
    profile = get_profile()
    config = gitea_config.load_config()
    reveal = _reveal_endpoints()
@@ -14403,13 +14450,23 @@ def gitea_assess_master_parity(
    Never mutates and makes no network calls. Read-only operations are never
    blocked by staleness; only mutating operations fail closed while stale.

+    The top-level ``startup_head``/``current_head``/``in_parity`` fields keep
+    their original meaning: the *Gitea-Tools server implementation* only. #739
+    F3 adds two separately labelled dimensions so a cross-repository namespace
+    can tell them apart — ``server_implementation`` (this installation, restated
+    under an explicit name) and ``target_repository`` (the configured canonical
+    target checkout and its last-known remote master). Only the server dimension
+    gates mutations.
+
    Returns:
        dict with 'in_parity', 'stale', 'restart_required', 'startup_head',
-        'current_head', 'mutation_gate_enforced', 'summary', 'reasons', and a
-        'report' recovery payload when stale.
+        'current_head', 'mutation_gate_enforced', 'summary', 'reasons',
+        'server_implementation', 'target_repository', and a 'report' recovery
+        payload when stale.
    """
    parity = _current_master_parity()
    enforced = not master_parity_gate.gate_disabled()
+    canonical_root, canonical_source = _configured_canonical_root()
    out = {
        "in_parity": parity["in_parity"],
        "stale": parity["stale"],
@@ -14421,6 +14478,17 @@ def gitea_assess_master_parity(
        "summary": master_parity_gate.format_parity(parity),
        "reasons": parity["reasons"],
        "process_root": PROJECT_ROOT,
+        "server_implementation": {
+            "installation_root": PROJECT_ROOT,
+            "startup_head": parity["startup_head"],
+            "current_head": parity["current_head"],
+            "stale": parity["stale"],
+            "determinable": parity["determinable"],
+        },
+        "target_repository": master_parity_gate.assess_target_repository_parity(
+            canonical_root=canonical_root,
+            source=canonical_source,
+        ),
    }
    if parity["stale"] and enforced:
        out["report"] = master_parity_gate.parity_report(parity)
@@ -166,3 +166,128 @@ def format_parity(assessment: dict) -> str:
    if not assessment.get("determinable"):
        return "parity indeterminate (baseline or current HEAD unknown)"
    return f"in parity at {_short(assessment.get('current_head'))}"
+
+
+# ---------------------------------------------------------------------------
+# Target-repository parity (#739 F3)
+#
+# Everything above measures ONE dimension: the Gitea-Tools server's own
+# implementation commit, comparing the SHA this process was loaded from against
+# the SHA now on disk at PROJECT_ROOT. That is deliberate and is left untouched
+# — it is what proves the in-memory capability gates are current.
+#
+# It is not, however, a statement about the repository a cross-repository
+# namespace actually mutates. The assessment below is a separate, separately
+# labelled dimension for the configured canonical target repository. It never
+# feeds the mutation gate and never changes startup_head/current_head.
+# ---------------------------------------------------------------------------
+
+DEFAULT_TARGET_TRACKING_REF = "refs/remotes/origin/master"
+
+
+def _git_capture(root: str, *args: str) -> str | None:
+    """Run a read-only git command in *root*; ``None`` on any failure.
+
+    Deliberately does not honour ``GITEA_TEST_CURRENT_HEAD``: that override
+    exists to pin the *server's* HEAD, and applying it here would make a target
+    repository silently report the server's forced SHA.
+    """
+    if not root:
+        return None
+    try:
+        res = subprocess.run(
+            ["git", "-C", root, *args],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except Exception:
+        return None
+    if res.returncode != 0:
+        return None
+    return (res.stdout or "").strip() or None
+
+
+def assess_target_repository_parity(
+    *,
+    canonical_root: str | None,
+    source: str | None,
+    tracking_ref: str = DEFAULT_TARGET_TRACKING_REF,
+) -> dict:
+    """Assess the configured cross-repository target checkout.
+
+    Reports the target's canonical root, repository identity, checked-out
+    commit, and last-known remote master commit, plus whether the checkout is
+    behind that ref. No network call is made: the remote side is read from the
+    existing remote-tracking ref, so a target that has never been fetched is
+    reported as indeterminate rather than guessed at.
+
+    An unconfigured namespace is ``configured=False`` and never ``stale`` — the
+    single-repository default has no second dimension to be stale about. A
+    configured root that cannot be read is ``determinable=False`` with reasons.
+    """
+    result = {
+        "configured": bool(canonical_root),
+        "canonical_repository_root": None,
+        "source": source,
+        "repository_slug": None,
+        "checkout_head": None,
+        "tracking_ref": tracking_ref,
+        "remote_tracking_head": None,
+        "determinable": False,
+        "stale": False,
+        "reasons": [],
+    }
+    if not canonical_root:
+        result["determinable"] = True
+        return result
+
+    result["canonical_repository_root"] = canonical_root
+    if not os.path.isdir(canonical_root):
+        result["reasons"].append(
+            f"configured canonical repository root '{canonical_root}' "
+            f"does not exist or is not a directory"
+        )
+        return result
+
+    toplevel = _git_capture(canonical_root, "rev-parse", "--show-toplevel")
+    if not toplevel:
+        result["reasons"].append(
+            f"configured canonical repository root '{canonical_root}' "
+            f"is not a git checkout"
+        )
+        return result
+
+    head = _git_capture(canonical_root, "rev-parse", "HEAD")
+    if not head:
+        result["reasons"].append(
+            f"target repository HEAD could not be read at '{canonical_root}'"
+        )
+        return result
+    result["checkout_head"] = head
+    result["determinable"] = True
+
+    remote_url = _git_capture(canonical_root, "remote", "get-url", "origin")
+    if remote_url:
+        # Local import keeps this module dependency-light for its startup role.
+        import remote_repo_guard
+
+        parsed = remote_repo_guard.parse_org_repo_from_remote_url(remote_url)
+        if parsed:
+            result["repository_slug"] = f"{parsed[0]}/{parsed[1]}"
+    if not result["repository_slug"]:
+        result["reasons"].append(
+            "target repository identity could not be derived from its git remote"
+        )
+
+    tracking_head = _git_capture(canonical_root, "rev-parse", tracking_ref)
+    if not tracking_head:
+        result["reasons"].append(
+            f"remote-tracking ref '{tracking_ref}' is unknown in the target "
+            f"checkout; target staleness is indeterminate (no fetch is "
+            f"performed by this assessment)"
+        )
+        return result
+    result["remote_tracking_head"] = tracking_head
+    result["stale"] = tracking_head != head
+    return result
@@ -0,0 +1,704 @@
+"""Complete canonical-root consumption for cross-repository MCP namespaces.
+
+#706 introduced an immutable, configured ``canonical_repository_root`` and
+routed the #274 filesystem guards and the session repository *slug* through it.
+Three consumption paths were left behind, and this module drives each one
+through its production entry point:
+
+* **A — remote initialization.** ``gitea_get_runtime_context`` never normalized
+  its ``remote`` argument, while ``gitea_whoami`` did. A fresh process whose
+  first native call is the runtime-context path therefore pinned the
+  ``dadeschools`` argument default instead of the profile's configured remote,
+  and — because first-bind is first-write-wins — no later correct call could
+  repair it.
+
+* **C — reconciler branch deletion.** The delete-branch repository-binding
+  guard derived its expected slug from ``_workspace_repository_slug``, which
+  reads the git remote of the *installation* checkout (always Gitea-Tools),
+  rather than from the session's configured canonical root.
+
+* **D — parity evidence.** ``gitea_assess_master_parity`` proves Gitea-Tools
+  *server implementation* parity only, which is intentional. It carried no
+  target-repository dimension at all, so a cross-repository namespace had no
+  evidence that its target checkout was current. The existing
+  ``startup_head``/``current_head`` semantics are preserved unchanged and the
+  target-repository assessment is reported under separately labelled fields.
+
+Groups B and E assert *intentional* behaviour (the #274 guards already consume
+the canonical root; the configuration surface already validates a
+repository-specific namespace) so that a regression in either is caught.
+
+Real git repositories are used throughout; no network calls are made.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import canonical_repository_root as crr  # noqa: E402
+import gitea_config  # noqa: E402
+import gitea_mcp_server as srv  # noqa: E402
+import master_parity_gate  # noqa: E402
+import mcp_server  # noqa: E402
+import namespace_workspace_binding as nwb  # noqa: E402
+import session_context_binding as session_ctx  # noqa: E402
+
+INSTALL_ORG = "Scaled-Tech-Consulting"
+INSTALL_REPO = "Gitea-Tools"
+INSTALL_SLUG = f"{INSTALL_ORG}/{INSTALL_REPO}"
+INSTALL_URL = f"https://gitea.prgs.cc/{INSTALL_SLUG}.git"
+
+TARGET_ORG = "Scaled-Tech-Consulting"
+TARGET_REPO = "mcp-control-plane"
+TARGET_SLUG = f"{TARGET_ORG}/{TARGET_REPO}"
+TARGET_URL = f"https://gitea.prgs.cc/{TARGET_SLUG}.git"
+
+
+def _git(cwd: str, *args: str) -> str:
+    res = subprocess.run(
+        ["git", "-C", cwd, *args], capture_output=True, text=True, check=True
+    )
+    return res.stdout.strip()
+
+
+def _init_repo(path: Path, remote_url: str, *, remote_name: str = "origin") -> str:
+    path.mkdir(parents=True, exist_ok=True)
+    _git(str(path), "init", "-q")
+    _git(str(path), "config", "user.email", "[email protected]")
+    _git(str(path), "config", "user.name", "Test")
+    _git(str(path), "remote", "add", remote_name, remote_url)
+    (path / "README.md").write_text("seed\n")
+    _git(str(path), "add", "README.md")
+    _git(str(path), "commit", "-q", "-m", "seed")
+    return os.path.realpath(str(path))
+
+
+# ---------------------------------------------------------------------------
+# Shared profile/config construction
+# ---------------------------------------------------------------------------
+_AUTHOR_OPS = [
+    "gitea.read",
+    "gitea.issue.create",
+    "gitea.issue.comment",
+    "gitea.pr.create",
+    "gitea.pr.comment",
+    "gitea.branch.create",
+    "gitea.branch.push",
+    "gitea.repo.commit",
+]
+_AUTHOR_FORBIDDEN = [
+    "gitea.pr.approve",
+    "gitea.pr.merge",
+    "gitea.pr.request_changes",
+]
+# A profile that may approve or merge must forbid authoring (gitea_config's
+# reviewer-identity deadlock rule).
+_REVIEWER_OPS = [
+    "gitea.read",
+    "gitea.pr.approve",
+    "gitea.pr.request_changes",
+    "gitea.pr.comment",
+    "gitea.issue.comment",
+]
+_REVIEWER_FORBIDDEN = ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"]
+_MERGER_OPS = ["gitea.read", "gitea.pr.merge", "gitea.pr.comment"]
+_MERGER_FORBIDDEN = [
+    "gitea.pr.create",
+    "gitea.branch.push",
+    "gitea.pr.approve",
+    "gitea.pr.request_changes",
+]
+_RECONCILER_OPS = [
+    "gitea.read",
+    "gitea.pr.close",
+    "gitea.pr.comment",
+    "gitea.branch.delete",
+]
+_RECONCILER_FORBIDDEN = [
+    "gitea.pr.create",
+    "gitea.branch.push",
+    "gitea.pr.approve",
+    "gitea.pr.merge",
+    "gitea.pr.request_changes",
+]
+
+_ROLE_MATRIX = {
+    "author": (_AUTHOR_OPS, _AUTHOR_FORBIDDEN),
+    "reviewer": (_REVIEWER_OPS, _REVIEWER_FORBIDDEN),
+    "merger": (_MERGER_OPS, _MERGER_FORBIDDEN),
+    "reconciler": (_RECONCILER_OPS, _RECONCILER_FORBIDDEN),
+}
+
+
+def _profile(
+    role: str,
+    *,
+    canonical_root: str | None = None,
+    allowed_repositories: list[str] | None = None,
+    username: str = "jcwalker3",
+) -> dict:
+    allowed, forbidden = _ROLE_MATRIX[role]
+    profile = {
+        "enabled": True,
+        "context": "prgs",
+        "role": role,
+        "username": username,
+        "base_url": "https://gitea.prgs.cc",
+        "auth": {"type": "env", "name": f"GITEA_TOKEN_PRGS_{role.upper()}"},
+        "allowed_operations": list(allowed),
+        "forbidden_operations": list(forbidden),
+        "execution_profile": f"prgs-{role}",
+    }
+    if canonical_root is not None:
+        profile["canonical_repository_root"] = canonical_root
+    if allowed_repositories is not None:
+        profile["allowed_repositories"] = list(allowed_repositories)
+    return profile
+
+
+def _config(profiles: dict) -> dict:
+    return {
+        "version": 2,
+        "rules": {"allow_runtime_switching": True},
+        "contexts": {
+            "prgs": {
+                "enabled": True,
+                "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
+            },
+            "mdcps": {
+                "enabled": True,
+                "gitea": {
+                    "enabled": True,
+                    "base_url": "https://gitea.dadeschools.net",
+                },
+            },
+        },
+        "profiles": profiles,
+    }
+
+
+class _ServerHarness(unittest.TestCase):
+    """Temp profiles.json + pinned install remote + no network."""
+
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self.tmp = self._dir.name
+        self.config_path = os.path.join(self.tmp, "profiles.json")
+        session_ctx._reset_session_context_for_testing()
+        gitea_config._active_profile_override = None
+        mcp_server._IDENTITY_CACHE.clear()
+        srv._MUTATION_AUTHORITY = None
+        # The install checkout's remote is Gitea-Tools regardless of the
+        # developer's layout; pin it so "install-derived" is deterministic.
+        self._remote_url = patch.object(
+            srv, "_local_git_remote_url", side_effect=self._install_remote_url
+        )
+        self._remote_url.start()
+
+    def tearDown(self):
+        self._remote_url.stop()
+        session_ctx._reset_session_context_for_testing()
+        gitea_config._active_profile_override = None
+        mcp_server._IDENTITY_CACHE.clear()
+        srv._MUTATION_AUTHORITY = None
+        self._dir.cleanup()
+
+    def _install_remote_url(self, remote_name):
+        return INSTALL_URL if remote_name in ("prgs", "origin") else None
+
+    def _write_config(self, profiles: dict) -> None:
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(_config(profiles)))
+
+    def _env(self, profile_name: str, **extra) -> dict:
+        env = {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": profile_name,
+            "GITEA_TOKEN_PRGS_AUTHOR": "t",
+            "GITEA_TOKEN_PRGS_REVIEWER": "t",
+            "GITEA_TOKEN_PRGS_MERGER": "t",
+            "GITEA_TOKEN_PRGS_RECONCILER": "t",
+        }
+        env.update(extra)
+        return env
+
+    def _api(self, method, url, header):
+        return {
+            "login": "jcwalker3",
+            "full_name": "Test",
+            "id": 1,
+            "email": "[email protected]",
+        }
+
+    def _live(self):
+        return patch("gitea_mcp_server.api_request", side_effect=self._api)
+
+
+# ===========================================================================
+# A. Remote initialization
+# ===========================================================================
+class TestRemoteInitializationFirstCall(_ServerHarness):
+    """A prgs namespace must never pin the dadeschools argument default."""
+
+    def setUp(self):
+        super().setUp()
+        self._write_config(
+            {"prgs-author": _profile("author", allowed_repositories=[INSTALL_SLUG])}
+        )
+
+    def test_runtime_context_first_call_does_not_bind_default_remote(self):
+        """Runtime-context as the FIRST native call, relying on the default."""
+        with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
+            srv.gitea_get_runtime_context()
+            ctx = session_ctx.get_session_context()
+        self.assertIsNotNone(ctx, "first call must establish a session binding")
+        self.assertEqual(ctx["remote"], "prgs")
+        self.assertEqual(ctx["host"], "gitea.prgs.cc")
+        self.assertEqual(ctx["org"], INSTALL_ORG)
+        self.assertEqual(ctx["repository"], INSTALL_REPO)
+
+    def test_runtime_context_first_call_reports_effective_remote(self):
+        with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
+            result = srv.gitea_get_runtime_context()
+        self.assertEqual(result["remote"], "prgs")
+
+    def test_whoami_first_call_remains_correct(self):
+        """Control: the already-normalizing path is unchanged."""
+        with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
+            srv.gitea_whoami()
+            ctx = session_ctx.get_session_context()
+        self.assertEqual(ctx["remote"], "prgs")
+        self.assertEqual(ctx["host"], "gitea.prgs.cc")
+
+    def test_explicit_remote_argument_still_honoured(self):
+        """Normalization only fills the default; explicit values are untouched."""
+        with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
+            result = srv.gitea_get_runtime_context(remote="prgs")
+        self.assertEqual(result["remote"], "prgs")
+
+    def test_mdcps_profile_default_is_not_rewritten_to_prgs(self):
+        """A dadeschools-hosted profile keeps the dadeschools remote."""
+        profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
+        profile["context"] = "mdcps"
+        profile["base_url"] = "https://gitea.dadeschools.net"
+        self._write_config({"prgs-author": profile})
+        with patch.dict(os.environ, self._env("prgs-author"), clear=False), self._live():
+            result = srv.gitea_get_runtime_context()
+        self.assertEqual(result["remote"], "dadeschools")
+
+
+# ===========================================================================
+# B. Remote/repository guard (intentional behaviour — regression fence)
+# ===========================================================================
+class TestCanonicalRootGuardBinding(_ServerHarness):
+    def setUp(self):
+        super().setUp()
+        self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
+        self.install_root = _init_repo(Path(self.tmp) / "install", INSTALL_URL)
+
+    def test_canonical_target_root_resolves_to_target_slug(self):
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.target_root,
+            source="profile.canonical_repository_root",
+            expected_slug=None,
+            process_project_root=self.install_root,
+            remote="prgs",
+            require_binding=True,
+        )
+        self.assertFalse(got.get("block"), got.get("reasons"))
+        self.assertEqual(got["resolved_slug"], TARGET_SLUG)
+
+    def test_install_root_identity_remains_gitea_tools(self):
+        got = crr.assess_canonical_repository_root(
+            configured_value=None,
+            source=None,
+            expected_slug=None,
+            process_project_root=self.install_root,
+            remote="prgs",
+        )
+        self.assertEqual(
+            os.path.realpath(got["canonical_repo_root"]), self.install_root
+        )
+
+    def test_guards_validate_against_canonical_target_root(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="author",
+            worktree_path=None,
+            process_project_root=self.install_root,
+            configured_canonical_root=self.target_root,
+        )
+        self.assertEqual(
+            os.path.realpath(ctx["canonical_repo_root"]), self.target_root
+        )
+
+    def test_explicit_coordinates_cannot_override_canonical_binding(self):
+        """Explicit org/repo may confirm, never authorize, a binding."""
+        confirm = session_ctx.assess_repository_override(
+            requested_org=TARGET_ORG,
+            requested_repo=TARGET_REPO,
+            bound_org=TARGET_ORG,
+            bound_repo=TARGET_REPO,
+        )
+        self.assertFalse(confirm.get("block"))
+        override = session_ctx.assess_repository_override(
+            requested_org=TARGET_ORG,
+            requested_repo=TARGET_REPO,
+            bound_org=INSTALL_ORG,
+            bound_repo=INSTALL_REPO,
+        )
+        self.assertTrue(override.get("block"))
+
+    def test_gitea_tools_rooted_namespace_cannot_reach_target_repository(self):
+        """No canonical root configured → session stays pinned to Gitea-Tools."""
+        profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
+        with patch.dict(os.environ, {}, clear=False):
+            os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
+            resolved = srv._trusted_session_repository(
+                profile, "prgs", for_mutation=True
+            )
+        self.assertEqual(resolved["repository"], INSTALL_REPO)
+        self.assertNotEqual(resolved["repository"], TARGET_REPO)
+
+    def test_target_rooted_namespace_binds_target_repository(self):
+        profile = _profile(
+            "author",
+            canonical_root=self.target_root,
+            allowed_repositories=[TARGET_SLUG],
+        )
+        resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
+        self.assertEqual(resolved["org"], TARGET_ORG)
+        self.assertEqual(resolved["repository"], TARGET_REPO)
+
+
+# ===========================================================================
+# C. Reconciler branch deletion
+# ===========================================================================
+class TestDeleteBranchRepositoryBinding(_ServerHarness):
+    """The binding guard must consult the canonical root, not PROJECT_ROOT.
+
+    No branch is ever deleted here: only the pre-deletion binding guard is
+    exercised.
+    """
+
+    def setUp(self):
+        super().setUp()
+        self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
+        self._write_config(
+            {
+                "prgs-reconciler": _profile(
+                    "reconciler",
+                    canonical_root=self.target_root,
+                    allowed_repositories=[TARGET_SLUG],
+                ),
+                "gt-reconciler": _profile(
+                    "reconciler", allowed_repositories=[INSTALL_SLUG]
+                ),
+            }
+        )
+
+    def test_target_rooted_reconciler_validates_target_deletion(self):
+        with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
+            block = srv._delete_branch_repository_binding_block(
+                "prgs", org=TARGET_ORG, repo=TARGET_REPO
+            )
+        self.assertIsNone(block, block)
+
+    def test_target_rooted_reconciler_fails_closed_for_install_repository(self):
+        with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
+            block = srv._delete_branch_repository_binding_block(
+                "prgs", org=INSTALL_ORG, repo=INSTALL_REPO
+            )
+        self.assertIsNotNone(block)
+        self.assertEqual(block["blocker_kind"], "repository_binding")
+        self.assertFalse(block["performed"])
+
+    def test_install_rooted_reconciler_fails_closed_for_target_repository(self):
+        with patch.dict(os.environ, self._env("gt-reconciler"), clear=False):
+            os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
+            block = srv._delete_branch_repository_binding_block(
+                "prgs", org=TARGET_ORG, repo=TARGET_REPO
+            )
+        self.assertIsNotNone(block)
+        self.assertEqual(block["blocker_kind"], "repository_binding")
+
+    def test_env_configured_canonical_root_is_honoured(self):
+        env = self._env("gt-reconciler")
+        env[crr.CANONICAL_ROOT_ENV] = self.target_root
+        with patch.dict(os.environ, env, clear=False):
+            allowed = srv._delete_branch_repository_binding_block(
+                "prgs", org=TARGET_ORG, repo=TARGET_REPO
+            )
+            blocked = srv._delete_branch_repository_binding_block(
+                "prgs", org=INSTALL_ORG, repo=INSTALL_REPO
+            )
+        self.assertIsNone(allowed, allowed)
+        self.assertIsNotNone(blocked)
+
+    def test_unresolvable_canonical_root_fails_closed(self):
+        env = self._env("gt-reconciler")
+        env[crr.CANONICAL_ROOT_ENV] = os.path.join(self.tmp, "does-not-exist")
+        with patch.dict(os.environ, env, clear=False):
+            block = srv._delete_branch_repository_binding_block(
+                "prgs", org=TARGET_ORG, repo=TARGET_REPO
+            )
+        self.assertIsNotNone(block)
+        self.assertEqual(block["blocker_kind"], "repository_binding")
+
+    def test_no_request_parameter_can_override_canonical_identity(self):
+        """Every explicit coordinate that is not the canonical target is
+        refused; none of them can *establish* the binding."""
+        # Both repositories share an org, so a wrong *org* case must name a
+        # genuinely different owner to be meaningful.
+        cases = [
+            (INSTALL_ORG, INSTALL_REPO),
+            (TARGET_ORG, INSTALL_REPO),
+            (TARGET_ORG, "some-other-repo"),
+            ("someone-else", TARGET_REPO),
+        ]
+        with patch.dict(os.environ, self._env("prgs-reconciler"), clear=False):
+            for org, repo in cases:
+                with self.subTest(org=org, repo=repo):
+                    block = srv._delete_branch_repository_binding_block(
+                        "prgs", org=org, repo=repo
+                    )
+                    self.assertIsNotNone(block, f"{org}/{repo} must fail closed")
+
+
+# ===========================================================================
+# D. Parity semantics
+# ===========================================================================
+class TestTargetRepositoryParityAssessment(unittest.TestCase):
+    """Server-implementation parity is preserved; target parity is additive."""
+
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self.tmp = self._dir.name
+
+    def tearDown(self):
+        self._dir.cleanup()
+
+    def _target(self) -> str:
+        return _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
+
+    def test_unconfigured_target_is_reported_not_stale(self):
+        got = master_parity_gate.assess_target_repository_parity(
+            canonical_root=None, source=None
+        )
+        self.assertFalse(got["configured"])
+        self.assertFalse(got["stale"])
+        self.assertIsNone(got["canonical_repository_root"])
+
+    def test_configured_target_reports_checkout_head_and_slug(self):
+        root = self._target()
+        head = _git(root, "rev-parse", "HEAD")
+        got = master_parity_gate.assess_target_repository_parity(
+            canonical_root=root, source="profile.canonical_repository_root"
+        )
+        self.assertTrue(got["configured"])
+        self.assertEqual(got["canonical_repository_root"], root)
+        self.assertEqual(got["checkout_head"], head)
+        self.assertEqual(got["repository_slug"], TARGET_SLUG)
+
+    def test_target_stale_when_remote_tracking_ref_is_ahead(self):
+        root = self._target()
+        head = _git(root, "rev-parse", "HEAD")
+        # Simulate a fetched remote-tracking ref that has advanced.
+        _git(root, "checkout", "-q", "-b", "advanced")
+        (Path(root) / "next.md").write_text("next\n")
+        _git(root, "add", "next.md")
+        _git(root, "commit", "-q", "-m", "advance")
+        advanced = _git(root, "rev-parse", "HEAD")
+        _git(root, "update-ref", "refs/remotes/origin/master", advanced)
+        _git(root, "checkout", "-q", "--detach", head)
+        got = master_parity_gate.assess_target_repository_parity(
+            canonical_root=root, source="profile.canonical_repository_root"
+        )
+        self.assertEqual(got["checkout_head"], head)
+        self.assertEqual(got["remote_tracking_head"], advanced)
+        self.assertTrue(got["stale"])
+
+    def test_missing_root_is_not_determinable_and_fails_closed(self):
+        got = master_parity_gate.assess_target_repository_parity(
+            canonical_root=os.path.join(self.tmp, "absent"),
+            source="profile.canonical_repository_root",
+        )
+        self.assertTrue(got["configured"])
+        self.assertFalse(got["determinable"])
+        self.assertTrue(got["reasons"])
+
+
+class TestParityToolEvidenceDimensions(_ServerHarness):
+    def setUp(self):
+        super().setUp()
+        self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
+        self._write_config(
+            {
+                "prgs-author": _profile(
+                    "author",
+                    canonical_root=self.target_root,
+                    allowed_repositories=[TARGET_SLUG],
+                )
+            }
+        )
+
+    def test_existing_server_parity_semantics_are_unchanged(self):
+        with patch.dict(os.environ, self._env("prgs-author"), clear=False):
+            result = srv.gitea_assess_master_parity(remote="prgs")
+        # startup_head/current_head keep their Gitea-Tools implementation
+        # meaning and must not be redefined to the target repository.
+        self.assertEqual(result["process_root"], srv.PROJECT_ROOT)
+        self.assertEqual(
+            result["startup_head"], srv._STARTUP_PARITY.get("startup_head")
+        )
+        self.assertIn("in_parity", result)
+
+    def test_evidence_distinguishes_server_and_target_dimensions(self):
+        with patch.dict(os.environ, self._env("prgs-author"), clear=False):
+            result = srv.gitea_assess_master_parity(remote="prgs")
+        server = result["server_implementation"]
+        self.assertEqual(server["installation_root"], srv.PROJECT_ROOT)
+        self.assertEqual(server["current_head"], result["current_head"])
+        self.assertIn("stale", server)
+
+        target = result["target_repository"]
+        self.assertTrue(target["configured"])
+        self.assertEqual(target["canonical_repository_root"], self.target_root)
+        self.assertEqual(target["repository_slug"], TARGET_SLUG)
+        self.assertEqual(
+            target["checkout_head"], _git(self.target_root, "rev-parse", "HEAD")
+        )
+        self.assertIn("stale", target)
+
+    def test_unconfigured_namespace_reports_target_as_unconfigured(self):
+        self._write_config(
+            {"prgs-author": _profile("author", allowed_repositories=[INSTALL_SLUG])}
+        )
+        env = self._env("prgs-author")
+        with patch.dict(os.environ, env, clear=False):
+            os.environ.pop(crr.CANONICAL_ROOT_ENV, None)
+            result = srv.gitea_assess_master_parity(remote="prgs")
+        self.assertFalse(result["target_repository"]["configured"])
+
+
+# ===========================================================================
+# E. Startup / configuration validation for a candidate namespace set
+# ===========================================================================
+class TestCandidateNamespaceConfiguration(_ServerHarness):
+    """Four repository-specific profiles for the target repository."""
+
+    def setUp(self):
+        super().setUp()
+        self.target_root = _init_repo(Path(self.tmp) / "mcp-control-plane", TARGET_URL)
+        self.profiles = {
+            f"mcpcp-{role}": _profile(
+                role,
+                canonical_root=self.target_root,
+                allowed_repositories=[TARGET_SLUG],
+            )
+            for role in ("author", "reviewer", "merger", "reconciler")
+        }
+        self._write_config(self.profiles)
+
+    def test_configuration_audit_succeeds(self):
+        with patch.dict(os.environ, self._env("mcpcp-author"), clear=False):
+            audit = srv.gitea_audit_config()
+        self.assertTrue(audit.get("configured"))
+        names = {row["name"] for row in audit["profiles"]}
+        self.assertEqual(names, set(self.profiles))
+
+    def test_no_inline_credentials_are_present(self):
+        raw = json.loads(Path(self.config_path).read_text())
+        for name, profile in raw["profiles"].items():
+            with self.subTest(profile=name):
+                self.assertNotIn("token", profile)
+                self.assertNotIn("password", profile)
+                self.assertNotIn("secret", profile)
+                self.assertIn(profile["auth"]["type"], ("env", "keychain"))
+
+    def test_bind_time_validation_succeeds_for_target_repository(self):
+        for name, profile in self.profiles.items():
+            with self.subTest(profile=name):
+                resolved = srv._trusted_session_repository(
+                    profile, "prgs", for_mutation=True
+                )
+                self.assertEqual(resolved["reasons"], [])
+                self.assertEqual(resolved["org"], TARGET_ORG)
+                self.assertEqual(resolved["repository"], TARGET_REPO)
+
+    def test_wrong_repository_root_fails_closed(self):
+        wrong_root = _init_repo(Path(self.tmp) / "wrong", INSTALL_URL)
+        profile = _profile(
+            "author", canonical_root=wrong_root, allowed_repositories=[TARGET_SLUG]
+        )
+        resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
+        self.assertIsNone(resolved["repository"])
+        self.assertTrue(resolved["reasons"])
+
+    def test_existing_install_profiles_are_unchanged_in_behaviour(self):
+        profile = _profile("author", allowed_repositories=[INSTALL_SLUG])
+        resolved = srv._trusted_session_repository(profile, "prgs", for_mutation=True)
+        self.assertEqual(resolved["org"], INSTALL_ORG)
+        self.assertEqual(resolved["repository"], INSTALL_REPO)
+
+    def _denied(self, profile: dict, operation: str) -> bool:
+        ok, _ = gitea_config.check_operation(
+            operation,
+            profile["allowed_operations"],
+            profile["forbidden_operations"],
+        )
+        return not ok
+
+    def test_role_separation_is_enforced(self):
+        expectations = {
+            "author": [
+                "gitea.pr.approve",
+                "gitea.pr.merge",
+                "gitea.pr.request_changes",
+            ],
+            "reviewer": ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"],
+            "merger": [
+                "gitea.pr.create",
+                "gitea.branch.push",
+                "gitea.pr.approve",
+                "gitea.pr.request_changes",
+            ],
+            "reconciler": [
+                "gitea.pr.create",
+                "gitea.branch.push",
+                "gitea.pr.approve",
+                "gitea.pr.merge",
+                "gitea.pr.request_changes",
+            ],
+        }
+        for role, denied_ops in expectations.items():
+            profile = self.profiles[f"mcpcp-{role}"]
+            for op in denied_ops:
+                with self.subTest(role=role, operation=op):
+                    self.assertTrue(
+                        self._denied(profile, op),
+                        f"{role} must not be permitted {op}",
+                    )
+
+    def test_each_role_retains_its_own_capability(self):
+        permitted = {
+            "author": "gitea.pr.create",
+            "reviewer": "gitea.pr.approve",
+            "merger": "gitea.pr.merge",
+            "reconciler": "gitea.branch.delete",
+        }
+        for role, op in permitted.items():
+            with self.subTest(role=role, operation=op):
+                self.assertFalse(self._denied(self.profiles[f"mcpcp-{role}"], op))
+
+
+if __name__ == "__main__":
+    unittest.main()