fix(mcp): complete canonical-root consumption for cross-repo namespaces (Closes #739)

#706 routed the #274 filesystem guards and the session repository slug through
the configured canonical_repository_root. Three consumption paths were left
deriving from the Gitea-Tools installation checkout.

F1 — gitea_get_runtime_context did not normalize its `remote` argument, while
gitea_whoami did. On a prgs-hosted namespace whose first native call was the
runtime-context path, the 'dadeschools' argument default was pinned: identity
resolved against the wrong host, and because first-bind is first-write-wins a
later correct gitea_whoami(remote="prgs") could not repair the binding. It now
calls _effective_remote before any host lookup, identity resolution, or session
seeding. Explicit remotes pass through untouched and a dadeschools-hosted
profile still resolves to dadeschools.

F2 — _delete_branch_repository_binding_block derived its expected slug from
_workspace_repository_slug, which reads _local_git_remote_url in PROJECT_ROOT
(always Gitea-Tools). For a cross-repository namespace this inverted the guard:
the genuinely bound target was rejected and Gitea-Tools was accepted. The
canonical-root branch already present in _trusted_session_repository is
extracted as _canonical_repository_slug and shared by both call sites. A
configured-but-unresolvable root now fails closed instead of falling back to
the installation identity; unconfigured namespaces keep the install-derived
default unchanged.

F3 — gitea_assess_master_parity measures Gitea-Tools server implementation
parity only. That is intentional and is preserved: startup_head, current_head,
in_parity, stale, and restart_required keep their existing meaning and values,
and only the server dimension gates mutations. Two separately labelled
dimensions are added — server_implementation (installation root and commit) and
target_repository (canonical root, repository slug, checkout head, last-known
remote master head, staleness) — so cross-repository commissioning evidence can
distinguish them. The target assessment makes no network call: the remote side
is read from the existing remote-tracking ref, and an unfetched target is
reported indeterminate rather than guessed at.

Verified intentional and left unchanged: the #274 workspace-membership,
author-mutation-worktree, and branches-only guards already validate against the
configured canonical root; explicit org/repo may confirm but never authorize a
binding; and the four-role repository-specific configuration surface already
passes audit and bind-time validation with a wrong root failing closed.

Tests: 31 new cases in tests/test_cross_repo_canonical_consumption.py using
real git repositories and no network. Full suite 3298 passed / 6 skipped
against a clean-baseline 3267; the 2 failures
(test_issue_702_review_findings_f1_f6, test_reconciler_supersession_close)
reproduce identically on unmodified master a8d2087 and are pre-existing.

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

gitea_mcp_server.py

@@ -1648,6 +1648,45 @@ def _workspace_repository_slug(remote: str | None) -> str | None:
     return session_ctx.format_repository_slug(parsed[0], parsed[1])
 
 
+def _canonical_repository_slug(
+    profile: dict,
+    remote: str | None,
+) -> tuple[str | None, list[str]]:
+    """Repository identity of the *configured* canonical root, if any (#739 F2).
+
+    Returns ``(slug, reasons)``. A namespace with no configured
+    ``canonical_repository_root`` yields ``(None, [])`` so callers keep the
+    install-derived single-repository default. A configured root that cannot be
+    resolved to a repository identity yields ``(None, reasons)`` — a fail-closed
+    signal, never a silent fallback to the installation checkout, because that
+    fallback is exactly what lets a cross-repository namespace validate against
+    the wrong repository.
+
+    Env-over-profile precedence and the existence / git-toplevel / resolvable
+    remote-identity validation all live in ``crr``.
+    """
+    configured_value, configured_source = crr.configured_canonical_root(
+        profile, os.environ
+    )
+    if not configured_value:
+        return None, []
+    assessment = crr.assess_canonical_repository_root(
+        configured_value=configured_value,
+        source=configured_source,
+        expected_slug=None,
+        process_project_root=PROJECT_ROOT,
+        remote=remote,
+        require_binding=True,
+    )
+    slug = assessment.get("resolved_slug")
+    if assessment.get("block") or not slug:
+        return None, list(assessment.get("reasons") or [
+            "configured canonical repository root has no resolvable "
+            "repository identity (fail closed)"
+        ])
+    return slug, []
+
+
 def _trusted_session_repository(
     profile: dict,
     remote: str | None,
@@ -1686,32 +1725,10 @@ def _trusted_session_repository(
     # self-authorizing). Env-over-profile precedence and fail-closed validation
     # (existence, git toplevel, resolvable remote identity) are handled by
     # ``crr``; unconfigured single-repo namespaces keep the install-derived slug.
-    configured_value, configured_source = crr.configured_canonical_root(
-        profile, os.environ
-    )
-    if configured_value:
-        assessment = crr.assess_canonical_repository_root(
-            configured_value=configured_value,
-            source=configured_source,
-            expected_slug=None,
-            process_project_root=PROJECT_ROOT,
-            remote=remote,
-            require_binding=True,
-        )
-        slug = assessment.get("resolved_slug")
-        if assessment.get("block") or not slug:
-            return {
-                "org": None,
-                "repository": None,
-                "reasons": assessment.get("reasons")
-                or [
-                    "configured canonical repository root has no resolvable "
-                    "repository identity; session repository cannot be "
-                    "authorized (fail closed)"
-                ],
-            }
-    else:
-        slug = _workspace_repository_slug(remote)
+    canonical_slug, canonical_reasons = _canonical_repository_slug(profile, remote)
+    if canonical_reasons:
+        return {"org": None, "repository": None, "reasons": canonical_reasons}
+    slug = canonical_slug or _workspace_repository_slug(remote)
     scope = session_ctx.assess_repository_scope(
         workspace_slug=slug,
         allowed=allowed,
@@ -8768,8 +8785,28 @@ def _delete_branch_repository_binding_block(
       (unproven target).
 
     Returns a structured block dict, or ``None`` when the target is authorized.
+
+    #739 F2: the trusted identity is the session's configured
+    ``canonical_repository_root`` when the namespace declares one. Deriving it
+    from ``_workspace_repository_slug`` alone reads the *installation* checkout
+    (``_local_git_remote_url`` runs in ``PROJECT_ROOT``, always Gitea-Tools),
+    which inverts the guard for a cross-repository namespace: the genuinely
+    bound target is rejected and Gitea-Tools is accepted. A configured root that
+    cannot be resolved fails closed rather than falling back to the install
+    identity; unconfigured namespaces keep the install-derived default.
     """
-    workspace_slug = _workspace_repository_slug(remote)
+    canonical_slug, canonical_reasons = _canonical_repository_slug(
+        get_profile(), remote
+    )
+    if canonical_reasons:
+        return {
+            "success": False,
+            "performed": False,
+            "required_permission": "gitea.branch.delete",
+            "reasons": canonical_reasons,
+            "blocker_kind": "repository_binding",
+        }
+    workspace_slug = canonical_slug or _workspace_repository_slug(remote)
     workspace_parts = (
         session_ctx.parse_repository_slug(workspace_slug) if workspace_slug else None
     )
@@ -11585,11 +11622,6 @@ 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,
@@ -11779,191 +11811,6 @@ 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,
@@ -14226,6 +14073,16 @@ def gitea_get_runtime_context(
         remote: Known instance — 'dadeschools' or 'prgs'.
         host: Override the Gitea host.
     """
+    # #739 F1: normalize the remote *before* any host lookup, identity
+    # resolution, or session seeding. ``remote`` defaults to 'dadeschools'
+    # across the whole tool surface, so a caller that omits it on a prgs-hosted
+    # namespace would otherwise pin the argument default: the identity lookup
+    # below would authenticate against the wrong host, and because first-bind is
+    # first-write-wins a later, correct ``gitea_whoami(remote="prgs")`` could not
+    # repair the binding. ``gitea_whoami`` has always normalized here; this is
+    # the same call, not a new policy. Explicit non-default remotes pass through
+    # untouched, and a dadeschools-hosted profile still resolves to dadeschools.
+    remote = _effective_remote(remote)
     profile = get_profile()
     config = gitea_config.load_config()
     reveal = _reveal_endpoints()
@@ -14403,13 +14260,23 @@ def gitea_assess_master_parity(
     Never mutates and makes no network calls. Read-only operations are never
     blocked by staleness; only mutating operations fail closed while stale.
 
+    The top-level ``startup_head``/``current_head``/``in_parity`` fields keep
+    their original meaning: the *Gitea-Tools server implementation* only. #739
+    F3 adds two separately labelled dimensions so a cross-repository namespace
+    can tell them apart — ``server_implementation`` (this installation, restated
+    under an explicit name) and ``target_repository`` (the configured canonical
+    target checkout and its last-known remote master). Only the server dimension
+    gates mutations.
+
     Returns:
         dict with 'in_parity', 'stale', 'restart_required', 'startup_head',
-        'current_head', 'mutation_gate_enforced', 'summary', 'reasons', and a
-        'report' recovery payload when stale.
+        'current_head', 'mutation_gate_enforced', 'summary', 'reasons',
+        'server_implementation', 'target_repository', and a 'report' recovery
+        payload when stale.
     """
     parity = _current_master_parity()
     enforced = not master_parity_gate.gate_disabled()
+    canonical_root, canonical_source = _configured_canonical_root()
     out = {
         "in_parity": parity["in_parity"],
         "stale": parity["stale"],
@@ -14421,6 +14288,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)
@@ -16672,8 +16550,6 @@ 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,7 +20,6 @@ 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,
@@ -31,21 +30,6 @@ 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(
     *,
@@ -113,7 +97,6 @@ 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)
@@ -134,76 +117,9 @@ 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
@@ -215,8 +131,6 @@ 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
@@ -254,8 +168,6 @@ 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:
@@ -304,7 +216,6 @@ _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|"
@@ -336,267 +247,6 @@ 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],
     *,

pr_work_lease.py

@@ -20,11 +20,7 @@ _FIELD_RE = re.compile(
     re.IGNORECASE | re.MULTILINE,
 )
 
-# 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"})
+_TERMINAL_REVIEWER_PHASES = frozenset({"done", "released", "blocked"})
 _ACTIVE_REVIEWER_PHASES = frozenset({
     "claimed",
     "validating",
@@ -157,72 +153,25 @@ 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, 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.
-    """
+    """Return the newest unexpired reviewer lease for *pr_number*, if any."""
     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 index in range(len(candidates) - 1, -1, -1):
-        lease = candidates[index]
+    for lease in reversed(candidates):
         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,10 +16,7 @@ _FIELD_RE = re.compile(
 )
 _FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
 
-# "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"})
+_TERMINAL_PHASES = frozenset({"done", "released", "blocked"})
 _ACTIVE_PHASES = frozenset({
     "claimed",
     "validating",

task_capability_map.py

@@ -137,17 +137,6 @@ 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_merger_lease_finalization.py

@@ -1,956 +0,0 @@
-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()