Merge pull request 'feat(mcp): use a 10-minute sliding TTL for reviewer and merger PR leases (Closes #747)' (#748) from feat/issue-747-sliding-lease-ttl into master

Reviewer and merger PR leases minted a fixed 120-minute expiry (#407 AC6/AC7)
and derived staleness from two further activity bands: stale at 30 minutes,
reclaimable at 60. A session that died — daemon crash, transport flap, client
restart — therefore kept a PR blocked for an hour before anyone could reclaim
it, and two hours before manual cleanup was sanctioned. #718 records the
resulting deadlock: a merge stalled behind a reviewer lease that had stopped
being live long before it stopped being authoritative.

The flaw was using a long fixed expiry as a proxy for "the owner is probably
still alive" instead of making the owner continuously prove liveness.

Sliding window
--------------
`LEASE_TTL_MINUTES = 10` now governs acquisition, and every write of the lease
marker re-derives `expires_at` from the moment of the write, so each heartbeat
slides the window forward. An actively heartbeating session is never evicted
and has no maximum lifetime; a dead one releases its hold within one TTL.

`LEASE_RENEWAL_MINUTES` is named separately from the acquisition TTL. Renewal
previously had no seam at all: the heartbeat slid the expiry only as a side
effect of re-defaulting the acquisition constant, so the two durations could
not be reasoned about or tuned independently. `format_lease_body` now takes an
explicit `ttl_minutes`, and the heartbeat passes the renewal window rather than
relying on that default.

Merger leases acquire through the same lease-body formatter, so they inherit
the identical window by construction rather than by a parallel constant.

Removal of the reclaim tier
---------------------------
`classify_lease_freshness` no longer returns `reclaimable`. Beyond being an
extra waiting tier, that band is unreachable under a sliding TTL: a heartbeat
stamps `last_activity` and `expires_at` together, so a lease idle for a full
TTL is necessarily already expired. Expiry is now the only takeover gate.

No reclaim path is lost. `find_active_reviewer_lease` already ignores expired
markers, so an expired foreign lease never gated acquisition; and the
`foreign_expired` classification carries the same
`NEXT_ACTION_RELEASE_EXPIRED_LEASE` the retired `foreign_reclaimable` did. The
two updated tests in `test_reviewer_pr_lease.py` assert exactly that: the
classification label changes, the sanctioned next action does not.

`STALE_WARNING_MINUTES` drops to 5 — half the window — so the warning still
fires while the owner can heartbeat and recover.

Diagnostics
-----------
Adds `lease_seconds_remaining`, and the heartbeat tool now returns
`ttl_minutes`, `expires_at`, and `seconds_remaining`, so an operator can
distinguish "held and live" from "held and dying" instead of only seeing that
a lease exists. All additions are additive; no existing key changed.

Also removes `pr_work_lease.DEFAULT_REVIEWER_LEASE_TTL_MINUTES`, a duplicate of
the reviewer TTL with no readers anywhere in the tree, which could only drift.

Legacy markers minted under the old 120-minute TTL still parse and are judged
against their own recorded `expires_at`, so no lease is retroactively expired
by this change.

Full suite: 3425 passed, 6 skipped, 2 failed. Both failures
(`test_issue_702_review_findings_f1_f6`, `test_reconciler_supersession_close`)
reproduce identically on clean master b05075fd25 and are pre-existing.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01QxXHZ7rqXtLgTusngaWgKZ

canonical_repository_root.py

@@ -0,0 +1,267 @@
+"""Immutable canonical repository root for cross-repository namespaces (#706).
+
+The Gitea-Tools MCP server historically derived the ``canonical_repo_root`` from
+the *install checkout* the server script lives in
+(``PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))``). A namespace
+that runs the same server script against an *external* repository (e.g.
+``eagenda-author`` targeting ``eAgenda``) then failed every mutation: the
+branches-only / worktree-membership guards (#274) compared the task workspace
+against the Gitea-Tools ``.git`` directory, which it can never belong to.
+
+This module separates two distinct concepts:
+
+* the immutable code/install root (``PROJECT_ROOT``) — where the server lives, and
+* the namespace-scoped **canonical repository root** — the working root of the
+  repository whose issues/PRs the namespace mutates.
+
+The canonical repository root is configured per namespace (profile field or an
+environment variable, typically set alongside the namespace ``cwd`` in the MCP
+config). It is validated (existence, git identity, git common-directory
+membership) and pinned immutably into the session context so a later call cannot
+forge or swap it. When *no* binding is configured the single-repo default is
+preserved unchanged: the canonical root is derived from the process checkout.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from typing import Mapping
+
+import remote_repo_guard
+
+# Namespace-scoped override, typically exported next to the server ``cwd`` in the
+# MCP config for a cross-repository namespace.
+CANONICAL_ROOT_ENV = "GITEA_CANONICAL_REPOSITORY_ROOT"
+
+# Candidate git remote names probed when deriving repository identity.
+_IDENTITY_REMOTE_CANDIDATES = ("prgs", "origin", "dadeschools", "mdcps")
+
+
+def configured_canonical_root(
+    profile: Mapping | None,
+    env: Mapping | None,
+) -> tuple[str | None, str | None]:
+    """Return ``(value, source)`` for the declared canonical repository root.
+
+    Precedence: the ``GITEA_CANONICAL_REPOSITORY_ROOT`` environment variable
+    (namespace-scoped) overrides the profile ``canonical_repository_root``
+    field. Blank values are treated as unset. Returns ``(None, None)`` when no
+    binding is declared (the single-repo default).
+    """
+    env_map = env if env is not None else os.environ
+    env_val = (env_map.get(CANONICAL_ROOT_ENV) or "").strip()
+    if env_val:
+        return env_val, f"{CANONICAL_ROOT_ENV} environment variable"
+    if profile:
+        prof_val = (profile.get("canonical_repository_root") or "").strip()
+        if prof_val:
+            return prof_val, "profile canonical_repository_root"
+    return None, None
+
+
+def resolve_repo_toplevel(path: str) -> str | None:
+    """Realpath of the git working-tree top level for *path*, or None."""
+    text = (path or "").strip()
+    if not text:
+        return None
+    try:
+        res = subprocess.run(
+            ["git", "-C", text, "rev-parse", "--show-toplevel"],
+            capture_output=True,
+            text=True,
+            check=True,
+        )
+    except Exception:
+        return None
+    top = (res.stdout or "").strip()
+    return os.path.realpath(top) if top else None
+
+
+def repository_identity_slug(path: str, *, remote: str | None = None) -> str | None:
+    """``owner/repository`` derived from a git remote configured at *path*.
+
+    Tries the caller-named remote first, then a small set of known remote names,
+    then whatever remote the repository actually has. Returns None when no remote
+    URL is parseable (identity cannot be proven).
+    """
+    text = (path or "").strip()
+    if not text:
+        return None
+
+    ordered: list[str] = []
+    for name in (remote, *_IDENTITY_REMOTE_CANDIDATES):
+        clean = (name or "").strip()
+        if clean and clean not in ordered:
+            ordered.append(clean)
+
+    try:
+        listed = subprocess.run(
+            ["git", "-C", text, "remote"],
+            capture_output=True,
+            text=True,
+            check=True,
+        ).stdout.split()
+    except Exception:
+        listed = []
+    for name in listed:
+        if name and name not in ordered:
+            ordered.append(name)
+
+    for name in ordered:
+        try:
+            url = subprocess.run(
+                ["git", "-C", text, "remote", "get-url", name],
+                capture_output=True,
+                text=True,
+                check=True,
+            ).stdout.strip()
+        except Exception:
+            continue
+        parsed = remote_repo_guard.parse_org_repo_from_remote_url(url)
+        if parsed:
+            return f"{parsed[0]}/{parsed[1]}"
+    return None
+
+
+def assess_canonical_repository_root(
+    *,
+    configured_value: str | None,
+    source: str | None,
+    expected_slug: str | None,
+    process_project_root: str,
+    remote: str | None = None,
+    require_binding: bool = False,
+) -> dict:
+    """Validate the canonical repository root binding, failing closed on forgery.
+
+    Returns a dict with ``proven`` / ``block`` / ``reasons`` plus the resolved
+    ``canonical_repo_root`` (the value downstream guards must use),
+    ``configured`` (whether a cross-repo binding was declared),
+    ``resolved_slug`` and ``source``.
+
+    Without a configured binding the single-repo default is preserved: the
+    canonical root is derived from *process_project_root* and never blocks
+    (unless *require_binding* explicitly demands one).
+
+    With a configured binding the path must exist, be a git repository, and —
+    when *expected_slug* is known — carry a matching repository identity. A
+    mismatched or (when *require_binding*) unprovable identity is a forged or
+    conflicting binding and fails closed.
+    """
+    process_root = os.path.realpath(process_project_root)
+    declared = (configured_value or "").strip()
+
+    if not declared:
+        if require_binding:
+            return _assessment(
+                proven=False,
+                reasons=[
+                    "no canonical_repository_root configured for a cross-repository "
+                    f"namespace; set {CANONICAL_ROOT_ENV} or the profile "
+                    "canonical_repository_root field (fail closed)"
+                ],
+                configured=False,
+                canonical_repo_root=process_root,
+                resolved_slug=None,
+                source=None,
+            )
+        # Single-repo default: canonical root follows the install checkout.
+        derived = resolve_repo_toplevel(process_root) or process_root
+        return _assessment(
+            proven=True,
+            reasons=[],
+            configured=False,
+            canonical_repo_root=derived,
+            resolved_slug=None,
+            source=None,
+        )
+
+    real = os.path.realpath(os.path.abspath(declared))
+    if not os.path.isdir(real):
+        return _assessment(
+            proven=False,
+            reasons=[
+                f"configured canonical repository root '{real}' does not exist "
+                "or is not a directory (fail closed)"
+            ],
+            configured=True,
+            canonical_repo_root=real,
+            resolved_slug=None,
+            source=source,
+        )
+
+    toplevel = resolve_repo_toplevel(real)
+    if not toplevel:
+        return _assessment(
+            proven=False,
+            reasons=[
+                f"configured canonical repository root '{real}' is not a git "
+                "repository (fail closed)"
+            ],
+            configured=True,
+            canonical_repo_root=real,
+            resolved_slug=None,
+            source=source,
+        )
+
+    resolved_slug = repository_identity_slug(toplevel, remote=remote)
+    reasons: list[str] = []
+    expected = (expected_slug or "").strip() or None
+    if expected:
+        if resolved_slug and resolved_slug.lower() != expected.lower():
+            reasons.append(
+                f"canonical repository root identity mismatch: '{toplevel}' resolves "
+                f"to repository '{resolved_slug}' but the session is authorized for "
+                f"'{expected}' (forged or conflicting binding, fail closed)"
+            )
+        elif not resolved_slug and require_binding:
+            reasons.append(
+                f"canonical repository root '{toplevel}' has no resolvable git "
+                f"remote identity to confirm authorization for '{expected}' "
+                "(fail closed)"
+            )
+
+    return _assessment(
+        proven=not reasons,
+        reasons=reasons,
+        configured=True,
+        canonical_repo_root=toplevel,
+        resolved_slug=resolved_slug,
+        source=source,
+    )
+
+
+def format_canonical_repository_root_error(assessment: Mapping) -> str:
+    """Single RuntimeError message for MCP preflight gates."""
+    root = assessment.get("canonical_repo_root") or "(unknown)"
+    source = assessment.get("source") or "(unconfigured)"
+    reasons = "; ".join(
+        assessment.get("reasons") or ["unknown canonical repository root violation"]
+    )
+    return (
+        f"Canonical repository root guard (#706): {reasons}. "
+        f"binding source: {source}; canonical repository root: {root}. "
+        "Configure a valid canonical_repository_root for the target repository "
+        "and relaunch; do not point it at the Gitea-Tools install checkout."
+    )
+
+
+def _assessment(
+    *,
+    proven: bool,
+    reasons: list[str],
+    configured: bool,
+    canonical_repo_root: str,
+    resolved_slug: str | None,
+    source: str | None,
+) -> dict:
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": list(reasons),
+        "configured": configured,
+        "canonical_repo_root": canonical_repo_root,
+        "resolved_slug": resolved_slug,
+        "source": source,
+    }

docs/canonical-repository-root.md

@@ -0,0 +1,153 @@
+# Installation root vs canonical target repository root
+
+## Purpose
+
+This document (tracked as issue #741, building on #706 and #739/#740) explains
+the two distinct filesystem roots the Gitea-Tools MCP server reasons about, why
+conflating them silently targets the wrong repository, and which rule applies
+when you add a new consumer.
+
+It is the repository-scope companion to
+[`gitea-execution-profiles.md`](gitea-execution-profiles.md) (the profile model)
+and [`gitea-dual-namespace-deployment.md`](gitea-dual-namespace-deployment.md)
+(the per-role namespace model).
+
+## The two roots
+
+| | Installation root | Canonical target repository root |
+|---|---|---|
+| What it is | The checkout the server *code* lives in | The working root of the repository whose issues/PRs/branches the namespace *mutates* |
+| How it is derived | `PROJECT_ROOT = os.path.dirname(os.path.abspath(__file__))` | Configured per namespace, then pinned immutably into the session |
+| Configured by | Nothing — it follows the script | `canonical_repository_root` profile field, or the `GITEA_CANONICAL_REPOSITORY_ROOT` environment variable |
+| Changes at runtime? | No | No — first bind wins for the life of the process |
+| Accessor | `PROJECT_ROOT` | `_canonical_local_git_root()` (filesystem) / `_canonical_repository_slug()` (identity) |
+
+For a **single-repository** namespace — every Gitea-Tools namespace today — the
+two roots are the same path, and nothing about the existing behaviour changes.
+The distinction only becomes observable once a namespace is pointed at a
+different repository.
+
+## Which root does my code need?
+
+Ask what the operation is *about*, not where the file happens to sit.
+
+**Use the installation root (`PROJECT_ROOT`)** when the operation concerns the
+Gitea-Tools software itself:
+
+- server implementation / version parity (`master_parity_gate`, the
+  `startup_head` vs `current_head` staleness gate);
+- loading the server's own workflow, schema and skill files;
+- self-code hashing and stale-runtime detection;
+- locating installed scripts such as `mirror_refs.sh`.
+
+These are intentionally install-scoped. Do not "fix" them.
+
+**Use the canonical target root (`_canonical_local_git_root()`)** when the
+operation concerns the repository being worked on:
+
+- `git remote get-url` for repository identity;
+- branch creation, push, and commit;
+- ancestry and merge-base proofs;
+- worktree inventory, cleanup, and branch deletion;
+- any local git subprocess whose result feeds a mutation guard.
+
+**If you cannot tell, fail closed.** An ambiguous consumer that guesses the
+install root is the exact defect class #741 exists to eliminate.
+
+## Why conflating them inverts the guards
+
+Before #741, `_local_git_remote_url()` ran `git remote get-url` with
+`cwd=PROJECT_ROOT` unconditionally. Every consumer of repository *identity* —
+`_resolve`, the #530 remote/repo guard, the anti-stomp org/repo fill,
+`_workspace_repository_slug` — therefore read the Gitea-Tools remote and called
+it "the workspace", no matter which repository the namespace was bound to.
+
+For a namespace whose canonical root points elsewhere, this **inverts** the
+guard rather than merely weakening it:
+
+- an operation naming the genuinely bound target repository is **rejected**,
+  because that slug does not appear in the Gitea-Tools remote URL;
+- an operation naming Gitea-Tools is **accepted**.
+
+The filesystem guards (#274 branches-only and worktree membership) had already
+been migrated to the canonical root by #706, so the two halves of a single
+assessment described two different repositories.
+
+A related subtlety: repository identity must not be derived by looking a remote
+up by *name*. A target checkout commonly names its remote `origin` rather than
+`prgs`, so a name-keyed lookup returns nothing and the omitted coordinates fall
+through to the remote-wide default *target* — an unrelated repository.
+`_canonical_repository_slug()` probes candidate remote names against the
+canonical root instead.
+
+`_canonical_local_git_root()` is now the one place a target root is resolved.
+Do not re-derive it; new code that needs a target root calls that helper.
+
+## Configuration
+
+Declare the binding on the profile, alongside `allowed_repositories`:
+
+```json
+{
+  "profiles": {
+    "example-author": {
+      "role": "author",
+      "canonical_repository_root": "/absolute/path/to/target-repo",
+      "allowed_repositories": ["Example-Org/target-repo"]
+    }
+  }
+}
+```
+
+The namespace-scoped environment variable
+`GITEA_CANONICAL_REPOSITORY_ROOT` overrides the profile field, and is normally
+exported next to the server `cwd` in the MCP client configuration.
+
+Validation is layered, and each layer fails closed:
+
+1. **Config load.** The path must be a non-empty absolute string. All supported
+   loaders — v1, v2-`environments`, and v2-`contexts` — validate it identically.
+   (Before #741 only the v2-`contexts` loader validated it, and
+   v2-`environments` silently *dropped* the field during flattening, so the
+   namespace fell back to the install root — a fail-open.)
+2. **Bind time.** The path must exist, be a git repository, and resolve to a
+   repository identity matching the session's authorized slug. A configured but
+   unresolvable root is never replaced by the install identity.
+3. **Every mutation.** The pinned root is compared against the live configured
+   value; a mismatch is treated as a forged or conflicting binding.
+
+`allowed_repositories` remains a separate authorization boundary (#714): the
+canonical root determines *which* repository is derived, and
+`allowed_repositories` determines whether the session may act on it. A root that
+resolves to a repository outside that list fails closed.
+
+## Explicit coordinates confirm, never override
+
+Explicit `org`/`repo` arguments may **confirm** an existing canonical binding.
+They can never establish, complete, or replace one. A request naming a
+repository that contradicts the binding fails closed, in both directions:
+
+- a Gitea-Tools-rooted namespace cannot mutate another repository;
+- a namespace rooted at another repository cannot mutate Gitea-Tools.
+
+This matters because both-explicit coordinates short-circuit the #530
+remote/repo match check, so without this rule a caller could name any repository
+and skip validation entirely.
+
+No request-supplied workspace, remote, owner, repository, or worktree can
+replace the immutable root.
+
+## Parity is reported per dimension
+
+`gitea_assess_master_parity` reports two separately labelled dimensions:
+
+- `server_implementation` — the Gitea-Tools installation checkout. Its
+  `startup_head` / `current_head` / `stale` / `restart_required` fields keep
+  their original meaning, and **only this dimension gates mutations**: the
+  running process executes the code it started with, so a merged fix is not live
+  until the daemon restarts.
+- `target_repository` — the configured canonical target checkout and its
+  last-known remote master.
+
+"In parity" is a statement about one dimension, never about the whole system.
+Read the dimension you actually care about.

gitea_auth.py

@@ -802,6 +802,10 @@ def get_profile():
         # environment variable must never widen or forge the set of
         # repositories a session may bind to.
         "allowed_repositories": _json_list("allowed_repositories"),
+        # #706 cross-repository canonical root binding. Config-sourced here (the
+        # namespace-scoped GITEA_CANONICAL_REPOSITORY_ROOT env override is applied
+        # by canonical_repository_root.configured_canonical_root, not widened here).
+        "canonical_repository_root": jp.get("canonical_repository_root") or None,
         "audit_label": audit_label,
         "token_source_name": token_source,
         "auth_source_type": auth_type,

gitea_config.py

@@ -272,6 +272,15 @@ def load_config(path=None):
         )
     if not isinstance(data.get("profiles"), dict):
         raise ConfigError(f"{path} must be a JSON object with a 'profiles' object")
+    # #741: the v1 path returns `data` unflattened, so nothing else validates
+    # the cross-repository binding before gitea_auth.get_profile() reads it.
+    # Validate it here so every supported loader treats the field identically
+    # (a relative or blank path must never reach the runtime guard).
+    for _name, _profile in data["profiles"].items():
+        if isinstance(_profile, dict):
+            _validate_canonical_repository_root(
+                _name, _profile.get("canonical_repository_root")
+            )
     return data
 
 
@@ -358,6 +367,14 @@ def _flatten_identity(env_name, svc_name, svc, ident_name, ident):
     for key in ("role", "username", "execution_profile", "audit_label"):
         if ident.get(key):
             profile[key] = ident[key]
+    # #741: the cross-repository binding must survive flattening. Previously
+    # this key was silently dropped here, so a v2-environments namespace that
+    # declared canonical_repository_root fell back to the *installation* root
+    # and mutated Gitea-Tools instead of its target repository — a fail-open.
+    # Validate it exactly as the v2-contexts loader does before propagating.
+    _validate_canonical_repository_root(addr, ident.get("canonical_repository_root"))
+    if ident.get("canonical_repository_root"):
+        profile["canonical_repository_root"] = ident["canonical_repository_root"]
     return addr, profile
 
 
@@ -502,6 +519,30 @@ def _validate_allowed_repositories(name, raw):
             )
 
 
+def _validate_canonical_repository_root(name, raw):
+    """Validate the optional per-profile canonical repository root (#706).
+
+    ``canonical_repository_root`` binds a cross-repository namespace to the
+    working root of its target repository (separate from the immutable
+    Gitea-Tools install checkout). It is an absolute filesystem path; existence
+    and git identity are validated at bind time by the runtime guard, not here
+    (config validation stays filesystem-independent). Absent means the
+    single-repo default and is allowed.
+    """
+    if raw is None:
+        return
+    if not isinstance(raw, str) or not raw.strip():
+        raise ConfigError(
+            f"profile '{name}' canonical_repository_root must be a non-empty "
+            "absolute path string to the target repository working root"
+        )
+    if not os.path.isabs(raw.strip()):
+        raise ConfigError(
+            f"profile '{name}' canonical_repository_root {raw!r} must be an "
+            "absolute path"
+        )
+
+
 def _reject_inline_secrets(kind, name, obj):
     for key in _INLINE_SECRET_KEYS:
         if key in obj:
@@ -577,6 +618,9 @@ def _load_v2_contexts(data, path):
         if not isinstance(allowed, list) or not isinstance(forbidden, list):
             raise ConfigError(f"profile '{name}' operation fields must be lists")
         _validate_allowed_repositories(name, raw.get("allowed_repositories"))
+        _validate_canonical_repository_root(
+            name, raw.get("canonical_repository_root")
+        )
         allowed_n = {_normalize_op("gitea", op, name) for op in allowed}
         forbidden_n = {_normalize_op("gitea", op, name) for op in forbidden}
         # Reviewer-identity deadlock rule (#100/#103) applies here unchanged.

gitea_mcp_server.py

@@ -194,6 +194,7 @@ MERGER_WORKTREE_ENV = "GITEA_MERGER_WORKTREE"
 RECONCILER_WORKTREE_ENV = "GITEA_RECONCILER_WORKTREE"
 
 import namespace_workspace_binding as nwb  # noqa: E402
+import canonical_repository_root as crr  # noqa: E402  # #706 cross-repo canonical root
 import mcp_namespace_health  # noqa: E402
 import stale_binding_recovery  # noqa: E402
 
@@ -376,6 +377,51 @@ def _assess_stale_active_binding(auto_recover: bool = False) -> dict:
     return report
 
 
+def _configured_canonical_root() -> tuple[str | None, str | None]:
+    """Declared cross-repository canonical root for the active namespace (#706).
+
+    Returns ``(value, source)`` from the profile/env binding, or ``(None, None)``
+    for the single-repo default. The raw declared value is threaded into
+    workspace resolution so the branches-only / membership guards evaluate
+    against the configured target repository; the strict identity/existence
+    checks run separately in :func:`_enforce_canonical_repository_root`.
+    """
+    try:
+        profile = get_profile()
+    except Exception:
+        profile = {}
+    return crr.configured_canonical_root(profile, os.environ)
+
+
+def _canonical_local_git_root() -> str:
+    """Local git working root for *target-repository* operations (#741).
+
+    Two distinct roots exist and must never be conflated:
+
+    * ``PROJECT_ROOT`` — the Gitea-Tools **installation** checkout, i.e. where
+      this server script lives. Server implementation/version parity, workflow
+      and skill file loading, and self-code staleness detection are anchored
+      here and must stay that way.
+    * the **canonical target repository root** — the working root of the
+      repository whose issues/PRs/branches the namespace actually mutates.
+
+    Every local git invocation that reads or proves *target repository* state
+    (remote identity, ancestry, worktree inventory, cleanup) must run here, or
+    a cross-repository namespace silently operates on Gitea-Tools instead.
+
+    Returns the resolved canonical target root when one is configured, else
+    ``PROJECT_ROOT`` — which preserves the single-repository default exactly.
+    A configured-but-invalid root is *not* silently replaced by the install
+    checkout: it is returned as-is so downstream git calls fail rather than
+    succeed against the wrong repository, and
+    :func:`_enforce_canonical_repository_root` fails the mutation closed first.
+    """
+    configured, _source = _configured_canonical_root()
+    if not configured:
+        return PROJECT_ROOT
+    return crr.resolve_repo_toplevel(configured) or os.path.realpath(configured)
+
+
 def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
     """Resolve the namespace-scoped workspace root inspected by pre-flight guards.
 
@@ -399,8 +445,9 @@ def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
 
 
 def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dict:
-    """Canonical namespace workspace + repository root for guards (#460/#510)."""
+    """Canonical namespace workspace + repository root for guards (#460/#510/#706)."""
     role = _effective_workspace_role()
+    configured_root, _source = _configured_canonical_root()
     return nwb.resolve_namespace_mutation_context(
         role_kind=role,
         worktree_path=worktree_path,
@@ -409,6 +456,7 @@ def _resolve_namespace_mutation_context(worktree_path: str | None = None) -> dic
             _reviewer_session_worktree() if role in {"reviewer", "merger"} else None
         ),
         profile_name=get_profile().get("profile_name"),
+        configured_canonical_root=configured_root,
     )
 
 
@@ -721,6 +769,54 @@ def record_preflight_check(
             _preflight_resolved_task = resolved_task
 
 
+def _enforce_canonical_repository_root(
+    worktree_path: str | None = None,
+    *,
+    remote: str | None = None,
+) -> None:
+    """#706: validate the immutable cross-repository canonical root binding.
+
+    A no-op for the single-repo default (no configured binding). When a
+    namespace declares a ``canonical_repository_root`` (profile/env), the
+    configured target repository must exist, be a git repository, and carry a
+    repository identity matching the session's authorized repository. Missing,
+    conflicting, or forged bindings fail closed, and the validated root is
+    checked against the immutable session pin so it cannot be swapped mid
+    session.
+    """
+    configured_value, source = _configured_canonical_root()
+    if not configured_value:
+        return
+
+    bound = session_ctx.get_session_context() or {}
+    expected_slug = session_ctx.format_repository_slug(
+        bound.get("org"), bound.get("repository")
+    )
+    assessment = crr.assess_canonical_repository_root(
+        configured_value=configured_value,
+        source=source,
+        expected_slug=expected_slug,
+        process_project_root=PROJECT_ROOT,
+        remote=remote,
+        require_binding=True,
+    )
+    if assessment["block"]:
+        raise RuntimeError(
+            crr.format_canonical_repository_root_error(assessment)
+        )
+
+    drift = session_ctx.assess_session_context(
+        profile_name=get_profile().get("profile_name"),
+        remote=remote,
+        canonical_repository_root=assessment["canonical_repo_root"],
+    )
+    if drift["block"]:
+        raise RuntimeError(
+            "Canonical repository root guard (#706): "
+            + "; ".join(drift["reasons"])
+        )
+
+
 def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
     """#274: author file/branch mutations must run from a branches/ worktree.
 
@@ -1115,6 +1211,7 @@ def verify_preflight_purity(
                     )
 
         # Historical path: root + branches after purity-order when dirty paths live.
+        _enforce_canonical_repository_root(worktree_path, remote=remote)
         _enforce_root_checkout_guard(worktree_path)
         _enforce_branches_only_author_mutation(worktree_path)
         _enforce_issue_scope_guard(
@@ -1148,6 +1245,7 @@ def verify_preflight_purity(
     # #683: under pytest unit isolation, FORCE_PRODUCTION_GUARDS still runs
     # production root + branches + issue scope (no silent no-op of guards).
     if production_active:
+        _enforce_canonical_repository_root(worktree_path, remote=remote)
         _enforce_root_checkout_guard(worktree_path)
         _enforce_branches_only_author_mutation(worktree_path)
         _enforce_issue_scope_guard(
@@ -1379,6 +1477,12 @@ def _verify_role_mutation_workspace(
     git_state = issue_lock_worktree.read_worktree_git_state(
         _resolve_preflight_workspace_path(worktree_path)
     )
+    # #741: thread the configured canonical root exactly as
+    # _resolve_namespace_mutation_context does. Omitting it here made the two
+    # paths disagree: the #274 branches-only / worktree-membership guards fell
+    # 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()
     assessment = nwb.assess_namespace_mutation_workspace(
         role_kind=role,
         worktree_path=worktree_path,
@@ -1389,6 +1493,7 @@ def _verify_role_mutation_workspace(
         ),
         profile_name=get_profile().get("profile_name"),
         current_branch=git_state.get("current_branch"),
+        configured_canonical_root=_configured_root,
     )
     if assessment["block"]:
         raise RuntimeError(
@@ -1579,6 +1684,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,
@@ -1606,7 +1750,21 @@ def _trusted_session_repository(
             "repository": None,
             "reasons": [str(exc)],
         }
-    slug = _workspace_repository_slug(remote)
+    # #706 F1: when a cross-repository canonical root is configured, the session
+    # repository identity must be derived from that validated *target* repository,
+    # not from the install-checkout git remote. ``_workspace_repository_slug``
+    # reads ``_local_git_remote_url`` in ``PROJECT_ROOT`` (always Gitea-Tools), so
+    # without this a cross-repo namespace pinned Gitea-Tools while #274 filesystem
+    # membership bound the external root, and ``_enforce_canonical_repository_root``
+    # then failed closed on a self-inflicted identity mismatch. The derived
+    # identity is still authorized by the profile allowlist below (never
+    # 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.
+    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,
@@ -1644,6 +1802,18 @@ def _seed_session_context(
     """
     expected = (profile.get("username") or "").strip() or None
     trusted = _trusted_session_repository(profile, remote)
+    # #706: pin the configured cross-repository canonical root immutably so a
+    # later call cannot forge/swap it. Store the resolved toplevel so drift
+    # checks compare against the same value the mutation gate validates.
+    configured_value, _crr_source = crr.configured_canonical_root(
+        profile, os.environ
+    )
+    canonical_root_pin = None
+    if configured_value:
+        canonical_root_pin = (
+            crr.resolve_repo_toplevel(configured_value)
+            or os.path.realpath(configured_value)
+        )
     return session_ctx.seed_session_context_if_unbound(
         profile_name=profile.get("profile_name") or "",
         remote=remote,
@@ -1654,6 +1824,7 @@ def _seed_session_context(
         role_kind=_profile_role_kind(profile),
         expected_username=expected,
         source=source,
+        canonical_repository_root=canonical_root_pin,
     )
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
@@ -2268,6 +2439,21 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
     filled_org = False
     filled_repo = False
     if not org_explicit or not repo_explicit:
+        parsed = None
+        # #741: for a cross-repository namespace, prefer the canonical root's
+        # own repository identity. `_local_git_remote_url` looks a remote up by
+        # *name*, but a target checkout commonly names its remote `origin`
+        # rather than `prgs`; that lookup then returns None and the omitted
+        # coordinates silently fell through to the REMOTES default target —
+        # i.e. a completely unrelated repository. `_canonical_repository_slug`
+        # probes the candidate remote names against the canonical root.
+        try:
+            canonical_slug, _reasons = _canonical_repository_slug(get_profile(), remote)
+        except Exception:
+            canonical_slug = None
+        if canonical_slug:
+            parsed = session_ctx.parse_repository_slug(canonical_slug)
+        if not parsed:
             parsed = remote_repo_guard.parse_org_repo_from_remote_url(
                 _local_git_remote_url(remote)
             )
@@ -2278,6 +2464,40 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
             if not repo_explicit:
                 resolved_repo = parsed[1]
                 filled_repo = True
+    # #741: explicit caller coordinates may *confirm* a cross-repository
+    # canonical binding but must never override it. Both-explicit coordinates
+    # short-circuit the #530 guard (remote_repo_guard.assess_remote_repo_match),
+    # so without this check a caller could name any repository and skip
+    # validation entirely. Unconfigured (single-repository) namespaces yield no
+    # canonical slug and are unaffected. An unresolvable configured root is not
+    # blocked here — reads must keep working; the fail-closed decision belongs
+    # to _enforce_canonical_repository_root at the mutation gate.
+    if org_explicit or repo_explicit:
+        try:
+            canonical_slug, _canonical_reasons = _canonical_repository_slug(
+                get_profile(), remote
+            )
+        except Exception:
+            canonical_slug = None
+        bound = (
+            session_ctx.parse_repository_slug(canonical_slug)
+            if canonical_slug
+            else None
+        )
+        if bound:
+            override = session_ctx.assess_repository_override(
+                requested_org=org,
+                requested_repo=repo,
+                bound_org=bound[0],
+                bound_repo=bound[1],
+            )
+            if override.get("block"):
+                raise RuntimeError(
+                    "Canonical repository binding guard (#741): "
+                    + "; ".join(override.get("reasons") or [])
+                    + ". Explicit org/repo may confirm the configured canonical "
+                    "repository binding but can never override it."
+                )
     _enforce_remote_repo_guard(
         remote,
         resolved_org,
@@ -2932,7 +3152,7 @@ def gitea_lock_issue(
         return blocked
 
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
-        worktree_path, PROJECT_ROOT
+        worktree_path, _canonical_local_git_root()
     )
     h, o, r = _resolve(remote, host, org, repo)
     active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
@@ -3234,7 +3454,7 @@ def gitea_create_pr(
     locked_worktree = lock_data.get("worktree_path")
 
     worktree_check = issue_lock_worktree.verify_pr_worktree_matches_lock(
-        locked_worktree, worktree_path, PROJECT_ROOT
+        locked_worktree, worktree_path, _canonical_local_git_root()
     )
     if worktree_check["block"]:
         raise ValueError(worktree_check["reasons"][0])
@@ -8364,13 +8584,24 @@ def gitea_merge_pr(
 
 
 def _local_git_remote_url(remote_name: str) -> str | None:
-    """Best-effort local ``git remote get-url`` for trust-gate corroboration."""
+    """Best-effort local ``git remote get-url`` for trust-gate corroboration.
+
+    #741: this runs in the **canonical target repository root**, not the
+    Gitea-Tools installation checkout. Every consumer of this function derives
+    *target repository identity* from it (``_resolve``, the #530 remote/repo
+    guard, the anti-stomp org/repo fill, ``_workspace_repository_slug``), so
+    running it in ``PROJECT_ROOT`` inverted those guards for a cross-repository
+    namespace: the genuinely bound target was rejected while Gitea-Tools was
+    accepted. Single-repository namespaces are unaffected —
+    :func:`_canonical_local_git_root` returns ``PROJECT_ROOT`` when no
+    cross-repository binding is configured.
+    """
     try:
         proc = subprocess.run(
             ["git", "remote", "get-url", remote_name],
             capture_output=True,
             text=True,
-            cwd=PROJECT_ROOT,
+            cwd=_canonical_local_git_root(),
         )
         if proc.returncode != 0:
             return None
@@ -8650,8 +8881,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
     )
@@ -8920,7 +9171,7 @@ def gitea_cleanup_merged_pr_branch(
         else f"origin/{target_branch}"
     )
     head_on_target = merged_cleanup_reconcile.is_head_ancestor_of_ref(
-        PROJECT_ROOT,
+        _canonical_local_git_root(),
         head_sha,
         target_ref,
     )
@@ -8955,7 +9206,7 @@ def gitea_cleanup_merged_pr_branch(
         repo=r,
         branch=head_branch,
         pr_number=pr_number,
-        project_root=PROJECT_ROOT,
+        project_root=_canonical_local_git_root(),
         auth=auth,
         base_api=base,
     )
@@ -9450,7 +9701,7 @@ def gitea_capture_branches_worktree_snapshot(
             "reasons": read_block,
             "permission_report": _permission_block_report("gitea.read"),
         }
-    root = PROJECT_ROOT
+    root = _canonical_local_git_root()
     if worktree_path:
         root = os.path.realpath(os.path.abspath(worktree_path))
         git_root = _get_git_root(root)
@@ -9560,14 +9811,14 @@ def gitea_reconcile_merged_cleanups(
                 h, o, r, auth, head_ref
             )
         head_on_master[int(pr["number"])] = merged_cleanup_reconcile.is_head_ancestor_of_ref(
-            PROJECT_ROOT, head_sha, master_ref
+            _canonical_local_git_root(), head_sha, master_ref
         )
 
     delete_capability_allowed = not _profile_operation_gate("gitea.branch.delete")
 
     # #534: discover reviewer scratch trees and active leases for those PRs.
     scratch_candidates = merged_cleanup_reconcile.discover_reviewer_scratch_worktrees(
-        PROJECT_ROOT
+        _canonical_local_git_root()
     )
     active_reviewer_leases: dict[int, bool] = {}
     pr_states: dict[int, dict] = {}
@@ -9593,7 +9844,7 @@ def gitea_reconcile_merged_cleanups(
         }
 
     report = merged_cleanup_reconcile.build_reconciliation_report(
-        project_root=PROJECT_ROOT,
+        project_root=_canonical_local_git_root(),
         closed_prs=merged_closed,
         open_prs=open_prs,
         remote_branch_exists=remote_branch_exists,
@@ -9633,7 +9884,7 @@ def gitea_reconcile_merged_cleanups(
                 repo=r,
                 branch=head_branch,
                 pr_number=pr_num_int,
-                project_root=PROJECT_ROOT,
+                project_root=_canonical_local_git_root(),
                 auth=auth,
                 base_api=base,
             )
@@ -9717,7 +9968,7 @@ def gitea_reconcile_merged_cleanups(
 
         if local_assessment.get("safe_to_remove_worktree"):
             result = merged_cleanup_reconcile.remove_local_worktree(
-                PROJECT_ROOT,
+                _canonical_local_git_root(),
                 head_branch,
                 worktree_path=local_assessment.get("worktree_path"),
             )
@@ -9727,7 +9978,7 @@ def gitea_reconcile_merged_cleanups(
         if not scratch.get("safe_to_remove_worktree"):
             continue
         result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
-            PROJECT_ROOT, scratch.get("worktree_path") or ""
+            _canonical_local_git_root(), scratch.get("worktree_path") or ""
         )
         actions.append({"action": "remove_reviewer_scratch_worktree", **result})
 
@@ -9825,7 +10076,7 @@ def gitea_assess_already_landed_reconciliation(
 
     assessment = reconciliation_workflow.assess_open_pr_reconciliation(
         pr=pr,
-        project_root=PROJECT_ROOT,
+        project_root=_canonical_local_git_root(),
         remote=remote,
         target_branch=target_branch,
     )
@@ -9921,14 +10172,14 @@ def gitea_scan_already_landed_open_prs(
     }
 
     target_fetch = reconciliation_workflow.fetch_target_branch(
-        PROJECT_ROOT, remote, target_branch
+        _canonical_local_git_root(), remote, target_branch
     )
 
     candidates: list[dict] = []
     for pr in open_prs:
         assessment = reconciliation_workflow.assess_open_pr_reconciliation(
             pr=pr,
-            project_root=PROJECT_ROOT,
+            project_root=_canonical_local_git_root(),
             remote=remote,
             target_branch=target_branch,
             target_fetch=target_fetch,
@@ -10025,7 +10276,7 @@ def gitea_audit_worktree_cleanup(
         active_issue_branches.add(str(lock["branch_name"]).strip())
 
     report = worktree_cleanup_audit.audit_branches_directory(
-        PROJECT_ROOT,
+        _canonical_local_git_root(),
         open_pr_branches=open_pr_branches,
         active_issue_branches=active_issue_branches,
         now=datetime.now(timezone.utc),
@@ -10108,7 +10359,7 @@ def gitea_reconcile_already_landed_pr(
 
     assessment = already_landed_reconcile.assess_already_landed_reconciliation(
         pr=pr,
-        project_root=PROJECT_ROOT,
+        project_root=_canonical_local_git_root(),
         remote=remote,
         target_branch=target_branch,
     )
@@ -10313,7 +10564,7 @@ def gitea_reconcile_superseded_by_merged_pr(
     )
 
     target_fetch = already_landed_reconcile.fetch_target_branch(
-        PROJECT_ROOT, remote, target_branch
+        _canonical_local_git_root(), remote, target_branch
     )
     superseding_head_sha = (
         (superseding_pr.get("head") or {}).get("sha")
@@ -10323,7 +10574,7 @@ def gitea_reconcile_superseded_by_merged_pr(
     ancestor = None
     if target_fetch.get("success"):
         ancestor = already_landed_reconcile.is_head_ancestor_of_ref(
-            PROJECT_ROOT,
+            _canonical_local_git_root(),
             superseding_head_sha,
             target_fetch.get("target_ref") or f"{remote}/{target_branch}",
         )
@@ -11467,6 +11718,11 @@ def gitea_acquire_merger_pr_lease(
     }, lease_provenance=merger_lease_adoption.build_lease_provenance(
         source=merger_lease_adoption.SOURCE_ACQUIRE_MERGER,
         comment_id=posted.get("id"),
+        # #742: bind the lease to this native runtime so a later owner-session
+        # finalization can prove it is the same daemon that acquired it.
+        native_token_fingerprint=(
+            mcp_daemon_guard.native_runtime_status().get("token_fingerprint")
+        ),
     ))
     return {
         "success": True,
@@ -11656,6 +11912,191 @@ def gitea_adopt_merger_pr_lease(
     }
 
 
+@mcp.tool()
+def gitea_release_merger_pr_lease(
+    pr_number: int,
+    worktree: str,
+    candidate_head: str,
+    outcome: str = merger_lease_adoption.OUTCOME_RELEASED,
+    reason: str | None = None,
+    issue_number: int | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Terminally release or abandon this merger session's own PR lease (#742).
+
+    Sanctioned owner-session finalization for a comment-backed merger lease when
+    the merge does **not** occur (non-retryable merge failure, aborted merge, or
+    a blocker that ends the attempt). Without it a failed merger lease can only
+    expire, blocking the queue for the remainder of its TTL.
+
+    Merger-only and owner-only. The caller must hold the exact in-session lease
+    it is finalizing; foreign-session release, reviewer profiles, and mismatched
+    repository / PR / candidate head / identity / profile / native runtime token
+    fingerprint all fail closed. Reviewer sessions use
+    ``gitea_release_reviewer_pr_lease`` instead — that tool is never a merger
+    path.
+
+    Finalization is append-only: it posts a terminal lease marker and never
+    edits or deletes prior ledger comments. It is idempotent — an already
+    terminal lease returns ``already_terminal`` and posts nothing. The PR, its
+    approval, the review decision lock, the branch, and the worktree are all
+    preserved; only the lease is ended.
+
+    Args:
+        pr_number: PR whose merger lease to finalize.
+        outcome: 'released' (clean end) or 'abandoned' (ended under a blocker).
+        reason: Short finalization reason recorded in the marker.
+        candidate_head: Pinned head SHA the lease was acquired against.
+
+    Returns:
+        dict reporting finalization status, reasons, and the marker comment id.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    comment_block = _profile_operation_gate("gitea.pr.comment")
+    if comment_block:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": comment_block,
+            "permission_report": _permission_block_report("gitea.pr.comment"),
+        }
+    # Merger-role proof: reviewer profiles hold gitea.pr.comment but never
+    # gitea.pr.merge, so this gate keeps the tool merger-only (#742).
+    merge_block = _profile_operation_gate("gitea.pr.merge")
+    if merge_block:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": merge_block,
+            "permission_report": _permission_block_report("gitea.pr.merge"),
+        }
+
+    try:
+        _verify_role_mutation_workspace(
+            remote,
+            worktree=worktree,
+            task="release_merger_pr_lease",
+            org=org,
+            repo=repo,
+        )
+    except RuntimeError as e:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": [str(e)],
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    profile = get_profile()
+    profile_name = profile.get("profile_name") or "unknown"
+    identity = _authenticated_username(h) or profile.get("username") or ""
+    repo_label = f"{o}/{r}"
+
+    session = reviewer_pr_lease.get_session_lease()
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo)
+
+    pr_live = api_request(
+        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
+    live_head = str(
+        (pr_live.get("head") or {}).get("sha") or pr_live.get("head_sha") or ""
+    ).strip()
+
+    assessment = merger_lease_adoption.assess_merger_lease_finalization(
+        comments,
+        pr_number=pr_number,
+        session=session,
+        actor_identity=identity,
+        actor_profile=profile_name,
+        actor_session_id=(session or {}).get("session_id"),
+        repo=repo_label,
+        worktree=worktree,
+        candidate_head=candidate_head,
+        live_head_sha=live_head or None,
+        outcome=outcome,
+        reason=reason,
+        runtime_token_fingerprint=(
+            mcp_daemon_guard.native_runtime_status().get("token_fingerprint")
+        ),
+        issue_number=issue_number,
+    )
+
+    if assessment.get("already_terminal"):
+        # Idempotent: the lease is already terminal on the ledger. Clear the
+        # in-session mirror so no stale proof survives, and post nothing.
+        reviewer_pr_lease.clear_session_lease()
+        return {
+            "success": True,
+            "finalized": False,
+            "already_terminal": True,
+            "pr_number": pr_number,
+            "outcome": assessment.get("outcome"),
+            "reasons": [],
+        }
+
+    if not assessment.get("finalize_allowed"):
+        return {
+            "success": False,
+            "finalized": False,
+            "already_terminal": False,
+            "pr_number": pr_number,
+            "reasons": assessment.get("reasons") or [],
+        }
+
+    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    with _audited(
+        "comment_pr",
+        host=h,
+        remote=remote,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        request_metadata={"source": "release_merger_pr_lease"},
+    ):
+        posted = api_request(
+            "POST", comment_url, auth, {"body": assessment["finalization_body"]}
+        )
+
+    if not isinstance(posted, dict) or not posted.get("id"):
+        # The lease is still live on the ledger: keep the in-session mirror so
+        # the owner can retry finalization rather than losing its own proof.
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": [
+                "terminal lease marker POST failed or returned no comment id "
+                "(lease left intact; retry finalization)"
+            ],
+        }
+
+    reviewer_pr_lease.clear_session_lease()
+
+    return {
+        "success": True,
+        "finalized": True,
+        "already_terminal": False,
+        "pr_number": pr_number,
+        "outcome": assessment.get("outcome"),
+        "finalization_reason": assessment.get("finalization_reason"),
+        "finalization_comment_id": posted.get("id"),
+        "finalized_lease_comment_id": assessment.get("lease_comment_id"),
+        "candidate_head": assessment.get("candidate_head"),
+        "repo": repo_label,
+        "reasons": [],
+    }
+
+
 @mcp.tool()
 def gitea_heartbeat_reviewer_pr_lease(
     pr_number: int,
@@ -11691,6 +12132,13 @@ def gitea_heartbeat_reviewer_pr_lease(
     verify_preflight_purity(remote, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
+    # Slide the sliding TTL forward (#747): the heartbeat is the liveness proof,
+    # so renewal is stated explicitly rather than inherited from the acquisition
+    # default.
+    beat_at = datetime.now(timezone.utc)
+    renewed_expiry = beat_at + timedelta(
+        minutes=reviewer_pr_lease.LEASE_RENEWAL_MINUTES
+    )
     body = reviewer_pr_lease.format_lease_body(
         repo=f"{o}/{r}",
         pr_number=pr_number,
@@ -11703,6 +12151,8 @@ def gitea_heartbeat_reviewer_pr_lease(
         candidate_head=candidate_head or session.get("candidate_head"),
         target_branch=session.get("target_branch") or "master",
         target_branch_sha=target_branch_sha or session.get("target_branch_sha"),
+        last_activity=beat_at,
+        ttl_minutes=reviewer_pr_lease.LEASE_RENEWAL_MINUTES,
     )
     comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
     with _audited(
@@ -11743,6 +12193,13 @@ def gitea_heartbeat_reviewer_pr_lease(
         "phase": phase,
         "comment_id": posted.get("id"),
         "session_lease": updated,
+        # Report the renewed window so an operator can tell "held and live"
+        # from "held and dying" (#747).
+        "ttl_minutes": reviewer_pr_lease.LEASE_RENEWAL_MINUTES,
+        "expires_at": renewed_expiry.replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z"),
+        "seconds_remaining": reviewer_pr_lease.LEASE_RENEWAL_MINUTES * 60,
         "reasons": [],
     }
 
@@ -13918,6 +14375,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()
@@ -14095,13 +14562,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
+        'current_head', 'mutation_gate_enforced', 'summary', 'reasons',
-        'report' recovery payload when stale.
+        '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"],
@@ -14113,6 +14590,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)
@@ -15359,7 +15847,7 @@ def gitea_update_pr_branch_by_merge(
 
     control_clean = True
     try:
-        porcelain = _get_workspace_porcelain(PROJECT_ROOT)
+        porcelain = _get_workspace_porcelain(_canonical_local_git_root())
         # Untracked-only dirt is ignored; tracked edits contaminate control.
         tracked_dirty = [
             line for line in (porcelain or "").splitlines()
@@ -16364,6 +16852,8 @@ def gitea_resolve_task_capability(
         "gitea_acquire_merger_pr_lease",
         "adopt_merger_pr_lease",
         "gitea_adopt_merger_pr_lease",
+        "release_merger_pr_lease",
+        "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
         "create_pr",

master_parity_gate.py

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

merger_lease_adoption.py

@@ -20,6 +20,7 @@ SOURCE_ADOPT = "gitea_adopt_merger_pr_lease"
 SOURCE_ACQUIRE = "gitea_acquire_reviewer_pr_lease"
 SOURCE_ACQUIRE_MERGER = "gitea_acquire_merger_pr_lease"
 SOURCE_HEARTBEAT = "gitea_heartbeat_reviewer_pr_lease"
+SOURCE_RELEASE_MERGER = "gitea_release_merger_pr_lease"
 
 SANCTIONED_PROVENANCE_SOURCES = frozenset({
     SOURCE_ADOPT,
@@ -30,6 +31,21 @@ SANCTIONED_PROVENANCE_SOURCES = frozenset({
 
 _MERGER_ADOPTABLE_FRESHNESS = frozenset({"active", "stale_warning"})
 
+# #742: owner-session terminal finalization of a merger-held lease. Append-only
+# marker; ledger history is never edited or deleted.
+MERGER_FINALIZATION_MARKER = "<!-- mcp-merger-lease-final:v1 -->"
+OUTCOME_RELEASED = "released"
+OUTCOME_ABANDONED = "abandoned"
+MERGER_FINALIZATION_OUTCOMES = frozenset({OUTCOME_RELEASED, OUTCOME_ABANDONED})
+DEFAULT_MERGER_FINALIZATION_REASON = "merge-not-performed"
+
+# Provenance sources whose in-session lease is merger-owned and therefore
+# finalizable by its owning merger session.
+MERGER_OWNED_PROVENANCE_SOURCES = frozenset({
+    SOURCE_ACQUIRE_MERGER,
+    SOURCE_ADOPT,
+})
+
 
 def format_adoption_body(
     *,
@@ -97,6 +113,7 @@ def build_lease_provenance(
     adopted_from_profile: str | None = None,
     adopted_from_reviewer_identity: str | None = None,
     adoption_reason: str | None = None,
+    native_token_fingerprint: str | None = None,
     recorded_at: datetime | None = None,
 ) -> dict[str, Any]:
     recorded_at = recorded_at or datetime.now(timezone.utc)
@@ -117,9 +134,76 @@ def build_lease_provenance(
         proof["adopted_from_reviewer_identity"] = adopted_from_reviewer_identity
     if adoption_reason:
         proof["adoption_reason"] = adoption_reason
+    if native_token_fingerprint:
+        proof["native_token_fingerprint"] = native_token_fingerprint
     return proof
 
 
+def assess_acquired_merger_lease_integrity(
+    session: dict[str, Any] | None,
+) -> list[str]:
+    """Ownership/integrity reasons blocking a ``SOURCE_ACQUIRE_MERGER`` lease (#742).
+
+    A merger lease minted by ``gitea_acquire_merger_pr_lease`` authorizes an
+    irreversible merge, so it is sanctioned only when the in-session record is
+    complete and self-consistent: comment marker, exact session identity, merger
+    profile/role, repository, PR number, and pinned candidate head. Any missing
+    or contradictory field fails closed — an incomplete record is not proof.
+    """
+    if not session:
+        return ["no in-session lease recorded"]
+    provenance = session.get("lease_provenance") or {}
+    if not isinstance(provenance, dict):
+        provenance = {}
+    reasons: list[str] = []
+
+    provenance_comment_id = provenance.get("comment_id")
+    session_comment_id = session.get("comment_id")
+    if not (provenance_comment_id or session_comment_id):
+        reasons.append(
+            "acquired merger lease has no comment marker id (comment-backed "
+            "proof required)"
+        )
+    elif (
+        provenance_comment_id is not None
+        and session_comment_id is not None
+        and provenance_comment_id != session_comment_id
+    ):
+        reasons.append(
+            "acquired merger lease provenance comment_id does not match the "
+            "session lease comment_id"
+        )
+
+    if not (session.get("session_id") or "").strip():
+        reasons.append("acquired merger lease has no session_id")
+    if not (session.get("reviewer_identity") or "").strip():
+        reasons.append("acquired merger lease has no holder identity")
+
+    profile = (session.get("profile") or "").strip()
+    if not profile:
+        reasons.append("acquired merger lease has no profile")
+    elif "merger" not in profile.lower():
+        reasons.append(
+            f"acquired merger lease profile '{profile}' is not a merger profile "
+            "(merger-only; fail closed)"
+        )
+
+    if not (session.get("repo") or "").strip():
+        reasons.append("acquired merger lease has no repository")
+
+    pr_number = session.get("pr_number")
+    if not isinstance(pr_number, int) or isinstance(pr_number, bool) or pr_number <= 0:
+        reasons.append("acquired merger lease has no valid PR number")
+
+    if not leases._normalize_sha(session.get("candidate_head")):
+        reasons.append(
+            "acquired merger lease has no pinned candidate_head (exact-head "
+            "scoping required)"
+        )
+
+    return reasons
+
+
 def is_sanctioned_session_lease(session: dict[str, Any] | None) -> bool:
     if not session:
         return False
@@ -131,6 +215,8 @@ def is_sanctioned_session_lease(session: dict[str, Any] | None) -> bool:
         return bool(provenance.get("comment_id")) and bool(
             provenance.get("adopted_from_session_id")
         )
+    if source == SOURCE_ACQUIRE_MERGER:
+        return not assess_acquired_merger_lease_integrity(session)
     if source in {SOURCE_ACQUIRE, SOURCE_HEARTBEAT}:
         return bool(session.get("comment_id") or provenance.get("comment_id"))
     return False
@@ -168,6 +254,8 @@ def describe_session_lease_proof(
     if source == SOURCE_ADOPT and sanctioned:
         kind = "sanctioned_adoption"
         reason = reason or DEFAULT_ADOPTION_REASON
+    elif source == SOURCE_ACQUIRE_MERGER and sanctioned:
+        kind = "sanctioned_acquire_merger"
     elif source == SOURCE_ACQUIRE and sanctioned:
         kind = "sanctioned_acquire"
     elif source == SOURCE_HEARTBEAT and sanctioned:
@@ -216,6 +304,7 @@ _SANCTIONED_LEASE_EVIDENCE_RE = re.compile(
     r"lease_proof_source\s*[:=]\s*gitea_acquire_reviewer_pr_lease|"
     r"lease_proof_source\s*[:=]\s*gitea_acquire_merger_pr_lease|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_adoption|"
+    r"lease_proof_kind\s*[:=]\s*sanctioned_acquire_merger|"
     r"lease_proof_kind\s*[:=]\s*sanctioned_acquire|"
     r"adoption_comment_id\s*[:=]|"
     r"sanctioned_adoption|"
@@ -247,6 +336,267 @@ def assess_manual_lease_proof_handoff(report_text: str) -> dict[str, Any]:
     }
 
 
+def format_merger_finalization_body(
+    *,
+    repo: str,
+    pr_number: int,
+    issue_number: int | None,
+    merger_identity: str,
+    merger_profile: str,
+    merger_session_id: str,
+    worktree: str,
+    candidate_head: str | None,
+    target_branch: str,
+    target_branch_sha: str | None,
+    outcome: str,
+    reason: str,
+    lease_comment_id: int | None,
+    finalized_at: datetime | None = None,
+) -> str:
+    """Render the append-only terminal marker for a merger-owned lease (#742).
+
+    The body carries a standard lease marker in a terminal phase, so the
+    existing newest-wins ledger (#577) ends the lease without editing or
+    deleting any prior comment.
+    """
+    finalized_at = finalized_at or datetime.now(timezone.utc)
+    finalized_text = finalized_at.astimezone(timezone.utc).replace(
+        microsecond=0
+    ).isoformat().replace("+00:00", "Z")
+    lease_body = leases.format_lease_body(
+        repo=repo,
+        pr_number=pr_number,
+        issue_number=issue_number,
+        reviewer_identity=merger_identity,
+        profile=merger_profile,
+        session_id=merger_session_id,
+        worktree=worktree,
+        phase=outcome,
+        candidate_head=candidate_head,
+        target_branch=target_branch,
+        target_branch_sha=target_branch_sha,
+        last_activity=finalized_at,
+        blocker=reason,
+    )
+    lines = [
+        MERGER_FINALIZATION_MARKER,
+        f"finalized_at: {finalized_text}",
+        f"finalized_by_identity: {merger_identity}",
+        f"finalized_by_profile: {merger_profile}",
+        f"finalized_by_session_id: {merger_session_id}",
+        f"finalization_outcome: {outcome}",
+        f"finalization_reason: {reason}",
+        f"finalized_lease_comment_id: {lease_comment_id or 'none'}",
+        lease_body,
+    ]
+    return "\n".join(lines)
+
+
+def is_merger_finalization_comment(body: str) -> bool:
+    return MERGER_FINALIZATION_MARKER in (body or "")
+
+
+def assess_merger_lease_finalization(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    session: dict[str, Any] | None,
+    actor_identity: str,
+    actor_profile: str,
+    actor_session_id: str | None,
+    repo: str,
+    worktree: str,
+    candidate_head: str | None,
+    live_head_sha: str | None = None,
+    outcome: str = OUTCOME_RELEASED,
+    reason: str | None = None,
+    runtime_token_fingerprint: str | None = None,
+    issue_number: int | None = None,
+    target_branch: str = "master",
+    target_branch_sha: str | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Decide whether a merger may terminally finalize its own lease (#742).
+
+    Owner-session only: the caller must hold the exact comment-backed merger
+    lease it is finalizing. Foreign sessions, reviewer profiles, and mismatched
+    repository/PR/head/token-fingerprint callers fail closed. Already-terminal
+    leases return ``already_terminal`` so repeat calls are idempotent and post
+    no second marker.
+    """
+    now = now or datetime.now(timezone.utc)
+    reasons: list[str] = []
+    outcome = (outcome or "").strip().lower()
+    finalization_reason = (reason or "").strip() or DEFAULT_MERGER_FINALIZATION_REASON
+
+    if outcome not in MERGER_FINALIZATION_OUTCOMES:
+        reasons.append(
+            f"outcome '{outcome or 'none'}' is not a terminal merger "
+            f"finalization outcome ({sorted(MERGER_FINALIZATION_OUTCOMES)})"
+        )
+
+    if "merger" not in (actor_profile or "").lower():
+        reasons.append(
+            f"profile '{actor_profile or 'unknown'}' is not a merger profile; "
+            "merger lease finalization is merger-only (fail closed). Reviewer "
+            "sessions use gitea_release_reviewer_pr_lease."
+        )
+
+    session = session or None
+    provenance = (session or {}).get("lease_provenance") or {}
+    if not isinstance(provenance, dict):
+        provenance = {}
+    source = (provenance.get("source") or "").strip()
+
+    if not session:
+        reasons.append(
+            "no in-session merger lease recorded; only the owning session may "
+            "finalize a merger lease"
+        )
+    elif source not in MERGER_OWNED_PROVENANCE_SOURCES:
+        reasons.append(
+            f"in-session lease provenance '{source or 'none'}' is not a "
+            "merger-owned lease; refusing to finalize"
+        )
+    elif not is_sanctioned_session_lease(session):
+        reasons.extend(
+            assess_acquired_merger_lease_integrity(session)
+            if source == SOURCE_ACQUIRE_MERGER
+            else ["in-session merger lease lacks sanctioned provenance"]
+        )
+
+    pinned = leases._normalize_sha(candidate_head)
+    live = leases._normalize_sha(live_head_sha)
+    if not pinned:
+        reasons.append(
+            "candidate_head is required for merger lease finalization "
+            "(exact-head scoping; fail closed)"
+        )
+    if live and pinned and live != pinned:
+        reasons.append(
+            "candidate_head does not match live PR head (fail closed)"
+        )
+
+    if session:
+        session_sid = (session.get("session_id") or "").strip()
+        actor_sid = (actor_session_id or "").strip()
+        if not actor_sid or session_sid != actor_sid:
+            reasons.append(
+                "session_id does not match the in-session merger lease owner; "
+                "foreign-session release is not permitted (fail closed)"
+            )
+        if session.get("pr_number") != pr_number:
+            reasons.append(
+                f"in-session merger lease is for PR #{session.get('pr_number')}, "
+                f"not #{pr_number}"
+            )
+        session_repo = (session.get("repo") or "").strip()
+        if session_repo and session_repo != (repo or "").strip():
+            reasons.append(
+                f"in-session merger lease repository '{session_repo}' does not "
+                f"match '{repo}' (fail closed)"
+            )
+        session_head = leases._normalize_sha(session.get("candidate_head"))
+        if pinned and session_head and session_head != pinned:
+            reasons.append(
+                "in-session merger lease candidate_head does not match the "
+                "supplied candidate_head (fail closed)"
+            )
+        session_identity = (session.get("reviewer_identity") or "").strip()
+        if session_identity and session_identity != (actor_identity or "").strip():
+            reasons.append(
+                "authenticated identity does not match the in-session merger "
+                "lease holder (fail closed)"
+            )
+        session_profile = (session.get("profile") or "").strip()
+        if session_profile and session_profile != (actor_profile or "").strip():
+            reasons.append(
+                "active profile does not match the in-session merger lease "
+                "profile (fail closed)"
+            )
+        recorded_fingerprint = (
+            provenance.get("native_token_fingerprint") or ""
+        ).strip()
+        live_fingerprint = (runtime_token_fingerprint or "").strip()
+        if (
+            recorded_fingerprint
+            and live_fingerprint
+            and recorded_fingerprint != live_fingerprint
+        ):
+            reasons.append(
+                "native runtime token fingerprint does not match the one "
+                "recorded when the merger lease was acquired (fail closed)"
+            )
+
+    lease_comment_id = (session or {}).get("comment_id") or provenance.get("comment_id")
+    entries = leases._lease_entries(comments, pr_number=pr_number)
+    if session and lease_comment_id is not None:
+        if not any(entry.get("comment_id") == lease_comment_id for entry in entries):
+            reasons.append(
+                f"lease marker comment {lease_comment_id} is not present on PR "
+                f"#{pr_number} (fail closed)"
+            )
+
+    active = leases.find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+    already_terminal = False
+    if active:
+        owner_session = (active.get("session_id") or "").strip()
+        if owner_session and owner_session != (actor_session_id or "").strip():
+            reasons.append(
+                f"active PR lease is owned by session_id={owner_session}; a "
+                "merger may only finalize its own lease (fail closed)"
+            )
+    else:
+        newest = entries[-1] if entries else None
+        newest_phase = ((newest or {}).get("phase") or "").strip().lower()
+        if newest and newest_phase in leases._TERMINAL_PHASES:
+            already_terminal = True
+        elif not entries:
+            reasons.append(
+                f"no comment-backed lease marker found on PR #{pr_number}"
+            )
+
+    finalize_allowed = not reasons and not already_terminal
+    body = None
+    if finalize_allowed and session:
+        body = format_merger_finalization_body(
+            repo=(session.get("repo") or repo),
+            pr_number=pr_number,
+            issue_number=(
+                issue_number
+                if issue_number is not None
+                else session.get("issue_number")
+            ),
+            merger_identity=actor_identity,
+            merger_profile=actor_profile,
+            merger_session_id=(actor_session_id or ""),
+            worktree=worktree or (session.get("worktree") or ""),
+            candidate_head=pinned,
+            target_branch=(
+                session.get("target_branch") or target_branch or "master"
+            ),
+            target_branch_sha=(
+                session.get("target_branch_sha") or target_branch_sha
+            ),
+            outcome=outcome,
+            reason=finalization_reason,
+            lease_comment_id=lease_comment_id,
+            finalized_at=now,
+        )
+
+    return {
+        "finalize_allowed": finalize_allowed,
+        "already_terminal": already_terminal,
+        "reasons": reasons,
+        "outcome": outcome,
+        "finalization_reason": finalization_reason,
+        "active_lease": active,
+        "finalization_body": body,
+        "lease_comment_id": lease_comment_id,
+        "candidate_head": pinned,
+    }
+
+
 def assess_adopt_merger_lease(
     comments: list[dict],
     *,

namespace_workspace_binding.py

@@ -109,8 +109,17 @@ def resolve_namespace_mutation_context(
     session_lease_worktree: str | None = None,
     worktree: str | None = None,
     profile_name: str | None = None,
+    configured_canonical_root: str | None = None,
 ) -> dict:
-    """Shared workspace resolution for runtime_context and mutation guards."""
+    """Shared workspace resolution for runtime_context and mutation guards.
+
+    When *configured_canonical_root* is supplied (a cross-repository namespace
+    bound to an external target repository, #706), the canonical repository root
+    is that configured target rather than the MCP install checkout. This keeps
+    the branches-only / worktree-membership guards (#274) evaluating against the
+    repository the namespace actually mutates. Without it the single-repo
+    default is preserved: the canonical root follows the process checkout.
+    """
     demotions: list[str] = []
     workspace, binding_source = resolve_namespace_workspace(
         role_kind=role_kind,
@@ -132,6 +141,10 @@ def resolve_namespace_mutation_context(
         env=env,
         profile_name=profile_name,
     )
+    configured = (configured_canonical_root or "").strip()
+    if configured:
+        canonical_root = os.path.realpath(configured)
+    else:
         canonical_root = amw.resolve_canonical_repo_root(process_root, process_root)
     return {
         "workspace_path": workspace,
@@ -258,6 +271,7 @@ def assess_namespace_mutation_workspace(
     session_lease_worktree: str | None = None,
     profile_name: str | None = None,
     current_branch: str | None = None,
+    configured_canonical_root: str | None = None,
 ) -> dict:
     """Evaluate namespace workspace binding before preflight/mutation."""
     ctx = resolve_namespace_mutation_context(
@@ -268,6 +282,7 @@ def assess_namespace_mutation_workspace(
         env=env,
         session_lease_worktree=session_lease_worktree,
         profile_name=profile_name,
+        configured_canonical_root=configured_canonical_root,
     )
     mutation_workspace = ctx["workspace_path"]
     binding_source = ctx["workspace_binding_source"]

pr_work_lease.py

@@ -20,7 +20,11 @@ _FIELD_RE = re.compile(
     re.IGNORECASE | re.MULTILINE,
 )
 
-_TERMINAL_REVIEWER_PHASES = frozenset({"done", "released", "blocked"})
+# Must mirror reviewer_pr_lease._TERMINAL_PHASES: both modules read the same
+# append-only lease markers, so a phase that is terminal in one and active in
+# the other yields two conflicting truths for the same comment (#742 review
+# 460). "abandoned" is the owner-session merger finalization outcome.
+_TERMINAL_REVIEWER_PHASES = frozenset({"done", "released", "blocked", "abandoned"})
 _ACTIVE_REVIEWER_PHASES = frozenset({
     "claimed",
     "validating",
@@ -32,7 +36,8 @@ _TERMINAL_CONFLICT_FIX_PHASES = frozenset({"released", "blocked", "done"})
 _ACTIVE_CONFLICT_FIX_PHASES = frozenset({"claimed", "pushing", "pushed"})
 
 DEFAULT_CONFLICT_FIX_TTL_MINUTES = 120
-DEFAULT_REVIEWER_LEASE_TTL_MINUTES = 120
+# The reviewer/merger PR-lease TTL lives in reviewer_pr_lease.LEASE_TTL_MINUTES
+# (#747). A second copy here had no readers and could only drift out of sync.
 
 
 def _parse_timestamp(value: str | None) -> datetime | None:
@@ -153,25 +158,72 @@ def _lease_phase_active(lease: dict, *, active_phases: frozenset[str]) -> bool:
     ))
 
 
+def _reviewer_chain_key(lease: dict) -> tuple | None:
+    """Identity of the lease chain a reviewer marker belongs to (#742).
+
+    A chain is one session's claim → heartbeat → terminal sequence, keyed by
+    repository, PR, candidate head, session id, identity, and profile. Returns
+    None when any component is missing: an incomplete or malformed marker has
+    no provable chain, so it can neither be cancelled by nor cancel anything.
+    """
+    raw = lease.get("raw_fields") or {}
+    repo = (raw.get("repo") or "").strip().lower()
+    session_id = (lease.get("session_id") or "").strip()
+    identity = (lease.get("reviewer_identity") or "").strip()
+    profile = (lease.get("profile") or "").strip()
+    head = lease.get("candidate_head")
+    pr_number = lease.get("pr_number")
+    if not (repo and session_id and identity and profile and head and pr_number):
+        return None
+    return (repo, pr_number, head, session_id, identity, profile)
+
+
+def _chain_terminated_after(entries: list[dict], index: int) -> bool:
+    """True when a later marker terminates the chain of ``entries[index]``.
+
+    Append-only newest-wins (#577 semantics, chain-scoped for #742): a terminal
+    marker ends only its *own* claim, so a foreign, forged, or malformed
+    terminal marker cannot cancel another session's valid active lease.
+    """
+    key = _reviewer_chain_key(entries[index])
+    if key is None:
+        return False
+    for later in entries[index + 1:]:
+        if (later.get("phase") or "").strip().lower() not in _TERMINAL_REVIEWER_PHASES:
+            continue
+        if _reviewer_chain_key(later) == key:
+            return True
+    return False
+
+
 def find_active_reviewer_lease(
     comments: list[dict],
     *,
     pr_number: int,
     now: datetime | None = None,
 ) -> dict[str, Any] | None:
-    """Return the newest unexpired reviewer lease for *pr_number*, if any."""
+    """Return the newest unexpired, non-terminated reviewer lease for *pr_number*.
+
+    Walking backward past a terminal marker used to resurrect the older claim of
+    the very chain that marker ended, so a released/abandoned finalization still
+    read as an active lease here while ``reviewer_pr_lease`` reported it ended
+    (#742). A claim is now skipped when a later marker terminates its own chain.
+    """
     now = now or datetime.now(timezone.utc)
     candidates = [
         entry for entry in _comment_entries(comments, pr_number=pr_number)
         if entry.get("lease_kind") == "reviewer"
     ]
-    for lease in reversed(candidates):
+    for index in range(len(candidates) - 1, -1, -1):
+        lease = candidates[index]
         if _lease_expired(lease, now=now):
             continue
         phase = (lease.get("phase") or "").strip().lower()
         if phase in _TERMINAL_REVIEWER_PHASES:
             continue
         if phase in _ACTIVE_REVIEWER_PHASES or phase:
+            if _chain_terminated_after(candidates, index):
+                continue
             return lease
     return None
 

reviewer_pr_lease.py

@@ -16,7 +16,10 @@ _FIELD_RE = re.compile(
 )
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
 
-_TERMINAL_PHASES = frozenset({"done", "released", "blocked"})
+# "abandoned" is the owner-session merger finalization outcome (#742). Without
+# it here, an abandoned marker would fall through to the generic non-empty
+# phase branch and keep re-arming the lease as active.
+_TERMINAL_PHASES = frozenset({"done", "released", "blocked", "abandoned"})
 _ACTIVE_PHASES = frozenset({
     "claimed",
     "validating",
@@ -26,9 +29,19 @@ _ACTIVE_PHASES = frozenset({
     "adopted",
 })
 
-DEFAULT_LEASE_TTL_MINUTES = 120
+# Reviewer and merger PR leases use a short *sliding* window (#747): a lease
-STALE_WARNING_MINUTES = 30
+# expires 10 minutes after its last heartbeat, and every heartbeat slides the
-RECLAIMABLE_MINUTES = 60
+# expiry forward. An actively heartbeating session is never evicted, while a
+# dead session releases its hold in at most one TTL instead of the two hours
+# the previous fixed 120-minute expiry allowed.
+LEASE_TTL_MINUTES = 10
+# Renewal is named separately from acquisition so the slide amount is tunable
+# without silently re-defining how long a fresh lease lives.
+LEASE_RENEWAL_MINUTES = 10
+# Retained for callers that imported the pre-#747 name.
+DEFAULT_LEASE_TTL_MINUTES = LEASE_TTL_MINUTES
+# Warn at half the window, while the owner can still heartbeat and recover.
+STALE_WARNING_MINUTES = 5
 
 _SESSION_LEASE: dict[str, Any] | None = None
 
@@ -77,10 +90,19 @@ def format_lease_body(
     target_branch_sha: str | None,
     last_activity: datetime | None = None,
     expires_at: datetime | None = None,
+    ttl_minutes: int = LEASE_TTL_MINUTES,
     blocker: str = "none",
 ) -> str:
+    """Serialize a lease marker.
+
+    Every write of this marker — acquisition, heartbeat, adoption — re-derives
+    ``expires_at`` from the moment of the write, which is what makes the TTL
+    slide (#747). Callers renewing an existing lease pass
+    ``ttl_minutes=LEASE_RENEWAL_MINUTES``; an explicit ``expires_at`` still
+    wins so a lease can be minted with a deliberate window.
+    """
     now = last_activity or datetime.now(timezone.utc)
-    expires = expires_at or (now + timedelta(minutes=DEFAULT_LEASE_TTL_MINUTES))
+    expires = expires_at or (now + timedelta(minutes=ttl_minutes))
     last_text = now.astimezone(timezone.utc).replace(microsecond=0).isoformat().replace(
         "+00:00", "Z"
     )
@@ -166,8 +188,30 @@ def _minutes_since_activity(lease: dict, *, now: datetime) -> float | None:
     return (now - last).total_seconds() / 60.0
 
 
+def lease_seconds_remaining(lease: dict, *, now: datetime | None = None) -> int | None:
+    """Seconds until *lease* expires, clamped at 0; ``None`` if unparsable.
+
+    Lets diagnostics distinguish "held and live" from "held and dying" (#747)
+    rather than only reporting that a lease exists.
+    """
+    expires_at = _parse_timestamp(lease.get("expires_at"))
+    if not expires_at:
+        return None
+    now = now or datetime.now(timezone.utc)
+    return max(0, int((expires_at - now).total_seconds()))
+
+
 def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str:
-    """Return active, stale_warning, reclaimable, expired, or terminal."""
+    """Return active, stale_warning, expired, or terminal.
+
+    Expiry is the only takeover gate (#747). The pre-#747 ``reclaimable`` band
+    sat between "stale" and "expired" and made a dead lease wait out a second
+    timer before anyone could reclaim it. Under a sliding TTL that band is also
+    unreachable: a heartbeat stamps ``last_activity`` and ``expires_at``
+    together, so a lease idle for a full TTL is already expired. Foreign
+    expired leases are handled by the ``foreign_expired`` classification, which
+    carries the same sanctioned release next-action the old tier did.
+    """
     now = now or datetime.now(timezone.utc)
     phase = (lease.get("phase") or "").strip().lower()
     if phase in _TERMINAL_PHASES:
@@ -177,8 +221,6 @@ def classify_lease_freshness(lease: dict, *, now: datetime | None = None) -> str
     minutes = _minutes_since_activity(lease, now=now)
     if minutes is None:
         return "active"
-    if minutes >= RECLAIMABLE_MINUTES:
-        return "reclaimable"
     if minutes >= STALE_WARNING_MINUTES:
         return "stale_warning"
     return "active"

session_context_binding.py

@@ -32,6 +32,7 @@ class _SessionContext:
     expected_username: str | None
     source: str
     pid: int
+    canonical_repository_root: str | None = None
 
     def as_dict(self) -> dict[str, Any]:
         return {
@@ -45,6 +46,7 @@ class _SessionContext:
             "expected_username": self.expected_username,
             "source": self.source,
             "pid": self.pid,
+            "canonical_repository_root": self.canonical_repository_root,
         }
 
 
@@ -143,6 +145,7 @@ def bind_session_context(
     role_kind: str | None = None,
     expected_username: str | None = None,
     source: str = "bind",
+    canonical_repository_root: str | None = None,
 ) -> dict[str, Any]:
     """Atomically bind/re-bind context (the explicit activation path)."""
     with _SESSION_CONTEXT_LOCK:
@@ -156,6 +159,7 @@ def bind_session_context(
             role_kind=role_kind,
             expected_username=expected_username,
             source=source,
+            canonical_repository_root=canonical_repository_root,
         )
 
 
@@ -170,6 +174,7 @@ def _bind_session_context_unlocked(
     role_kind: str | None,
     expected_username: str | None,
     source: str,
+    canonical_repository_root: str | None = None,
 ) -> dict[str, Any]:
     """Store a complete immutable context while the caller holds the lock."""
     global _SESSION_CONTEXT
@@ -184,6 +189,7 @@ def _bind_session_context_unlocked(
         expected_username=(expected_username or "").strip() or None,
         source=source,
         pid=os.getpid(),
+        canonical_repository_root=(canonical_repository_root or "").strip() or None,
     )
     return _SESSION_CONTEXT.as_dict()
 
@@ -199,6 +205,7 @@ def seed_session_context_if_unbound(
     role_kind: str | None = None,
     expected_username: str | None = None,
     source: str = "seed",
+    canonical_repository_root: str | None = None,
 ) -> dict[str, Any]:
     """Atomically bind only when this process has no current context.
 
@@ -219,6 +226,7 @@ def seed_session_context_if_unbound(
                 role_kind=role_kind,
                 expected_username=expected_username,
                 source=source,
+                canonical_repository_root=canonical_repository_root,
             )
         return _SESSION_CONTEXT.as_dict()
 
@@ -232,6 +240,7 @@ def assess_session_context(
     repository: str | None = None,
     org: str | None = None,
     expected_username: str | None = None,
+    canonical_repository_root: str | None = None,
     require_bound: bool = False,
     require_complete: bool = False,
 ) -> dict[str, Any]:
@@ -272,6 +281,7 @@ def assess_session_context(
     live_identity = (identity or "").strip() or None
     live_repo = (repository or "").strip() or None
     live_org = (org or "").strip() or None
+    live_canonical = (canonical_repository_root or "").strip() or None
 
     if ctx.get("profile_name") and live_profile and live_profile != ctx.get("profile_name"):
         reasons.append(
@@ -307,6 +317,13 @@ def assess_session_context(
             f"org drift: live '{live_org}' != bound "
             f"'{ctx.get('org')}' (fail closed)"
         )
+    bound_canonical = ctx.get("canonical_repository_root")
+    if bound_canonical and live_canonical and live_canonical != bound_canonical:
+        reasons.append(
+            f"canonical repository root drift: live '{live_canonical}' != bound "
+            f"'{bound_canonical}' (forged or conflicting cross-repository "
+            "binding, fail closed)"
+        )
 
     expected = expected_username or ctx.get("expected_username")
     if expected and live_identity and live_identity != expected:
@@ -580,6 +597,7 @@ def mutation_context_audit_fields(
         "session_org": data.get("org"),
         "session_role_kind": data.get("role_kind"),
         "session_context_source": data.get("source"),
+        "session_canonical_repository_root": data.get("canonical_repository_root"),
     }
 
 

task_capability_map.py

@@ -137,6 +137,17 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "merger",
     },
+    # #742: owner-session terminal release/abandon of a merger-held lease when
+    # the merge does not occur. Apply path posts an append-only terminal lease
+    # marker (gitea.pr.comment); merger-only, never a reviewer path.
+    "release_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
+    "gitea_release_merger_pr_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "merger",
+    },
     # #691: guarded non-owner cleanup of obsolete comment-backed reviewer leases.
     # Apply path posts lease release + audit comments (gitea.pr.comment).
     "cleanup_obsolete_reviewer_comment_lease": {

tests/test_cross_repo_canonical_consumption.py

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

tests/test_issue_706_canonical_repository_root.py

@@ -0,0 +1,423 @@
+"""Issue #706: immutable startup/profile canonical_repository_root capability.
+
+Cross-repository MCP namespaces (e.g. ``eagenda-author``) must bind their
+canonical repository root to the *configured target repository*, not to the
+Gitea-Tools install checkout the server script lives in. These tests prove:
+
+* the configured binding is resolved from profile/env with env precedence,
+* the target repository identity and git common-directory membership are
+  validated (AC3),
+* the branches-only guard (#274) is enforced *inside the target repo* (AC4),
+* missing / conflicting / forged bindings fail closed (AC5),
+* prgs and mdcps namespaces stay simultaneously isolated (AC6),
+* the session binding pins the canonical root immutably,
+* no weakening of the install-root single-repo default (AC8).
+
+Uses two distinct real git repositories/worktrees (AC7).
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import canonical_repository_root as crr  # noqa: E402
+import gitea_config  # noqa: E402
+import namespace_workspace_binding as nwb  # noqa: E402
+import session_context_binding as session_ctx  # noqa: E402
+
+
+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:
+    """Create a real git repo with one commit and a remote; return realpath root."""
+    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))
+
+
+def _add_worktree(repo_root: str, worktree_path: Path, branch: str) -> str:
+    _git(repo_root, "worktree", "add", "-q", "-b", branch, str(worktree_path))
+    return os.path.realpath(str(worktree_path))
+
+
+# ---------------------------------------------------------------------------
+# configured_canonical_root: resolution + precedence
+# ---------------------------------------------------------------------------
+class TestConfiguredCanonicalRoot(unittest.TestCase):
+    def test_unconfigured_returns_none(self):
+        value, source = crr.configured_canonical_root({}, {})
+        self.assertIsNone(value)
+        self.assertIsNone(source)
+
+    def test_profile_field_used(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "/repo/eAgenda"}, {}
+        )
+        self.assertEqual(value, "/repo/eAgenda")
+        self.assertIn("profile", source)
+
+    def test_env_overrides_profile(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "/repo/eAgenda"},
+            {crr.CANONICAL_ROOT_ENV: "/repo/other"},
+        )
+        self.assertEqual(value, "/repo/other")
+        self.assertIn(crr.CANONICAL_ROOT_ENV, source)
+
+    def test_blank_values_ignored(self):
+        value, source = crr.configured_canonical_root(
+            {"canonical_repository_root": "   "},
+            {crr.CANONICAL_ROOT_ENV: ""},
+        )
+        self.assertIsNone(value)
+        self.assertIsNone(source)
+
+
+# ---------------------------------------------------------------------------
+# assess_canonical_repository_root: default (single-repo) path
+# ---------------------------------------------------------------------------
+class TestUnconfiguredFallback(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = self._tmp.name
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_unconfigured_falls_back_to_process_root(self):
+        root = _init_repo(
+            Path(self.tmp) / "install",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        got = crr.assess_canonical_repository_root(
+            configured_value=None,
+            source=None,
+            expected_slug=None,
+            process_project_root=root,
+        )
+        self.assertTrue(got["proven"])
+        self.assertFalse(got["block"])
+        self.assertFalse(got["configured"])
+        self.assertEqual(got["canonical_repo_root"], root)
+
+
+# ---------------------------------------------------------------------------
+# assess_canonical_repository_root: configured cross-repo path
+# ---------------------------------------------------------------------------
+class TestConfiguredCrossRepo(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.install = _init_repo(
+            self.tmp / "Gitea-Tools",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        self.target = _init_repo(
+            self.tmp / "mcp-control-plane",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_valid_configured_root_binds_to_target(self):
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.target,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["proven"], got.get("reasons"))
+        self.assertFalse(got["block"])
+        self.assertTrue(got["configured"])
+        self.assertEqual(got["canonical_repo_root"], self.target)
+        self.assertNotEqual(got["canonical_repo_root"], self.install)
+
+    def test_missing_path_fails_closed(self):
+        got = crr.assess_canonical_repository_root(
+            configured_value=str(self.tmp / "does-not-exist"),
+            source="profile canonical_repository_root",
+            expected_slug=None,
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertFalse(got["proven"])
+        self.assertTrue(any("exist" in r for r in got["reasons"]))
+
+    def test_non_git_path_fails_closed(self):
+        plain = self.tmp / "plain"
+        plain.mkdir()
+        got = crr.assess_canonical_repository_root(
+            configured_value=str(plain),
+            source="profile canonical_repository_root",
+            expected_slug=None,
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("git" in r for r in got["reasons"]))
+
+    def test_identity_mismatch_fails_closed(self):
+        # Configured root is the install repo, but session expects the target
+        # repo identity: a forged/conflicting binding.
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.install,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("identity" in r for r in got["reasons"]))
+
+    def test_unprovable_identity_blocks_when_required(self):
+        noremote = _init_repo(self.tmp / "noremote-src", "x", remote_name="origin")
+        _git(noremote, "remote", "remove", "origin")
+        got = crr.assess_canonical_repository_root(
+            configured_value=noremote,
+            source="profile canonical_repository_root",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.install,
+            require_binding=True,
+        )
+        self.assertTrue(got["block"])
+
+    def test_repository_identity_slug_reads_remote(self):
+        self.assertEqual(
+            crr.repository_identity_slug(self.target),
+            "Scaled-Tech-Consulting/mcp-control-plane",
+        )
+
+
+# ---------------------------------------------------------------------------
+# Workspace membership + branches-only enforced inside the TARGET repo (AC4)
+# ---------------------------------------------------------------------------
+class TestNamespaceContextUsesConfiguredRoot(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.install = _init_repo(
+            self.tmp / "Gitea-Tools",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git",
+        )
+        self.target = _init_repo(
+            self.tmp / "mcp-control-plane",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+        (Path(self.target) / "branches").mkdir()
+        self.target_wt = _add_worktree(
+            self.target,
+            Path(self.target) / "branches" / "author-issue-1",
+            "feat/issue-1",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_context_canonical_root_is_configured_target(self):
+        ctx = nwb.resolve_namespace_mutation_context(
+            role_kind="author",
+            worktree_path=self.target_wt,
+            process_project_root=self.install,
+            env={},
+            configured_canonical_root=self.target,
+        )
+        self.assertEqual(ctx["canonical_repo_root"], self.target)
+        self.assertFalse(ctx["roots_aligned"])
+
+    def test_target_worktree_is_member_of_target_root(self):
+        got = nwb.amw.assess_workspace_repo_membership(
+            workspace_path=self.target_wt,
+            canonical_repo_root=self.target,
+        )
+        self.assertTrue(got["proven"], got.get("reasons"))
+
+    def test_target_worktree_not_member_of_install_root(self):
+        got = nwb.amw.assess_workspace_repo_membership(
+            workspace_path=self.target_wt,
+            canonical_repo_root=self.install,
+        )
+        self.assertTrue(got["block"])
+
+    def test_branches_guard_passes_inside_target(self):
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="author",
+            worktree_path=self.target_wt,
+            worktree=None,
+            process_project_root=self.install,
+            env={},
+            current_branch="feat/issue-1",
+            configured_canonical_root=self.target,
+        )
+        self.assertFalse(assessment["block"], assessment.get("reasons"))
+        self.assertEqual(assessment["canonical_repo_root"], self.target)
+
+    def test_target_control_checkout_blocks_branches_guard(self):
+        assessment = nwb.assess_namespace_mutation_workspace(
+            role_kind="author",
+            worktree_path=self.target,  # stable target checkout, not branches/
+            worktree=None,
+            process_project_root=self.install,
+            env={},
+            current_branch="master",
+            configured_canonical_root=self.target,
+        )
+        self.assertTrue(assessment["block"])
+
+
+# ---------------------------------------------------------------------------
+# Immutable session pin (AC1/AC5)
+# ---------------------------------------------------------------------------
+class TestSessionCanonicalRootPin(unittest.TestCase):
+    def setUp(self):
+        os.environ["PYTEST_CURRENT_TEST"] = "t"
+        session_ctx._reset_session_context_for_testing()
+
+    def tearDown(self):
+        session_ctx._reset_session_context_for_testing()
+
+    def test_seed_stores_canonical_root(self):
+        ctx = session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        self.assertEqual(ctx["canonical_repository_root"], "/repo/mcp-control-plane")
+
+    def test_canonical_root_drift_fails_closed(self):
+        session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        got = session_ctx.assess_session_context(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            canonical_repository_root="/repo/forged-elsewhere",
+        )
+        self.assertTrue(got["block"])
+        self.assertTrue(any("canonical" in r for r in got["reasons"]))
+
+    def test_matching_canonical_root_passes(self):
+        session_ctx.seed_session_context_if_unbound(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            host="gitea.prgs.cc",
+            identity="svc",
+            repository="mcp-control-plane",
+            org="Scaled-Tech-Consulting",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        got = session_ctx.assess_session_context(
+            profile_name="mcp-control-plane-author",
+            remote="prgs",
+            canonical_repository_root="/repo/mcp-control-plane",
+        )
+        self.assertFalse(got["block"], got["reasons"])
+
+
+# ---------------------------------------------------------------------------
+# Simultaneous prgs / mdcps isolation (AC6)
+# ---------------------------------------------------------------------------
+class TestSimultaneousIsolation(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        self.tmp = Path(self._tmp.name)
+        self.prgs = _init_repo(
+            self.tmp / "prgs-repo",
+            "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git",
+        )
+        self.mdcps = _init_repo(
+            self.tmp / "mdcps-repo",
+            "https://gitea.dadeschools.net/dadeschools/eAgenda.git",
+        )
+
+    def tearDown(self):
+        self._tmp.cleanup()
+
+    def test_each_namespace_binds_its_own_target(self):
+        a = crr.assess_canonical_repository_root(
+            configured_value=self.prgs,
+            source="env",
+            expected_slug="Scaled-Tech-Consulting/mcp-control-plane",
+            process_project_root=self.tmp.as_posix(),
+        )
+        b = crr.assess_canonical_repository_root(
+            configured_value=self.mdcps,
+            source="env",
+            expected_slug="dadeschools/eAgenda",
+            process_project_root=self.tmp.as_posix(),
+        )
+        self.assertEqual(a["canonical_repo_root"], self.prgs)
+        self.assertEqual(b["canonical_repo_root"], self.mdcps)
+        self.assertNotEqual(a["canonical_repo_root"], b["canonical_repo_root"])
+
+    def test_cross_wired_identity_blocks(self):
+        # prgs path claimed under the mdcps identity → forged binding.
+        got = crr.assess_canonical_repository_root(
+            configured_value=self.prgs,
+            source="env",
+            expected_slug="dadeschools/eAgenda",
+            process_project_root=self.tmp.as_posix(),
+        )
+        self.assertTrue(got["block"])
+
+
+# ---------------------------------------------------------------------------
+# Config validation of the profile field (AC5: malformed bindings fail closed)
+# ---------------------------------------------------------------------------
+class TestConfigValidation(unittest.TestCase):
+    def test_absent_is_allowed(self):
+        # single-repo default: no field configured
+        gitea_config._validate_canonical_repository_root("p", None)
+
+    def test_valid_absolute_path_ok(self):
+        gitea_config._validate_canonical_repository_root(
+            "p", "/Users/x/Development/mcp-control-plane"
+        )
+
+    def test_relative_path_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root(
+                "p", "relative/path"
+            )
+
+    def test_empty_string_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root("p", "   ")
+
+    def test_non_string_rejected(self):
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config._validate_canonical_repository_root("p", ["/x"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_706_f1_seed_identity_integration.py

@@ -0,0 +1,221 @@
+"""Issue #706 F1 integration regression (review 457).
+
+The unit tests in ``test_issue_706_canonical_repository_root.py`` exercise
+``crr.assess_canonical_repository_root`` / ``session_ctx.seed_session_context_if_unbound``
+with a *preselected* slug, so they never drive the real end-to-end path that
+review 457 found broken:
+
+    _seed_session_context -> _trusted_session_repository -> _workspace_repository_slug
+    -> _local_git_remote_url (cwd=PROJECT_ROOT, always Gitea-Tools)
+
+Before the fix, the session repository identity was pinned from the *install*
+checkout remote even when a cross-repository ``canonical_repository_root`` was
+configured to an external repository (e.g. mcp-control-plane). The mutation
+preflight then derived ``expected_slug`` from that install-derived pin and
+``_enforce_canonical_repository_root`` failed closed on a self-inflicted
+identity mismatch, so the stated cross-repo namespaces stayed blocked.
+
+These tests use two *real* temporary git repositories and drive the live
+``mcp_server._seed_session_context`` and ``mcp_server._enforce_canonical_repository_root``
+functions, asserting the session pins the *configured target* identity and that
+enforcement accepts the target worktree — while every fail-closed property is
+preserved.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import mcp_server  # noqa: E402  # loads gitea_mcp_server.py into this namespace
+import canonical_repository_root as crr  # noqa: E402
+import session_context_binding as session_ctx  # noqa: E402
+
+INSTALL_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+TARGET_SLUG = "Scaled-Tech-Consulting/mcp-control-plane"
+TARGET_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/mcp-control-plane.git"
+OTHER_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/other-repo.git"
+
+
+def _git(cwd: str, *args: str) -> str:
+    return subprocess.run(
+        ["git", "-C", cwd, *args], capture_output=True, text=True, check=True
+    ).stdout.strip()
+
+
+def _init_repo(path: Path, remote_url: str) -> 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", "prgs", 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))
+
+
+def _add_worktree(repo_root: str, wt: Path, branch: str) -> str:
+    _git(repo_root, "worktree", "add", "-q", "-b", branch, str(wt))
+    return os.path.realpath(str(wt))
+
+
+def _profile(role: str, *, canonical: str | None = None,
+             allowed=(TARGET_SLUG,)) -> dict:
+    p = {
+        "profile_name": f"mcp-control-plane-{role}",
+        "role": role,
+        "username": "svc",
+        "allowed_operations": ["gitea.read"],
+        "forbidden_operations": [],
+        "allowed_repositories": list(allowed),
+    }
+    if canonical is not None:
+        p["canonical_repository_root"] = canonical
+    return p
+
+
+class _Base(unittest.TestCase):
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory()
+        tmp = Path(self._tmp.name)
+        self.install = _init_repo(tmp / "Gitea-Tools", INSTALL_URL)
+        self.target = _init_repo(tmp / "mcp-control-plane", TARGET_URL)
+        (Path(self.target) / "branches").mkdir()
+        self.target_wt = _add_worktree(
+            self.target, Path(self.target) / "branches" / "author-issue-1",
+            "feat/issue-1",
+        )
+        # PROJECT_ROOT and the install git remote are the Gitea-Tools install
+        # checkout — exactly the source that (mis)seeded the session before.
+        self._p_root = mock.patch.object(mcp_server, "PROJECT_ROOT", self.install)
+        self._p_url = mock.patch.object(
+            mcp_server, "_local_git_remote_url", return_value=INSTALL_URL
+        )
+        self._p_root.start()
+        self._p_url.start()
+        session_ctx._reset_session_context_for_testing()
+
+    def tearDown(self):
+        mock.patch.stopall()
+        session_ctx._reset_session_context_for_testing()
+
+    def _seed(self, profile, env=None):
+        session_ctx._reset_session_context_for_testing()
+        with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
+                mock.patch.dict(os.environ, env or {}, clear=False):
+            return mcp_server._seed_session_context(
+                profile=profile, remote="prgs", host="gitea.prgs.cc",
+                identity="svc",
+            )
+
+    def _enforce(self, profile, env=None):
+        with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
+                mock.patch.dict(os.environ, env or {}, clear=False):
+            mcp_server._enforce_canonical_repository_root(
+                self.target_wt, remote="prgs"
+            )
+
+
+class TestF1SeedsConfiguredTargetIdentity(_Base):
+    def test_env_config_seeds_target_not_install(self):
+        ctx = self._seed(
+            _profile("reviewer"),
+            env={crr.CANONICAL_ROOT_ENV: self.target},
+        )
+        self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
+        self.assertEqual(ctx["repository"], "mcp-control-plane")
+        # Regression guard: must NOT be the install repo.
+        self.assertNotEqual(ctx["repository"], "Gitea-Tools")
+
+    def test_profile_field_seeds_target_not_install(self):
+        ctx = self._seed(_profile("merger", canonical=self.target))
+        self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
+        self.assertEqual(ctx["repository"], "mcp-control-plane")
+        self.assertNotEqual(ctx["repository"], "Gitea-Tools")
+
+    def test_enforce_accepts_target_after_seed_reviewer(self):
+        prof = _profile("reviewer", canonical=self.target)
+        self._seed(prof)
+        # Would raise RuntimeError on the self-inflicted identity mismatch
+        # before the fix.
+        self._enforce(prof)
+
+    def test_enforce_accepts_target_after_seed_merger(self):
+        prof = _profile("merger", canonical=self.target)
+        self._seed(prof)
+        self._enforce(prof)
+
+    def test_env_overrides_profile_for_seed(self):
+        # profile points at install; env points at the real target → env wins.
+        prof = _profile("reviewer", canonical=self.install,
+                        allowed=(TARGET_SLUG,))
+        ctx = self._seed(prof, env={crr.CANONICAL_ROOT_ENV: self.target})
+        self.assertEqual(ctx["repository"], "mcp-control-plane")
+
+
+class TestUnconfiguredDefaultUnchanged(_Base):
+    def test_unconfigured_keeps_install_identity(self):
+        prof = _profile("author", allowed=("Scaled-Tech-Consulting/Gitea-Tools",))
+        ctx = self._seed(prof)  # no env, no profile canonical field
+        self.assertEqual(ctx["repository"], "Gitea-Tools")
+        self.assertEqual(ctx["org"], "Scaled-Tech-Consulting")
+
+
+class TestFailClosed(_Base):
+    def _trusted(self, profile, env=None, *, for_mutation=True):
+        with mock.patch.object(mcp_server, "get_profile", return_value=profile), \
+                mock.patch.dict(os.environ, env or {}, clear=False):
+            return mcp_server._trusted_session_repository(
+                profile, "prgs", for_mutation=for_mutation
+            )
+
+    def test_nonexistent_configured_root_fails_closed(self):
+        res = self._trusted(
+            _profile("reviewer"),
+            env={crr.CANONICAL_ROOT_ENV: self.target + "-missing"},
+        )
+        self.assertIsNone(res["repository"])
+        self.assertTrue(res["reasons"])
+
+    def test_non_git_configured_root_fails_closed(self):
+        plain = Path(self._tmp.name) / "plain"
+        plain.mkdir()
+        res = self._trusted(
+            _profile("reviewer"), env={crr.CANONICAL_ROOT_ENV: str(plain)}
+        )
+        self.assertIsNone(res["repository"])
+        self.assertTrue(any("git" in r for r in res["reasons"]))
+
+    def test_unallowlisted_target_identity_fails_closed(self):
+        # Configured target is valid, but the profile does not authorize it.
+        res = self._trusted(
+            _profile("reviewer", canonical=self.target,
+                     allowed=("Scaled-Tech-Consulting/Gitea-Tools",)),
+        )
+        self.assertIsNone(res["repository"])
+        self.assertTrue(any("scope" in r.lower() for r in res["reasons"]))
+
+    def test_forged_identity_conflict_fails_closed_at_enforce(self):
+        # Seed the target, then present a *different* configured root at
+        # enforcement time: the session pin no longer matches → fail closed.
+        other = _init_repo(Path(self._tmp.name) / "other", OTHER_URL)
+        prof_seed = _profile("reviewer", canonical=self.target)
+        self._seed(prof_seed)
+        prof_drift = _profile(
+            "reviewer", canonical=other,
+            allowed=(TARGET_SLUG, "Scaled-Tech-Consulting/other-repo"),
+        )
+        with self.assertRaises(RuntimeError):
+            self._enforce(prof_drift)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_741_canonical_root_consumers.py

@@ -0,0 +1,768 @@
+"""Complete PROJECT_ROOT elimination in cross-repository MCP operations (#741).
+
+#706 introduced the immutable ``canonical_repository_root``; #739/#740 routed
+three consumption paths through it. This module covers the paths #740 left
+behind, whose shape is uniform: the *filesystem* guards were migrated to the
+canonical root, but *repository identity* still bottomed out in
+``_local_git_remote_url``'s hardcoded ``cwd=PROJECT_ROOT`` — always the
+Gitea-Tools installation checkout.
+
+The consequences asserted here:
+
+* **Identity inversion.** ``_resolve``, the #530 remote/repo guard and the
+  anti-stomp org/repo fill all derived the *target* repository from the
+  *install* checkout, so a cross-repository namespace resolved Gitea-Tools
+  coordinates while its branch/parity facts came from the target repo.
+* **Guard disagreement.** ``_verify_role_mutation_workspace`` omitted
+  ``configured_canonical_root``, so the #274 branches-only / worktree-membership
+  guards validated ``Gitea-Tools/branches/`` rather than the bound target.
+* **Explicit-coordinate override.** Both-explicit ``org``/``repo`` short-circuit
+  ``assess_remote_repo_match``, so caller coordinates bypassed validation
+  entirely rather than merely *confirming* the binding.
+* **Configuration fail-open.** ``_flatten_identity`` silently dropped
+  ``canonical_repository_root``, so a v2-``environments`` namespace fell back to
+  the install root; the v1 path never validated the field at all.
+
+A role-by-operation matrix drives author, reviewer, merger and reconciler
+against: correct cross-repository target, wrong repository, wrong worktree,
+explicit matching coordinates, explicit mismatched coordinates, missing
+canonical root, invalid canonical root, immutable binding after first bind, and
+request-override attempts.
+
+Real git repositories are used throughout. No network calls are made, no branch
+is deleted, and no merge is performed.
+"""
+
+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 mcp_server  # 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"
+
+THIRD_REPO = "Timesheet"
+THIRD_SLUG = f"{INSTALL_ORG}/{THIRD_REPO}"
+
+# tests/conftest.py installs an autouse fixture
+# (mutation_profile_fixture.install_deterministic_remote_urls) that *permanently
+# reassigns* srv._local_git_remote_url to a stub mapping remote names to fixed
+# URLs. That stub deliberately ignores the working directory, which is exactly
+# the behaviour this module must verify — so these tests would silently assert
+# against the stub rather than production code in a full-suite run. Capture the
+# genuine implementation at import time (before any fixture executes) and
+# reinstall it per test.
+_REAL_LOCAL_GIT_REMOTE_URL = srv._local_git_remote_url
+
+
+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)
+    # Distinct content per repo: identical seed content, author and timestamp
+    # otherwise produce byte-identical commits and therefore an identical SHA,
+    # which would make the parity-dimension assertions vacuously true.
+    (path / "README.md").write_text(f"seed {remote_url}\n")
+    _git(str(path), "add", "README.md")
+    _git(str(path), "commit", "-q", "-m", f"seed {remote_name}")
+    return os.path.realpath(str(path))
+
+
+_ROLE_OPS = {
+    "author": (
+        [
+            "gitea.read",
+            "gitea.issue.create",
+            "gitea.issue.comment",
+            "gitea.pr.create",
+            "gitea.pr.comment",
+            "gitea.branch.create",
+            "gitea.branch.push",
+            "gitea.repo.commit",
+        ],
+        ["gitea.pr.approve", "gitea.pr.merge", "gitea.pr.request_changes"],
+    ),
+    "reviewer": (
+        [
+            "gitea.read",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+            "gitea.pr.comment",
+            "gitea.issue.comment",
+        ],
+        ["gitea.pr.create", "gitea.branch.push", "gitea.pr.merge"],
+    ),
+    "merger": (
+        ["gitea.read", "gitea.pr.merge", "gitea.pr.comment"],
+        [
+            "gitea.pr.create",
+            "gitea.branch.push",
+            "gitea.pr.approve",
+            "gitea.pr.request_changes",
+        ],
+    ),
+    "reconciler": (
+        ["gitea.read", "gitea.pr.close", "gitea.pr.comment", "gitea.branch.delete"],
+        [
+            "gitea.pr.create",
+            "gitea.branch.push",
+            "gitea.pr.approve",
+            "gitea.pr.merge",
+            "gitea.pr.request_changes",
+        ],
+    ),
+}
+
+ROLES = tuple(_ROLE_OPS)
+
+
+def _profile(role: str, *, canonical_root: str | None = None) -> dict:
+    allowed, forbidden = _ROLE_OPS[role]
+    profile = {
+        "enabled": True,
+        "context": "prgs",
+        "role": role,
+        "username": "jcwalker3",
+        "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}",
+        "allowed_repositories": [TARGET_SLUG],
+    }
+    if canonical_root is not None:
+        profile["canonical_repository_root"] = canonical_root
+    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"},
+            }
+        },
+        "profiles": profiles,
+    }
+
+
+class _CrossRepoHarness(unittest.TestCase):
+    """Real install + target git repos, temp profiles.json, no network."""
+
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self.tmp = self._dir.name
+        self.config_path = os.path.join(self.tmp, "profiles.json")
+
+        # The real install checkout carries a remote literally named `prgs`;
+        # a freshly cloned target repository normally names its remote `origin`.
+        # Reproducing that asymmetry is the point: identity derivation must not
+        # depend on the remote happening to share the `remote=` argument's name.
+        self.install_root = _init_repo(
+            Path(self.tmp) / "install", INSTALL_URL, remote_name="prgs"
+        )
+        self.target_root = _init_repo(
+            Path(self.tmp) / "target", TARGET_URL, remote_name="origin"
+        )
+        self.third_root = _init_repo(
+            Path(self.tmp) / "third", f"https://gitea.prgs.cc/{THIRD_SLUG}.git"
+        )
+        self.not_a_repo = os.path.join(self.tmp, "plain-dir")
+        os.makedirs(self.not_a_repo, exist_ok=True)
+
+        session_ctx._reset_session_context_for_testing()
+        gitea_config._active_profile_override = None
+        mcp_server._IDENTITY_CACHE.clear()
+        srv._MUTATION_AUTHORITY = None
+
+        # PROJECT_ROOT is wherever the server file physically lives; pin it to a
+        # real Gitea-Tools-identified checkout so "install-derived" is
+        # deterministic regardless of the developer's layout.
+        self._project_root = patch.object(srv, "PROJECT_ROOT", self.install_root)
+        self._project_root.start()
+        # Undo the autouse deterministic-remote stub for this module only.
+        self._real_remote_url = patch.object(
+            srv, "_local_git_remote_url", _REAL_LOCAL_GIT_REMOTE_URL
+        )
+        self._real_remote_url.start()
+
+    def tearDown(self):
+        self._real_remote_url.stop()
+        self._project_root.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 _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, role: str, **extra) -> dict:
+        env = {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": f"prgs-{role}",
+            "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 _bind(self, role: str, canonical_root: str | None):
+        """Activate *role* with *canonical_root* and return an env patch ctx."""
+        self._write_config(
+            {f"prgs-{role}": _profile(role, canonical_root=canonical_root)}
+        )
+        return patch.dict(os.environ, self._env(role), clear=True)
+
+
+# ===========================================================================
+# 1. The central helper (single source of root resolution)
+# ===========================================================================
+class TestCanonicalLocalGitRoot(_CrossRepoHarness):
+    """_canonical_local_git_root is the one place a target root is derived."""
+
+    def test_unconfigured_namespace_uses_installation_root(self):
+        with self._bind("author", None):
+            self.assertEqual(srv._canonical_local_git_root(), self.install_root)
+
+    def test_configured_namespace_uses_target_root(self):
+        with self._bind("author", self.target_root):
+            self.assertEqual(srv._canonical_local_git_root(), self.target_root)
+
+    def test_configured_root_resolves_to_git_toplevel(self):
+        nested = os.path.join(self.target_root, "branches", "wt")
+        os.makedirs(nested, exist_ok=True)
+        with self._bind("author", nested):
+            # A subdirectory of the target repo still resolves to its toplevel.
+            self.assertEqual(srv._canonical_local_git_root(), self.target_root)
+
+    def test_invalid_root_never_silently_becomes_installation_root(self):
+        """A configured-but-broken root must not fall back to Gitea-Tools."""
+        with self._bind("author", self.not_a_repo):
+            resolved = srv._canonical_local_git_root()
+            self.assertNotEqual(resolved, self.install_root)
+            self.assertEqual(resolved, os.path.realpath(self.not_a_repo))
+
+    def test_env_override_beats_profile_binding(self):
+        self._write_config(
+            {"prgs-author": _profile("author", canonical_root=self.third_root)}
+        )
+        env = self._env("author", GITEA_CANONICAL_REPOSITORY_ROOT=self.target_root)
+        with patch.dict(os.environ, env, clear=True):
+            self.assertEqual(srv._canonical_local_git_root(), self.target_root)
+
+
+# ===========================================================================
+# 2. Repository identity — the upstream inversion
+# ===========================================================================
+class TestRepositoryIdentityDerivation(_CrossRepoHarness):
+    """_local_git_remote_url and its consumers must read the target repo."""
+
+    def test_remote_url_reads_target_not_installation(self):
+        with self._bind("author", self.target_root):
+            self.assertEqual(srv._local_git_remote_url("origin"), TARGET_URL)
+
+    def test_remote_url_unconfigured_still_reads_installation(self):
+        with self._bind("author", None):
+            # The install checkout's remote is named `prgs`, matching production.
+            self.assertEqual(srv._local_git_remote_url("prgs"), INSTALL_URL)
+
+    def test_workspace_slug_follows_canonical_root(self):
+        with self._bind("author", self.target_root):
+            self.assertEqual(srv._workspace_repository_slug("origin"), TARGET_SLUG)
+
+    def test_workspace_slug_unconfigured_is_installation(self):
+        with self._bind("author", None):
+            self.assertEqual(srv._workspace_repository_slug("prgs"), INSTALL_SLUG)
+
+    def test_every_role_derives_the_same_target_identity(self):
+        for role in ROLES:
+            with self.subTest(role=role):
+                with self._bind(role, self.target_root):
+                    self.assertEqual(
+                        srv._workspace_repository_slug("origin"), TARGET_SLUG
+                    )
+
+
+# ===========================================================================
+# 3. _resolve — omitted and explicit coordinates
+# ===========================================================================
+class TestResolveTargetCoordinates(_CrossRepoHarness):
+    """Omitted coordinates follow the binding; explicit ones may only confirm."""
+
+    def test_omitted_coordinates_resolve_to_target_repository(self):
+        for role in ROLES:
+            with self.subTest(role=role):
+                with self._bind(role, self.target_root):
+                    _host, org, repo = srv._resolve("prgs", None, None, None)
+                    self.assertEqual((org, repo), (TARGET_ORG, TARGET_REPO))
+
+    def test_omitted_coordinates_unconfigured_resolve_to_installation(self):
+        with self._bind("author", None):
+            _host, org, repo = srv._resolve("prgs", None, None, None)
+            self.assertEqual((org, repo), (INSTALL_ORG, INSTALL_REPO))
+
+    def test_explicit_matching_coordinates_confirm_the_binding(self):
+        for role in ROLES:
+            with self.subTest(role=role):
+                with self._bind(role, self.target_root):
+                    _host, org, repo = srv._resolve(
+                        "prgs", None, TARGET_ORG, TARGET_REPO
+                    )
+                    self.assertEqual((org, repo), (TARGET_ORG, TARGET_REPO))
+
+    def test_explicit_mismatched_repository_cannot_override_binding(self):
+        for role in ROLES:
+            with self.subTest(role=role):
+                with self._bind(role, self.target_root):
+                    with self.assertRaises(RuntimeError) as ctx:
+                        srv._resolve("prgs", None, TARGET_ORG, INSTALL_REPO)
+                    msg = str(ctx.exception)
+                    self.assertIn("#741", msg)
+                    self.assertIn("fail closed", msg)
+
+    def test_explicit_mismatched_organization_cannot_override_binding(self):
+        with self._bind("author", self.target_root):
+            with self.assertRaises(RuntimeError):
+                srv._resolve("prgs", None, "SomeoneElse", TARGET_REPO)
+
+    def test_gitea_tools_rooted_namespace_cannot_target_control_plane(self):
+        """Direction 1: an install-rooted namespace must not reach the target."""
+        with self._bind("author", self.install_root):
+            with self.assertRaises(RuntimeError):
+                srv._resolve("prgs", None, TARGET_ORG, TARGET_REPO)
+
+    def test_control_plane_rooted_namespace_cannot_target_gitea_tools(self):
+        """Direction 2: a target-rooted namespace must not reach Gitea-Tools."""
+        with self._bind("author", self.target_root):
+            with self.assertRaises(RuntimeError):
+                srv._resolve("prgs", None, INSTALL_ORG, INSTALL_REPO)
+
+    def test_unconfigured_namespace_keeps_existing_explicit_behaviour(self):
+        """Same-repository behaviour is unchanged (no new fail-closed path)."""
+        with self._bind("author", None):
+            _host, org, repo = srv._resolve("prgs", None, INSTALL_ORG, INSTALL_REPO)
+            self.assertEqual((org, repo), (INSTALL_ORG, INSTALL_REPO))
+
+
+# ===========================================================================
+# 4. #274 workspace guard — the two guard paths must agree
+# ===========================================================================
+class TestMutationWorkspaceGuardBinding(_CrossRepoHarness):
+    """Both guard paths must agree on which repository they protect."""
+
+    def _contexts(self, worktree: str | None = None):
+        return srv._resolve_namespace_mutation_context(worktree)
+
+    def test_mutation_context_uses_target_root(self):
+        with self._bind("author", self.target_root):
+            self.assertEqual(self._contexts()["canonical_repo_root"], self.target_root)
+
+    def test_mutation_context_unconfigured_uses_installation_root(self):
+        with self._bind("author", None):
+            self.assertEqual(self._contexts()["canonical_repo_root"], self.install_root)
+
+    def test_role_mutation_workspace_guard_agrees_with_mutation_context(self):
+        """The omitted configured_canonical_root made these two disagree."""
+        import namespace_workspace_binding as nwb
+
+        for role in ROLES:
+            with self.subTest(role=role):
+                with self._bind(role, self.target_root):
+                    configured, _src = srv._configured_canonical_root()
+                    assessment = nwb.assess_namespace_mutation_workspace(
+                        role_kind=role,
+                        worktree_path=None,
+                        worktree=None,
+                        process_project_root=srv.PROJECT_ROOT,
+                        profile_name=f"prgs-{role}",
+                        configured_canonical_root=configured,
+                    )
+                    self.assertEqual(
+                        assessment["canonical_repo_root"],
+                        self._contexts()["canonical_repo_root"],
+                    )
+                    self.assertEqual(
+                        assessment["canonical_repo_root"], self.target_root
+                    )
+
+    def test_wrong_worktree_is_rejected_against_target_root(self):
+        """A worktree belonging to the install repo is not in the target repo."""
+        import author_mutation_worktree as amw
+
+        with self._bind("author", self.target_root):
+            foreign = os.path.join(self.install_root, "branches", "wt")
+            os.makedirs(foreign, exist_ok=True)
+            membership = amw.assess_workspace_repo_membership(
+                workspace_path=foreign,
+                canonical_repo_root=self.target_root,
+            )
+            self.assertTrue(membership["block"])
+
+
+# ===========================================================================
+# 5. Canonical-root validation — missing / invalid / mismatched
+# ===========================================================================
+class TestCanonicalRootFailClosed(_CrossRepoHarness):
+    """Missing, invalid, ambiguous and mismatched roots fail closed."""
+
+    def _assess(self, value, *, expected=TARGET_SLUG, require=True):
+        return crr.assess_canonical_repository_root(
+            configured_value=value,
+            source="test",
+            expected_slug=expected,
+            process_project_root=self.install_root,
+            remote="origin",
+            require_binding=require,
+        )
+
+    def test_missing_root_fails_closed_when_binding_required(self):
+        result = self._assess(None)
+        self.assertTrue(result["block"])
+        self.assertFalse(result["configured"])
+
+    def test_missing_root_is_the_single_repo_default_when_not_required(self):
+        result = self._assess(None, expected=None, require=False)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["canonical_repo_root"], self.install_root)
+
+    def test_nonexistent_root_fails_closed(self):
+        result = self._assess(os.path.join(self.tmp, "nope"))
+        self.assertTrue(result["block"])
+        self.assertIn("does not exist", " ".join(result["reasons"]))
+
+    def test_non_git_directory_fails_closed(self):
+        result = self._assess(self.not_a_repo)
+        self.assertTrue(result["block"])
+        self.assertIn("not a git repository", " ".join(result["reasons"]))
+
+    def test_mismatched_repository_identity_fails_closed(self):
+        result = self._assess(self.third_root)
+        self.assertTrue(result["block"])
+        self.assertIn("mismatch", " ".join(result["reasons"]))
+
+    def test_matching_repository_identity_is_proven(self):
+        result = self._assess(self.target_root)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["resolved_slug"], TARGET_SLUG)
+        self.assertEqual(result["canonical_repo_root"], self.target_root)
+
+    def test_unresolvable_root_yields_fail_closed_reasons_not_fallback(self):
+        with self._bind("author", self.not_a_repo):
+            slug, reasons = srv._canonical_repository_slug(srv.get_profile(), "origin")
+            self.assertIsNone(slug)
+            self.assertTrue(reasons)
+
+
+# ===========================================================================
+# 6. Immutability — first bind wins, requests cannot replace the root
+# ===========================================================================
+class TestCanonicalRootImmutability(_CrossRepoHarness):
+    """No request-supplied value can establish or swap the pinned root."""
+
+    def test_first_bind_pins_target_root(self):
+        with self._bind("author", self.target_root):
+            with patch(
+                "gitea_mcp_server.api_request",
+                side_effect=lambda *a, **k: {
+                    "login": "jcwalker3",
+                    "full_name": "T",
+                    "id": 1,
+                    "email": "[email protected]",
+                },
+            ):
+                srv.gitea_whoami(remote="prgs")
+            bound = session_ctx.get_session_context()
+            self.assertEqual(bound["canonical_repository_root"], self.target_root)
+
+    def test_binding_is_first_write_wins(self):
+        with self._bind("author", self.target_root):
+            session_ctx.seed_session_context_if_unbound(
+                profile_name="prgs-author",
+                remote="prgs",
+                host="gitea.prgs.cc",
+                identity="jcwalker3",
+                org=TARGET_ORG,
+                repository=TARGET_REPO,
+                role_kind="author",
+                source="test",
+                canonical_repository_root=self.target_root,
+            )
+            # A second, contradictory seed must not replace the pin.
+            session_ctx.seed_session_context_if_unbound(
+                profile_name="prgs-author",
+                remote="prgs",
+                host="gitea.prgs.cc",
+                identity="jcwalker3",
+                org=INSTALL_ORG,
+                repository=INSTALL_REPO,
+                role_kind="author",
+                source="test-2",
+                canonical_repository_root=self.install_root,
+            )
+            bound = session_ctx.get_session_context()
+            self.assertEqual(bound["canonical_repository_root"], self.target_root)
+            self.assertEqual(bound["repository"], TARGET_REPO)
+
+    def test_request_supplied_worktree_cannot_replace_the_root(self):
+        with self._bind("author", self.target_root):
+            ctx = srv._resolve_namespace_mutation_context(self.install_root)
+            # The workspace argument may be demoted/inspected, but the canonical
+            # repository root stays the configured target.
+            self.assertEqual(ctx["canonical_repo_root"], self.target_root)
+
+    def test_request_supplied_coordinates_cannot_replace_the_root(self):
+        with self._bind("author", self.target_root):
+            with self.assertRaises(RuntimeError):
+                srv._resolve("prgs", None, INSTALL_ORG, INSTALL_REPO)
+            self.assertEqual(srv._canonical_local_git_root(), self.target_root)
+
+
+# ===========================================================================
+# 7. Parity dimensions stay separately labelled (#739 F3 preserved)
+# ===========================================================================
+class TestParityDimensionSeparation(_CrossRepoHarness):
+    """Server-implementation parity stays anchored to the install checkout."""
+
+    def test_server_dimension_is_installation_not_target(self):
+        import master_parity_gate
+
+        with self._bind("author", self.target_root):
+            head = master_parity_gate.read_git_head(srv.PROJECT_ROOT)
+            self.assertEqual(head, _git(self.install_root, "rev-parse", "HEAD"))
+            self.assertNotEqual(head, _git(self.target_root, "rev-parse", "HEAD"))
+
+    def test_target_dimension_reads_the_target_checkout(self):
+        with self._bind("author", self.target_root):
+            self.assertEqual(
+                _git(srv._canonical_local_git_root(), "rev-parse", "HEAD"),
+                _git(self.target_root, "rev-parse", "HEAD"),
+            )
+
+
+# ===========================================================================
+# 8. Configuration loaders validate the binding consistently (AC13)
+# ===========================================================================
+class TestConfigurationLoaderValidation(unittest.TestCase):
+    """Every supported loader treats canonical_repository_root identically."""
+
+    def setUp(self):
+        self._dir = tempfile.TemporaryDirectory()
+        self.tmp = self._dir.name
+
+    def tearDown(self):
+        self._dir.cleanup()
+
+    def _write(self, data: dict) -> str:
+        path = os.path.join(self.tmp, "profiles.json")
+        with open(path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(data))
+        return path
+
+    # --- v1 ---------------------------------------------------------------
+    def _v1(self, root):
+        return {
+            "version": 1,
+            "profiles": {
+                "prgs-author": {
+                    "base_url": "https://gitea.prgs.cc",
+                    "auth": {"type": "env", "name": "GITEA_TOKEN"},
+                    "canonical_repository_root": root,
+                }
+            },
+        }
+
+    def test_v1_rejects_relative_canonical_root(self):
+        path = self._write(self._v1("relative/path"))
+        with self.assertRaises(gitea_config.ConfigError) as ctx:
+            gitea_config.load_config(path)
+        self.assertIn("absolute", str(ctx.exception))
+
+    def test_v1_rejects_blank_canonical_root(self):
+        path = self._write(self._v1("   "))
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config.load_config(path)
+
+    def test_v1_accepts_absolute_canonical_root(self):
+        path = self._write(self._v1("/abs/target"))
+        loaded = gitea_config.load_config(path)
+        self.assertEqual(
+            loaded["profiles"]["prgs-author"]["canonical_repository_root"],
+            "/abs/target",
+        )
+
+    def test_v1_without_the_field_is_unchanged(self):
+        data = self._v1("/abs/target")
+        del data["profiles"]["prgs-author"]["canonical_repository_root"]
+        loaded = gitea_config.load_config(self._write(data))
+        self.assertNotIn("canonical_repository_root", loaded["profiles"]["prgs-author"])
+
+    # --- v2 environments --------------------------------------------------
+    def _v2_env(self, root):
+        ident = {
+            "auth": {"type": "env", "name": "GITEA_TOKEN"},
+            "base_url": "https://gitea.prgs.cc",
+            "username": "jcwalker3",
+            "allowed_operations": ["gitea.read"],
+        }
+        if root is not None:
+            ident["canonical_repository_root"] = root
+        return {
+            "version": 2,
+            "environments": {
+                "prgs": {"services": {"gitea": {"identities": {"author": ident}}}}
+            },
+        }
+
+    def test_v2_environments_propagates_canonical_root(self):
+        """Regression: the field was silently dropped during flattening."""
+        loaded = gitea_config.load_config(self._write(self._v2_env("/abs/target")))
+        profile = loaded["profiles"]["prgs.gitea.author"]
+        self.assertEqual(profile["canonical_repository_root"], "/abs/target")
+
+    def test_v2_environments_rejects_relative_canonical_root(self):
+        with self.assertRaises(gitea_config.ConfigError) as ctx:
+            gitea_config.load_config(self._write(self._v2_env("relative/path")))
+        self.assertIn("absolute", str(ctx.exception))
+
+    def test_v2_environments_without_the_field_is_unchanged(self):
+        loaded = gitea_config.load_config(self._write(self._v2_env(None)))
+        self.assertNotIn(
+            "canonical_repository_root", loaded["profiles"]["prgs.gitea.author"]
+        )
+
+    # --- v2 contexts (already validated; guard against regression) ---------
+    def test_v2_contexts_still_rejects_relative_canonical_root(self):
+        data = {
+            "version": 2,
+            "contexts": {
+                "prgs": {
+                    "enabled": True,
+                    "gitea": {"enabled": True, "base_url": "https://gitea.prgs.cc"},
+                }
+            },
+            "profiles": {
+                "prgs-author": {
+                    "enabled": True,
+                    "context": "prgs",
+                    "username": "jcwalker3",
+                    "auth": {"type": "env", "name": "GITEA_TOKEN"},
+                    "allowed_operations": ["gitea.read"],
+                    "canonical_repository_root": "relative/path",
+                }
+            },
+        }
+        with self.assertRaises(gitea_config.ConfigError):
+            gitea_config.load_config(self._write(data))
+
+
+# ===========================================================================
+# 9. Role-by-operation matrix
+# ===========================================================================
+class TestRoleOperationMatrix(_CrossRepoHarness):
+    """Each role, each cross-repository condition, one assertion per cell."""
+
+    def test_correct_target_resolves_for_every_role(self):
+        for role in ROLES:
+            with self.subTest(role=role, case="correct-target"):
+                with self._bind(role, self.target_root):
+                    _h, org, repo = srv._resolve("prgs", None, None, None)
+                    self.assertEqual((org, repo), (TARGET_ORG, TARGET_REPO))
+
+    def test_wrong_repository_is_rejected_for_every_role(self):
+        for role in ROLES:
+            with self.subTest(role=role, case="wrong-repo"):
+                with self._bind(role, self.target_root):
+                    with self.assertRaises(RuntimeError):
+                        srv._resolve("prgs", None, INSTALL_ORG, INSTALL_REPO)
+
+    def test_explicit_matching_coordinates_pass_for_every_role(self):
+        for role in ROLES:
+            with self.subTest(role=role, case="explicit-match"):
+                with self._bind(role, self.target_root):
+                    _h, org, repo = srv._resolve("prgs", None, TARGET_ORG, TARGET_REPO)
+                    self.assertEqual((org, repo), (TARGET_ORG, TARGET_REPO))
+
+    def test_explicit_mismatch_fails_closed_for_every_role(self):
+        for role in ROLES:
+            with self.subTest(role=role, case="explicit-mismatch"):
+                with self._bind(role, self.target_root):
+                    with self.assertRaises(RuntimeError):
+                        srv._resolve("prgs", None, INSTALL_ORG, THIRD_REPO)
+
+    def test_missing_canonical_root_preserves_single_repo_default(self):
+        for role in ROLES:
+            with self.subTest(role=role, case="missing-root"):
+                with self._bind(role, None):
+                    self.assertEqual(srv._canonical_local_git_root(), self.install_root)
+
+    def test_invalid_canonical_root_never_becomes_install_root(self):
+        for role in ROLES:
+            with self.subTest(role=role, case="invalid-root"):
+                with self._bind(role, self.not_a_repo):
+                    self.assertNotEqual(
+                        srv._canonical_local_git_root(), self.install_root
+                    )
+
+    def test_wrong_worktree_rejected_for_every_role(self):
+        import author_mutation_worktree as amw
+
+        foreign = os.path.join(self.install_root, "branches", "foreign")
+        os.makedirs(foreign, exist_ok=True)
+        for role in ROLES:
+            with self.subTest(role=role, case="wrong-worktree"):
+                with self._bind(role, self.target_root):
+                    membership = amw.assess_workspace_repo_membership(
+                        workspace_path=foreign,
+                        canonical_repo_root=srv._canonical_local_git_root(),
+                    )
+                    self.assertTrue(membership["block"])
+
+    def test_request_override_attempt_rejected_for_every_role(self):
+        for role in ROLES:
+            with self.subTest(role=role, case="request-override"):
+                with self._bind(role, self.target_root):
+                    with self.assertRaises(RuntimeError):
+                        srv._resolve("prgs", None, "Attacker", "evil-repo")
+                    self.assertEqual(srv._canonical_local_git_root(), self.target_root)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_747_sliding_lease_ttl.py

@@ -0,0 +1,210 @@
+"""Tests for the 10-minute sliding TTL on reviewer and merger PR leases (#747).
+
+The lease ledger previously minted a fixed 120-minute expiry and derived
+staleness from separate 30/60-minute activity bands. A dead session therefore
+held a PR for up to two hours. These tests pin the sliding-window contract:
+acquisition mints a 10-minute expiry, every heartbeat slides it forward, and an
+expired lease is immediately reclaimable with no intermediate waiting tier.
+"""
+
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import reviewer_pr_lease as leases
+
+
+def _body(
+    *,
+    session_id: str = "session-a",
+    pr_number: int = 747,
+    phase: str = "claimed",
+    last_activity: datetime | None = None,
+    expires_at: datetime | None = None,
+    ttl_minutes: int | None = None,
+) -> str:
+    kwargs = {}
+    if ttl_minutes is not None:
+        kwargs["ttl_minutes"] = ttl_minutes
+    return leases.format_lease_body(
+        repo="Scaled-Tech-Consulting/Gitea-Tools",
+        pr_number=pr_number,
+        issue_number=747,
+        reviewer_identity="rev1",
+        profile="prgs-reviewer",
+        session_id=session_id,
+        worktree="branches/review-pr747",
+        phase=phase,
+        candidate_head="a" * 40,
+        target_branch="master",
+        target_branch_sha="b" * 40,
+        last_activity=last_activity,
+        expires_at=expires_at,
+        **kwargs,
+    )
+
+
+def _comment(**kwargs) -> dict:
+    return {"id": 1, "body": _body(**kwargs), "user": {"login": "rev1"}}
+
+
+def _minutes_ago(minutes: int) -> datetime:
+    return datetime.now(timezone.utc) - timedelta(minutes=minutes)
+
+
+class TestSlidingTTLConstant(unittest.TestCase):
+    """AC6: one named constant per lease kind, no duplicated literals."""
+
+    def test_ttl_is_ten_minutes(self):
+        self.assertEqual(leases.LEASE_TTL_MINUTES, 10)
+
+    def test_renewal_window_is_separately_named(self):
+        self.assertEqual(leases.LEASE_RENEWAL_MINUTES, 10)
+
+
+class TestAcquisitionTTL(unittest.TestCase):
+    """AC1 / AC2: reviewer and merger acquisition both mint now + 10 minutes."""
+
+    def test_acquire_mints_ten_minute_expiry(self):
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(last_activity=now))
+        expires = leases._parse_timestamp(lease["expires_at"])
+        self.assertEqual(expires, now + timedelta(minutes=10))
+
+    def test_merger_acquisition_shares_the_same_window(self):
+        # Merger acquisition funnels through the same lease-body formatter, so
+        # the reviewer TTL is the merger TTL by construction.
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(phase="merging", last_activity=now))
+        expires = leases._parse_timestamp(lease["expires_at"])
+        self.assertEqual(expires, now + timedelta(minutes=10))
+
+
+class TestHeartbeatSlides(unittest.TestCase):
+    """AC3: a heartbeat slides expires_at to now + 10 minutes."""
+
+    def test_heartbeat_slides_expiry_forward(self):
+        acquired = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        beat = acquired + timedelta(minutes=7)
+        first = leases.parse_lease_comment(_body(last_activity=acquired))
+        renewed = leases.parse_lease_comment(_body(last_activity=beat))
+
+        first_expiry = leases._parse_timestamp(first["expires_at"])
+        renewed_expiry = leases._parse_timestamp(renewed["expires_at"])
+
+        self.assertEqual(renewed_expiry, beat + timedelta(minutes=10))
+        self.assertGreater(renewed_expiry, first_expiry)
+
+    def test_renewal_window_is_independently_tunable(self):
+        # The renewal amount must not be hardwired to the acquisition TTL;
+        # format_lease_body accepts an explicit window.
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(last_activity=now, ttl_minutes=3))
+        expires = leases._parse_timestamp(lease["expires_at"])
+        self.assertEqual(expires, now + timedelta(minutes=3))
+
+
+class TestFreshnessBands(unittest.TestCase):
+    """AC5: expiry is the only gate; no intermediate reclaim tier."""
+
+    def test_fresh_lease_is_active(self):
+        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(1)))
+        self.assertEqual(leases.classify_lease_freshness(lease), "active")
+
+    def test_idle_past_half_ttl_warns_before_expiry(self):
+        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(6)))
+        self.assertEqual(leases.classify_lease_freshness(lease), "stale_warning")
+
+    def test_lease_expires_after_ten_idle_minutes(self):
+        lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(11)))
+        self.assertEqual(leases.classify_lease_freshness(lease), "expired")
+
+    def test_no_separate_reclaimable_tier_remains(self):
+        # The old 60-minute reclaim band sat between "stale" and "expired" and
+        # blocked acquisition. Under a sliding TTL an idle lease is already
+        # expired, so the tier must not reappear at any idle duration.
+        for minutes in (11, 30, 65, 121, 600):
+            lease = leases.parse_lease_comment(_body(last_activity=_minutes_ago(minutes)))
+            self.assertEqual(
+                leases.classify_lease_freshness(lease),
+                "expired",
+                f"idle {minutes}m should be expired, not a waiting tier",
+            )
+
+
+class TestExpiredLeaseIsImmediatelyReclaimable(unittest.TestCase):
+    """AC5: another session takes over an expired lease with no extra wait."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def _acquire_against(self, comments: list[dict]) -> dict:
+        return leases.assess_acquire_lease(
+            comments,
+            pr_number=747,
+            reviewer_identity="rev2",
+            profile="prgs-reviewer",
+            session_id="session-b",
+            repo="Scaled-Tech-Consulting/Gitea-Tools",
+            issue_number=747,
+            worktree="branches/review-pr747-b",
+            candidate_head="c" * 40,
+            target_branch="master",
+            target_branch_sha="d" * 40,
+        )
+
+    def test_expired_foreign_lease_does_not_block_acquisition(self):
+        comments = [_comment(session_id="dead-session", last_activity=_minutes_ago(11))]
+        result = self._acquire_against(comments)
+        self.assertTrue(result["acquire_allowed"], result["reasons"])
+
+    def test_live_foreign_lease_still_blocks_acquisition(self):
+        comments = [_comment(session_id="live-session", last_activity=_minutes_ago(2))]
+        result = self._acquire_against(comments)
+        self.assertFalse(result["acquire_allowed"])
+        self.assertTrue(any("already has active" in r for r in result["reasons"]))
+
+
+class TestRemainingTimeReporting(unittest.TestCase):
+    """AC7: diagnostics can distinguish 'held and live' from 'held and dying'."""
+
+    def test_seconds_remaining_on_live_lease(self):
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(last_activity=now))
+        remaining = leases.lease_seconds_remaining(lease, now=now + timedelta(minutes=4))
+        self.assertEqual(remaining, 360)
+
+    def test_seconds_remaining_is_zero_when_expired(self):
+        now = datetime(2026, 7, 18, 12, 0, 0, tzinfo=timezone.utc)
+        lease = leases.parse_lease_comment(_body(last_activity=now))
+        remaining = leases.lease_seconds_remaining(lease, now=now + timedelta(minutes=30))
+        self.assertEqual(remaining, 0)
+
+    def test_seconds_remaining_is_none_without_parsable_expiry(self):
+        self.assertIsNone(leases.lease_seconds_remaining({"expires_at": "not-a-time"}))
+
+
+class TestLegacyLeaseRows(unittest.TestCase):
+    """AC9: leases minted under the old 120-minute TTL still evaluate."""
+
+    def test_legacy_two_hour_expiry_is_honoured_until_it_passes(self):
+        now = datetime.now(timezone.utc)
+        legacy = leases.parse_lease_comment(
+            _body(last_activity=now - timedelta(minutes=90), expires_at=now + timedelta(minutes=30))
+        )
+        # Still inside its originally minted window: not expired, but idle long
+        # enough to warn.
+        self.assertEqual(leases.classify_lease_freshness(legacy), "stale_warning")
+
+    def test_legacy_row_past_its_own_expiry_is_expired(self):
+        now = datetime.now(timezone.utc)
+        legacy = leases.parse_lease_comment(
+            _body(last_activity=now - timedelta(minutes=180), expires_at=now - timedelta(minutes=60))
+        )
+        self.assertEqual(leases.classify_lease_freshness(legacy), "expired")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_merger_lease_finalization.py

@@ -0,0 +1,956 @@
+import sys as _sys
+from pathlib import Path as _Path
+_sys.path.insert(0, str(_Path(__file__).resolve().parent))
+from mutation_profile_fixture import shared_mutation_env  # noqa: E402,F401
+"""Acquired merger-lease provenance + owner finalization (#742).
+
+Covers the two confirmed defects behind the PR #740 failure:
+
+1. a lease minted by ``gitea_acquire_merger_pr_lease`` was listed as a
+   sanctioned provenance source but rejected by ``is_sanctioned_session_lease``
+   and reported as unsanctioned by ``describe_session_lease_proof``, so the
+   merge gate refused the merger's own freshly acquired lease;
+2. no merger-role owner-session operation existed to terminally release or
+   abandon that lease, leaving natural expiry as the only exit.
+"""
+
+import os
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import gitea_mcp_server as mcp_server  # noqa: E402
+import merger_lease_adoption as mla  # noqa: E402
+import pr_work_lease as pwl  # noqa: E402
+import reviewer_pr_lease as leases  # noqa: E402
+import task_capability_map as tcm  # noqa: E402
+
+HEAD = "c" * 40
+OTHER_HEAD = "d" * 40
+REPO = "Scaled-Tech-Consulting/Gitea-Tools"
+PR = 740
+
+
+def _lease_comment(
+    pr_number: int = PR,
+    session_id: str = "merger-session",
+    *,
+    phase: str = "claimed",
+    profile: str = "prgs-merger",
+    identity: str = "merger-user",
+    candidate_head: str = HEAD,
+    comment_id: int = 12354,
+) -> dict:
+    body = leases.format_lease_body(
+        repo=REPO,
+        pr_number=pr_number,
+        issue_number=742,
+        reviewer_identity=identity,
+        profile=profile,
+        session_id=session_id,
+        worktree="branches/merger-pr740",
+        phase=phase,
+        candidate_head=candidate_head,
+        target_branch="master",
+        target_branch_sha="e" * 40,
+        last_activity=datetime.now(timezone.utc),
+    )
+    return {"id": comment_id, "body": body, "user": {"login": identity}}
+
+
+def _acquired_session(**overrides) -> dict:
+    session = {
+        "pr_number": PR,
+        "issue_number": 742,
+        "session_id": "merger-session",
+        "reviewer_identity": "merger-user",
+        "profile": "prgs-merger",
+        "worktree": "branches/merger-pr740",
+        "phase": "claimed",
+        "candidate_head": HEAD,
+        "target_branch": "master",
+        "target_branch_sha": "e" * 40,
+        "repo": REPO,
+        "comment_id": 12354,
+    }
+    session.update(overrides)
+    return session
+
+
+def _record_acquired(session: dict | None = None, **provenance_kwargs) -> dict:
+    provenance = mla.build_lease_provenance(
+        source=mla.SOURCE_ACQUIRE_MERGER,
+        comment_id=provenance_kwargs.pop("comment_id", 12354),
+        **provenance_kwargs,
+    )
+    return leases.record_session_lease(
+        session if session is not None else _acquired_session(),
+        lease_provenance=provenance,
+    )
+
+
+class TestAcquiredMergerProvenance(unittest.TestCase):
+    """AC1/AC2/AC4/AC5/AC6 — provenance sanctioning and proof description."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def test_acquired_merger_lease_is_sanctioned(self):
+        session = _record_acquired()
+        self.assertTrue(mla.is_sanctioned_session_lease(session))
+        self.assertEqual(mla.assess_acquired_merger_lease_integrity(session), [])
+
+    def test_proof_description_is_explicit_acquired_merger_kind(self):
+        session = _record_acquired()
+        proof = mla.describe_session_lease_proof(session)
+        self.assertEqual(proof["lease_proof_source"], mla.SOURCE_ACQUIRE_MERGER)
+        self.assertEqual(proof["lease_proof_kind"], "sanctioned_acquire_merger")
+        self.assertTrue(proof["lease_proof_sanctioned"])
+        self.assertEqual(proof["lease_proof_comment_id"], 12354)
+
+    def test_manual_session_seed_without_provenance_still_rejected(self):
+        session = leases.record_session_lease(_acquired_session())
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+        proof = mla.describe_session_lease_proof(session)
+        self.assertEqual(proof["lease_proof_kind"], "unsanctioned_manual_seed")
+
+    def test_unknown_provenance_still_rejected(self):
+        session = leases.record_session_lease(
+            _acquired_session(),
+            lease_provenance={"source": "totally_made_up_tool", "comment_id": 1},
+        )
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+        self.assertEqual(
+            mla.describe_session_lease_proof(session)["lease_proof_kind"],
+            "unsanctioned",
+        )
+
+    def test_forged_acquired_provenance_without_comment_marker_rejected(self):
+        session = leases.record_session_lease(
+            _acquired_session(comment_id=None),
+            lease_provenance={"source": mla.SOURCE_ACQUIRE_MERGER},
+        )
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+        self.assertTrue(
+            any(
+                "comment marker" in reason
+                for reason in mla.assess_acquired_merger_lease_integrity(session)
+            )
+        )
+
+    def test_comment_id_mismatch_between_provenance_and_session_rejected(self):
+        session = _record_acquired(_acquired_session(comment_id=999))
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+
+    def test_reviewer_profile_lease_is_not_acquired_merger_proof(self):
+        session = _record_acquired(_acquired_session(profile="prgs-reviewer"))
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+        self.assertTrue(
+            any(
+                "not a merger profile" in reason
+                for reason in mla.assess_acquired_merger_lease_integrity(session)
+            )
+        )
+
+    def test_incomplete_fields_fail_closed(self):
+        for field, value in (
+            ("session_id", ""),
+            ("reviewer_identity", ""),
+            ("profile", ""),
+            ("repo", ""),
+            ("pr_number", None),
+            ("candidate_head", None),
+            ("candidate_head", "not-a-sha"),
+        ):
+            with self.subTest(field=field, value=value):
+                leases.clear_session_lease()
+                session = _record_acquired(_acquired_session(**{field: value}))
+                self.assertFalse(
+                    mla.is_sanctioned_session_lease(session),
+                    f"{field}={value!r} must not be sanctioned",
+                )
+
+    def test_existing_reviewer_acquire_and_heartbeat_kinds_unchanged(self):
+        for source, kind in (
+            (mla.SOURCE_ACQUIRE, "sanctioned_acquire"),
+            (mla.SOURCE_HEARTBEAT, "sanctioned_heartbeat"),
+        ):
+            with self.subTest(source=source):
+                leases.clear_session_lease()
+                session = leases.record_session_lease(
+                    {
+                        "pr_number": PR,
+                        "session_id": "reviewer-session",
+                        "candidate_head": HEAD,
+                        "comment_id": 101,
+                    },
+                    lease_provenance=mla.build_lease_provenance(
+                        source=source, comment_id=101
+                    ),
+                )
+                self.assertTrue(mla.is_sanctioned_session_lease(session))
+                self.assertEqual(
+                    mla.describe_session_lease_proof(session)["lease_proof_kind"],
+                    kind,
+                )
+
+    def test_adoption_provenance_unchanged(self):
+        session = leases.record_session_lease(
+            {
+                "pr_number": PR,
+                "session_id": "merger-session",
+                "candidate_head": HEAD,
+                "comment_id": 202,
+            },
+            lease_provenance=mla.build_lease_provenance(
+                source=mla.SOURCE_ADOPT,
+                comment_id=202,
+                adopted_from_session_id="reviewer-session",
+            ),
+        )
+        self.assertTrue(mla.is_sanctioned_session_lease(session))
+        self.assertEqual(
+            mla.describe_session_lease_proof(session)["lease_proof_kind"],
+            "sanctioned_adoption",
+        )
+
+    def test_adoption_without_source_session_still_rejected(self):
+        session = leases.record_session_lease(
+            {"pr_number": PR, "session_id": "merger-session", "comment_id": 202},
+            lease_provenance=mla.build_lease_provenance(
+                source=mla.SOURCE_ADOPT, comment_id=202
+            ),
+        )
+        self.assertFalse(mla.is_sanctioned_session_lease(session))
+
+
+class TestMergeAuthorizationGate(unittest.TestCase):
+    """AC3 — the merge gate accepts a valid freshly acquired merger lease."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def test_acquired_merger_lease_passes_mutation_lease_gate(self):
+        comments = [_lease_comment()]
+        _record_acquired()
+        result = leases.assess_mutation_lease_gate(
+            pr_number=PR,
+            comments=comments,
+            reviewer_identity="merger-user",
+            session_id="merger-session",
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertFalse(result["block"], result["reasons"])
+
+    def test_manual_seed_still_blocked_by_mutation_lease_gate(self):
+        comments = [_lease_comment()]
+        leases.record_session_lease(_acquired_session())
+        result = leases.assess_mutation_lease_gate(
+            pr_number=PR,
+            comments=comments,
+            reviewer_identity="merger-user",
+            session_id="merger-session",
+            mutation="merge",
+            live_head_sha=HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("sanctioned provenance" in r for r in result["reasons"])
+        )
+
+    def test_head_mismatch_still_blocked_with_acquired_lease(self):
+        comments = [_lease_comment()]
+        _record_acquired()
+        result = leases.assess_mutation_lease_gate(
+            pr_number=PR,
+            comments=comments,
+            reviewer_identity="merger-user",
+            session_id="merger-session",
+            mutation="merge",
+            live_head_sha=OTHER_HEAD,
+            pinned_head_sha=HEAD,
+        )
+        self.assertTrue(result["block"])
+
+
+class TestFinalizationAssessment(unittest.TestCase):
+    """AC7/AC8/AC9/AC10 — owner-only, append-only, idempotent finalization."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def _assess(self, comments, session, **overrides):
+        kwargs = {
+            "pr_number": PR,
+            "session": session,
+            "actor_identity": "merger-user",
+            "actor_profile": "prgs-merger",
+            "actor_session_id": "merger-session",
+            "repo": REPO,
+            "worktree": "branches/merger-pr740",
+            "candidate_head": HEAD,
+            "live_head_sha": HEAD,
+        }
+        kwargs.update(overrides)
+        return mla.assess_merger_lease_finalization(comments, **kwargs)
+
+    def test_owner_merger_may_release_its_own_acquired_lease(self):
+        session = _record_acquired()
+        result = self._assess([_lease_comment()], session)
+        self.assertTrue(result["finalize_allowed"], result["reasons"])
+        self.assertFalse(result["already_terminal"])
+        self.assertIn(mla.MERGER_FINALIZATION_MARKER, result["finalization_body"])
+        self.assertIn("phase: released", result["finalization_body"])
+
+    def test_abandon_outcome_supported(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()],
+            session,
+            outcome=mla.OUTCOME_ABANDONED,
+            reason="non-retryable-merge-failure",
+        )
+        self.assertTrue(result["finalize_allowed"], result["reasons"])
+        self.assertIn("phase: abandoned", result["finalization_body"])
+        self.assertIn(
+            "finalization_reason: non-retryable-merge-failure",
+            result["finalization_body"],
+        )
+
+    def test_unknown_outcome_rejected(self):
+        session = _record_acquired()
+        result = self._assess([_lease_comment()], session, outcome="deleted")
+        self.assertFalse(result["finalize_allowed"])
+
+    def test_foreign_session_release_rejected(self):
+        session = _record_acquired()
+        comments = [_lease_comment(session_id="someone-else")]
+        result = self._assess(comments, session, actor_session_id="someone-else")
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(
+            any("does not match" in r or "owned by" in r for r in result["reasons"])
+        )
+
+    def test_active_foreign_lease_on_thread_rejected(self):
+        session = _record_acquired()
+        comments = [
+            _lease_comment(),
+            _lease_comment(session_id="other-merger", comment_id=12400),
+        ]
+        result = self._assess(comments, session)
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("owned by session_id" in r for r in result["reasons"]))
+
+    def test_reviewer_profile_cannot_finalize_merger_lease(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()], session, actor_profile="prgs-reviewer"
+        )
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(
+            any("not a merger profile" in r for r in result["reasons"])
+        )
+
+    def test_reviewer_sourced_session_lease_is_not_merger_owned(self):
+        session = leases.record_session_lease(
+            _acquired_session(profile="prgs-reviewer"),
+            lease_provenance=mla.build_lease_provenance(
+                source=mla.SOURCE_ACQUIRE, comment_id=12354
+            ),
+        )
+        result = self._assess([_lease_comment(profile="prgs-reviewer")], session)
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(
+            any("merger-owned" in r for r in result["reasons"]), result["reasons"]
+        )
+
+    def test_no_session_lease_rejected(self):
+        result = self._assess([_lease_comment()], None)
+        self.assertFalse(result["finalize_allowed"])
+
+    def test_wrong_pr_rejected(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment(pr_number=741, session_id="merger-session")],
+            session,
+            pr_number=741,
+        )
+        self.assertFalse(result["finalize_allowed"])
+
+    def test_wrong_repository_rejected(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()], session, repo="Scaled-Tech-Consulting/Other"
+        )
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("repository" in r for r in result["reasons"]))
+
+    def test_wrong_head_rejected(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()], session, candidate_head=OTHER_HEAD
+        )
+        self.assertFalse(result["finalize_allowed"])
+
+    def test_wrong_identity_rejected(self):
+        session = _record_acquired()
+        result = self._assess(
+            [_lease_comment()], session, actor_identity="someone-else"
+        )
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("identity" in r for r in result["reasons"]))
+
+    def test_token_fingerprint_mismatch_rejected(self):
+        session = _record_acquired(native_token_fingerprint="fp-acquire")
+        result = self._assess(
+            [_lease_comment()], session, runtime_token_fingerprint="fp-different"
+        )
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("fingerprint" in r for r in result["reasons"]))
+
+    def test_matching_token_fingerprint_allowed(self):
+        session = _record_acquired(native_token_fingerprint="fp-acquire")
+        result = self._assess(
+            [_lease_comment()], session, runtime_token_fingerprint="fp-acquire"
+        )
+        self.assertTrue(result["finalize_allowed"], result["reasons"])
+
+    def test_missing_lease_marker_on_thread_rejected(self):
+        session = _record_acquired()
+        result = self._assess([_lease_comment(comment_id=99999)], session)
+        self.assertFalse(result["finalize_allowed"])
+        self.assertTrue(any("lease marker comment" in r for r in result["reasons"]))
+
+    def test_already_terminal_lease_is_idempotent(self):
+        session = _record_acquired()
+        comments = [
+            _lease_comment(),
+            _lease_comment(phase="released", comment_id=12360),
+        ]
+        result = self._assess(comments, session)
+        self.assertTrue(result["already_terminal"])
+        self.assertFalse(result["finalize_allowed"])
+        self.assertIsNone(result["finalization_body"])
+        self.assertEqual(result["reasons"], [])
+
+    def test_abandoned_marker_ends_the_lease(self):
+        comments = [
+            _lease_comment(),
+            _lease_comment(phase="abandoned", comment_id=12361),
+        ]
+        self.assertIsNone(
+            leases.find_active_reviewer_lease(comments, pr_number=PR)
+        )
+
+    def test_finalization_is_append_only(self):
+        session = _record_acquired()
+        original = _lease_comment()
+        comments = [original]
+        result = self._assess(comments, session)
+        self.assertTrue(result["finalize_allowed"], result["reasons"])
+        # The assessment never mutates or removes prior ledger entries.
+        self.assertEqual(comments, [original])
+        self.assertEqual(result["lease_comment_id"], 12354)
+
+
+class TestCrossModuleTerminalPhaseAgreement(unittest.TestCase):
+    """#742 review 460 — reviewer_pr_lease and pr_work_lease must agree.
+
+    Both modules parse the same append-only lease markers. A phase that is
+    terminal in one and active in the other produces two conflicting truths for
+    the same comment, so an abandoned finalization would still read as an active
+    lease to the conflict-fix acquire and PR-sync inventory paths.
+    """
+
+    TERMINAL_PHASES = ("released", "blocked", "done", "abandoned")
+    NONTERMINAL_PHASES = ("claimed", "validating")
+
+    def _both_active(self, phase: str) -> tuple[bool, bool]:
+        comments = [_lease_comment(phase=phase)]
+        return (
+            leases.find_active_reviewer_lease(comments, pr_number=PR) is not None,
+            pwl.find_active_reviewer_lease(comments, pr_number=PR) is not None,
+        )
+
+    def test_abandoned_marker_is_not_active_in_pr_work_lease(self):
+        _, pwl_active = self._both_active("abandoned")
+        self.assertFalse(pwl_active)
+
+    def test_terminal_phases_agree_across_modules(self):
+        for phase in self.TERMINAL_PHASES:
+            with self.subTest(phase=phase):
+                rpl_active, pwl_active = self._both_active(phase)
+                self.assertFalse(rpl_active)
+                self.assertFalse(pwl_active)
+                self.assertEqual(rpl_active, pwl_active)
+
+    def test_nonterminal_phases_remain_active_in_both_modules(self):
+        for phase in self.NONTERMINAL_PHASES:
+            with self.subTest(phase=phase):
+                rpl_active, pwl_active = self._both_active(phase)
+                self.assertTrue(rpl_active)
+                self.assertTrue(pwl_active)
+
+    def test_terminal_phase_sets_are_mirrored(self):
+        self.assertEqual(
+            set(leases._TERMINAL_PHASES), set(pwl._TERMINAL_REVIEWER_PHASES)
+        )
+
+    def test_default_released_finalization_behavior_unchanged(self):
+        # The default outcome is still 'released', and a released marker ends
+        # the lease for both modules exactly as before this change.
+        leases.clear_session_lease()
+        try:
+            session = _record_acquired()
+            result = mla.assess_merger_lease_finalization(
+                [_lease_comment()],
+                pr_number=PR,
+                session=session,
+                actor_identity="merger-user",
+                actor_profile="prgs-merger",
+                actor_session_id="merger-session",
+                repo=REPO,
+                worktree="branches/merger-pr740",
+                candidate_head=HEAD,
+                live_head_sha=HEAD,
+            )
+            self.assertTrue(result["finalize_allowed"], result["reasons"])
+            self.assertEqual(result["outcome"], mla.OUTCOME_RELEASED)
+            self.assertIn("phase: released", result["finalization_body"])
+            self.assertEqual(
+                result["finalization_reason"],
+                mla.DEFAULT_MERGER_FINALIZATION_REASON,
+            )
+        finally:
+            leases.clear_session_lease()
+
+    def test_finalization_appends_and_never_rewrites_prior_markers(self):
+        claimed = _lease_comment()
+        original_body = claimed["body"]
+        ledger = [claimed]
+        # A terminal marker is appended alongside the original claim; the
+        # earlier marker is left byte-for-byte intact and is never removed.
+        terminal = _lease_comment(phase="abandoned", comment_id=12361)
+        ledger.append(terminal)
+        self.assertEqual(len(ledger), 2)
+        self.assertEqual(ledger[0]["body"], original_body)
+        self.assertEqual(ledger[0]["id"], 12354)
+        # Newest-wins (#577): the appended terminal marker ends the lease.
+        self.assertIsNone(leases.find_active_reviewer_lease(ledger, pr_number=PR))
+
+    def test_full_ledger_terminal_phases_end_the_lease_in_both_modules(self):
+        for phase in self.TERMINAL_PHASES:
+            with self.subTest(phase=phase):
+                ledger = [
+                    _lease_comment(),
+                    _lease_comment(phase=phase, comment_id=12361),
+                ]
+                for module in (leases, pwl):
+                    self.assertIsNone(
+                        module.find_active_reviewer_lease(ledger, pr_number=PR),
+                        f"{module.__name__} still active after {phase}",
+                    )
+
+
+class TestFullLedgerNewestWins(unittest.TestCase):
+    """#742 second remediation — chain-scoped newest-wins in pr_work_lease.
+
+    On a realistic append-only ledger (claim -> heartbeat -> terminal),
+    pr_work_lease.find_active_reviewer_lease skipped the terminal marker and
+    walked backward to the older claim of that same chain, resurrecting it. A
+    terminal marker must end its own chain, while a foreign, forged, or
+    malformed terminal marker must never cancel someone else's active lease.
+    """
+
+    OTHER_SESSION = "other-session"
+
+    def _claim(self, **overrides):
+        return _lease_comment(**overrides)
+
+    def _terminal(self, phase="released", comment_id=12361, **overrides):
+        return _lease_comment(phase=phase, comment_id=comment_id, **overrides)
+
+    def _pwl_active(self, ledger):
+        return pwl.find_active_reviewer_lease(ledger, pr_number=PR)
+
+    # --- the chain must end -------------------------------------------------
+
+    def test_claim_then_each_terminal_phase_leaves_no_active_lease(self):
+        for phase in ("released", "blocked", "done", "abandoned"):
+            with self.subTest(phase=phase):
+                self.assertIsNone(
+                    self._pwl_active([self._claim(), self._terminal(phase=phase)])
+                )
+
+    def test_claim_heartbeat_terminal_leaves_no_active_lease(self):
+        for phase in ("released", "blocked", "done", "abandoned"):
+            with self.subTest(phase=phase):
+                ledger = [
+                    self._claim(),
+                    self._lease_validating(),
+                    self._terminal(phase=phase, comment_id=12362),
+                ]
+                self.assertIsNone(self._pwl_active(ledger))
+                self.assertIsNone(
+                    leases.find_active_reviewer_lease(ledger, pr_number=PR)
+                )
+
+    def _lease_validating(self):
+        return _lease_comment(phase="validating", comment_id=12355)
+
+    # --- foreign / forged terminal markers must not cancel ------------------
+
+    def test_foreign_session_terminal_does_not_cancel_active_claim(self):
+        ledger = [
+            self._claim(),
+            self._terminal(session_id=self.OTHER_SESSION),
+        ]
+        active = self._pwl_active(ledger)
+        self.assertIsNotNone(active)
+        self.assertEqual(active["session_id"], "merger-session")
+
+    def test_mismatched_chain_fields_do_not_cancel_active_claim(self):
+        cases = {
+            "identity": {"identity": "someone-else"},
+            "profile": {"profile": "prgs-reviewer"},
+            "candidate_head": {"candidate_head": OTHER_HEAD},
+        }
+        for label, override in cases.items():
+            with self.subTest(field=label):
+                ledger = [self._claim(), self._terminal(**override)]
+                self.assertIsNotNone(
+                    self._pwl_active(ledger),
+                    f"terminal with wrong {label} must not cancel the claim",
+                )
+
+    def test_terminal_for_another_pr_does_not_cancel_active_claim(self):
+        ledger = [
+            self._claim(),
+            self._terminal(pr_number=741, comment_id=12363),
+        ]
+        self.assertIsNotNone(self._pwl_active(ledger))
+
+    def test_terminal_for_another_repository_does_not_cancel_active_claim(self):
+        foreign = self._terminal()
+        foreign["body"] = foreign["body"].replace(
+            REPO, "Scaled-Tech-Consulting/Other-Repo"
+        )
+        self.assertIsNotNone(self._pwl_active([self._claim(), foreign]))
+
+    def test_malformed_terminal_marker_does_not_hide_active_claim(self):
+        malformed = self._terminal()
+        # Strip the session id line: no provable chain, so it cancels nothing.
+        malformed["body"] = "\n".join(
+            line
+            for line in malformed["body"].splitlines()
+            if not line.startswith("session_id:")
+        )
+        active = self._pwl_active([self._claim(), malformed])
+        self.assertIsNotNone(active)
+        self.assertEqual(active["session_id"], "merger-session")
+
+    # --- multi-chain ledgers ------------------------------------------------
+
+    def test_newer_active_chain_survives_older_terminated_chain(self):
+        ledger = [
+            self._claim(),
+            self._terminal(comment_id=12361),
+            _lease_comment(session_id=self.OTHER_SESSION, comment_id=12370),
+        ]
+        active = self._pwl_active(ledger)
+        self.assertIsNotNone(active)
+        self.assertEqual(active["session_id"], self.OTHER_SESSION)
+
+    def test_newest_valid_chain_selected_across_multiple_histories(self):
+        ledger = [
+            _lease_comment(session_id="chain-1", comment_id=12300),
+            _lease_comment(session_id="chain-1", phase="released", comment_id=12301),
+            _lease_comment(session_id="chain-2", comment_id=12310),
+            _lease_comment(session_id="chain-2", phase="abandoned", comment_id=12311),
+            _lease_comment(session_id="chain-3", comment_id=12320),
+        ]
+        active = self._pwl_active(ledger)
+        self.assertIsNotNone(active)
+        self.assertEqual(active["session_id"], "chain-3")
+
+    def test_all_chains_terminated_leaves_no_active_lease(self):
+        ledger = [
+            _lease_comment(session_id="chain-1", comment_id=12300),
+            _lease_comment(session_id="chain-1", phase="released", comment_id=12301),
+            _lease_comment(session_id="chain-2", comment_id=12310),
+            _lease_comment(session_id="chain-2", phase="abandoned", comment_id=12311),
+        ]
+        self.assertIsNone(self._pwl_active(ledger))
+
+    # --- no regression in existing behavior ---------------------------------
+
+    def test_single_marker_behavior_matches_reviewer_pr_lease(self):
+        for phase in (
+            "claimed", "validating", "approved", "merging",
+            "released", "blocked", "done", "abandoned",
+        ):
+            with self.subTest(phase=phase):
+                ledger = [_lease_comment(phase=phase)]
+                self.assertEqual(
+                    pwl.find_active_reviewer_lease(ledger, pr_number=PR) is not None,
+                    leases.find_active_reviewer_lease(ledger, pr_number=PR)
+                    is not None,
+                )
+
+    def test_expired_claim_stays_inactive_and_freshness_unchanged(self):
+        past = datetime.now(timezone.utc) - timedelta(hours=4)
+        expired = {
+            "id": 12399,
+            "user": {"login": "merger-user"},
+            "body": leases.format_lease_body(
+                repo=REPO,
+                pr_number=PR,
+                issue_number=742,
+                reviewer_identity="merger-user",
+                profile="prgs-merger",
+                session_id="expired-session",
+                worktree="branches/merger-pr740",
+                phase="claimed",
+                candidate_head=HEAD,
+                target_branch="master",
+                target_branch_sha="e" * 40,
+                last_activity=past,
+                expires_at=past,
+            ),
+        }
+        self.assertIsNone(self._pwl_active([expired]))
+        self.assertIsNone(
+            leases.find_active_reviewer_lease([expired], pr_number=PR)
+        )
+
+    def test_active_claim_without_any_terminal_marker_remains_active(self):
+        self.assertIsNotNone(self._pwl_active([self._claim()]))
+
+    def test_production_readers_agree_after_release_tool_finalization(self):
+        """The ledger the release tool actually writes must read as terminal."""
+        leases.clear_session_lease()
+        try:
+            session = _record_acquired()
+            result = mla.assess_merger_lease_finalization(
+                [_lease_comment()],
+                pr_number=PR,
+                session=session,
+                actor_identity="merger-user",
+                actor_profile="prgs-merger",
+                actor_session_id="merger-session",
+                repo=REPO,
+                worktree="branches/merger-pr740",
+                candidate_head=HEAD,
+                live_head_sha=HEAD,
+                outcome=mla.OUTCOME_ABANDONED,
+            )
+            self.assertTrue(result["finalize_allowed"], result["reasons"])
+            posted = {
+                "id": 12370,
+                "user": {"login": "merger-user"},
+                "body": result["finalization_body"],
+            }
+            ledger = [_lease_comment(), posted]
+            for module in (leases, pwl):
+                with self.subTest(module=module.__name__):
+                    self.assertIsNone(
+                        module.find_active_reviewer_lease(ledger, pr_number=PR)
+                    )
+            # Append-only: the claim marker is still present and unmodified.
+            self.assertEqual(len(ledger), 2)
+            self.assertEqual(ledger[0]["id"], 12354)
+        finally:
+            leases.clear_session_lease()
+
+
+class TestReleaseMergerLeaseTool(unittest.TestCase):
+    """AC7/AC9/AC10/AC11 — the native tool contract end to end."""
+
+    def setUp(self):
+        mcp_server._preflight_whoami_called = True
+        mcp_server._preflight_capability_called = True
+        mcp_server._preflight_resolved_role = "merger"
+        mcp_server._preflight_resolved_task = "release_merger_pr_lease"
+        leases.clear_session_lease()
+
+    def tearDown(self):
+        leases.clear_session_lease()
+
+    def _merger_profile(self, **extra):
+        p = {
+            "username": "merger-user",
+            "profile_name": "prgs-merger",
+            "role": "merger",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.merge",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve",
+                "gitea.pr.review",
+                "gitea.pr.request_changes",
+            ],
+        }
+        p.update(extra)
+        return p
+
+    def _reviewer_profile(self):
+        return {
+            "username": "reviewer-user",
+            "profile_name": "prgs-reviewer",
+            "role": "reviewer",
+            "allowed_operations": [
+                "gitea.read",
+                "gitea.pr.comment",
+                "gitea.pr.review",
+                "gitea.pr.approve",
+            ],
+            "forbidden_operations": ["gitea.pr.merge"],
+        }
+
+    def _call(self, **kwargs):
+        params = {
+            "pr_number": PR,
+            "worktree": "branches/merger-pr740",
+            "candidate_head": HEAD,
+            "remote": "prgs",
+            "org": "Scaled-Tech-Consulting",
+            "repo": "Gitea-Tools",
+        }
+        params.update(kwargs)
+        return mcp_server.gitea_release_merger_pr_lease(**params)
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._authenticated_username", return_value="merger-user")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch(
+        "gitea_mcp_server._resolve",
+        return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
+    )
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_merger_releases_own_acquired_lease(
+        self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_comments.return_value = [_lease_comment()]
+        mock_api.side_effect = [
+            {"state": "open", "head": {"sha": HEAD}},
+            {"id": 12370},
+        ]
+        _record_acquired()
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = self._call()
+        self.assertTrue(res["success"], res)
+        self.assertTrue(res["finalized"])
+        self.assertEqual(res["finalization_comment_id"], 12370)
+        self.assertEqual(res["outcome"], mla.OUTCOME_RELEASED)
+        # AC11: no active in-session lease survives finalization.
+        self.assertIsNone(leases.get_session_lease())
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._authenticated_username", return_value="merger-user")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch(
+        "gitea_mcp_server._resolve",
+        return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
+    )
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_merger_cannot_release_foreign_session_lease(
+        self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_comments.return_value = [_lease_comment(session_id="other-merger")]
+        mock_api.side_effect = [{"state": "open", "head": {"sha": HEAD}}]
+        _record_acquired()
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = self._call()
+        self.assertFalse(res["success"], res)
+        self.assertFalse(res["finalized"])
+        self.assertTrue(res["reasons"])
+        # Fail closed: no comment POST was attempted.
+        self.assertEqual(mock_api.call_count, 1)
+        self.assertIsNotNone(leases.get_session_lease())
+
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server._authenticated_username", return_value="merger-user")
+    @patch("gitea_mcp_server._auth", return_value="token pass")
+    @patch(
+        "gitea_mcp_server._resolve",
+        return_value=("gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools"),
+    )
+    @patch("gitea_mcp_server.get_profile")
+    @patch("gitea_mcp_server._fetch_pr_comments")
+    @patch("gitea_mcp_server._verify_role_mutation_workspace")
+    def test_already_terminal_release_is_idempotent(
+        self, _verify, mock_comments, mock_profile, _resolve, _auth, _user, mock_api
+    ):
+        mock_profile.return_value = self._merger_profile()
+        mock_comments.return_value = [
+            _lease_comment(),
+            _lease_comment(phase="released", comment_id=12360),
+        ]
+        mock_api.side_effect = [{"state": "open", "head": {"sha": HEAD}}]
+        _record_acquired()
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-merger"}):
+            res = self._call()
+        self.assertTrue(res["success"], res)
+        self.assertFalse(res["finalized"])
+        self.assertTrue(res["already_terminal"])
+        # No second marker posted.
+        self.assertEqual(mock_api.call_count, 1)
+        self.assertIsNone(leases.get_session_lease())
+
+    @patch("gitea_mcp_server.get_profile")
+    def test_reviewer_profile_cannot_invoke_merger_release(self, mock_profile):
+        mock_profile.return_value = self._reviewer_profile()
+        _record_acquired()
+        with patch.dict(os.environ, {"GITEA_MCP_PROFILE_NAME": "prgs-reviewer"}):
+            res = self._call()
+        self.assertFalse(res["success"], res)
+        self.assertFalse(res["finalized"])
+        self.assertTrue(res["reasons"])
+        # The lease is untouched by a rejected cross-role call.
+        self.assertIsNotNone(leases.get_session_lease())
+
+
+class TestReleaseMergerLeaseCapability(unittest.TestCase):
+    """AC9 — capability map binds the task to gitea.pr.comment + merger role."""
+
+    def test_task_and_tool_alias_are_merger_scoped(self):
+        for task in ("release_merger_pr_lease", "gitea_release_merger_pr_lease"):
+            with self.subTest(task=task):
+                self.assertEqual(
+                    tcm.required_permission(task), "gitea.pr.comment"
+                )
+                self.assertEqual(tcm.required_role(task), "merger")
+
+    def test_reviewer_release_tool_remains_reviewer_scoped(self):
+        self.assertTrue(hasattr(mcp_server, "gitea_release_reviewer_pr_lease"))
+        self.assertNotEqual(
+            mcp_server.gitea_release_reviewer_pr_lease,
+            mcp_server.gitea_release_merger_pr_lease,
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_reviewer_pr_lease.py

@@ -79,22 +79,26 @@ class TestReviewerLeaseAcquire(unittest.TestCase):
 
 
 class TestReviewerLeaseFreshness(unittest.TestCase):
-    def test_stale_warning_after_30_minutes(self):
+    def test_stale_warning_at_half_the_sliding_window(self):
+        # #747 warns at half the 10-minute window, while the owner can still
+        # heartbeat and keep the lease.
         lease = leases.parse_lease_comment(
-            _lease_comment(382, "session-a", minutes_ago=35)["body"]
+            _lease_comment(382, "session-a", minutes_ago=6)["body"]
         )
         self.assertEqual(
             leases.classify_lease_freshness(lease),
             "stale_warning",
         )
 
-    def test_reclaimable_after_60_minutes(self):
+    def test_expired_once_the_sliding_window_lapses(self):
+        # Pre-#747 a 65-minute-idle lease was "reclaimable" and had to wait out
+        # a second timer. It is now simply expired and immediately takeable.
         lease = leases.parse_lease_comment(
             _lease_comment(382, "session-a", minutes_ago=65)["body"]
         )
         self.assertEqual(
             leases.classify_lease_freshness(lease),
-            "reclaimable",
+            "expired",
         )
 
 
@@ -281,7 +285,10 @@ class TestReviewerLeaseHandoffDiagnose(unittest.TestCase):
         self.assertEqual(result["active_lease"]["comment_id"], 8647)
         self.assertFalse(result["mutation_allowed"])
 
-    def test_foreign_reclaimable_release_expired(self):
+    def test_foreign_expired_release_expired(self):
+        # Pre-#747 this classified as "foreign_reclaimable" after the 60-minute
+        # activity band. Under the sliding TTL the lease is simply expired, and
+        # the sanctioned next action is unchanged.
         reclaim = _lease_comment(
             592, "foreign-old", phase="claimed", minutes_ago=65
         )
@@ -293,7 +300,7 @@ class TestReviewerLeaseHandoffDiagnose(unittest.TestCase):
             current_reviewer_identity="sysadmin",
             proposed_worktree="branches/review-pr-592",
         )
-        self.assertEqual(result["classification"], "foreign_reclaimable")
+        self.assertEqual(result["classification"], "foreign_expired")
         self.assertEqual(
             result["next_action"], leases.NEXT_ACTION_RELEASE_EXPIRED_LEASE
         )