fix(mcp): bind post-merge moot-lease cleanup to the reconciler capability (Closes #745)

gitea_cleanup_post_merge_moot_lease (#515) posts a terminal `phase: released`
lease marker — a durable mutation of the PR lease ledger — but had no entry in
the canonical task-capability map and no role binding. Entry gated on
gitea.read, apply gated only on gitea.pr.comment, so any profile holding the
comment permission (author, reviewer, merger) reached the mutation path, while
the reconciler could not satisfy the operator-required resolve-exact-task ->
mutation sequence because no cleanup task was resolvable at all. The apply path
also called verify_preflight_purity(remote) with no task/org/repo, skipping the
resolved-task, canonical-root and explicit-target checks its siblings perform,
and passed request org/repo straight into _resolve.

Capability map and router:

- Map `cleanup_post_merge_moot_lease` and the tool-name alias
  `gitea_cleanup_post_merge_moot_lease` to gitea.pr.comment + reconciler.
  Both names carry an identical contract; unknown names keep failing closed on
  the map's KeyError.
- Add both to role_session_router RECONCILER_TASKS and TASK_REQUIRED_ROLE so
  the map and the router cannot disagree (the #723 defect-A class).

Tool enforcement (apply path only):

- Require the session to have resolved exactly the cleanup task; resolving a
  different task, including a sibling reconciler task, does not authorize it.
- Require the reconciler role, checked independently of the permission gate.
- Require gitea.pr.comment.
- Validate explicit org/repo against the canonical repository identity derived
  from the session binding, so request parameters can never redirect the
  mutation, and forward worktree_path/task/org/repo to verify_preflight_purity
  for canonical-root, workspace and anti-stomp binding (#733/#739).
- Require matching dry-run evidence proving lease_moot and cleanup_allowed for
  the same PR, lease session, candidate head and lease marker id, with optional
  caller expectations checked against the live lease.

New post_merge_moot_lease_gate module holds the pure authorization logic and an
append-only dry-run ledger: entries are only ever appended, lookup is
newest-wins, and dry_run_history hands out copies. Live, non-moot, superseded,
mismatched, malformed and foreign-repository leases all fail closed;
already-terminal cleanup stays idempotent.

The read-only apply=false assessment deliberately stays reachable under
gitea.read with no role gate, matching gitea_cleanup_stale_review_decision_lock
and gitea_cleanup_obsolete_reviewer_comment_lease, so an operator can diagnose a
stuck lease from any attached namespace. This choice is documented in the map,
the tool docstring and docs/gitea-execution-profiles.md, and is directly tested.

_delete_branch_repository_binding_block is generalized into
_repository_binding_block(required_permission=...) and retained as a thin
delete-path alias so #733/#739 coverage keeps exercising its permission label.

Tests: new tests/test_issue_745_moot_lease_reconciler_gate.py (39 tests, 35
subtests) covers the map/router contract, alias parity, unknown-name rejection,
role and task gates, dry-run/apply sequencing, superseded and malformed leases,
repository binding, ledger append-only behavior, idempotency, and asserts no
production PR/session/marker is referenced. tests/test_post_merge_moot_lease.py
is updated from the permission-only model to reconciler + dry-run evidence, with
a new negative test pinning that a merger can no longer apply.

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

docs/gitea-execution-profiles.md

@@ -317,6 +317,44 @@ Least-privilege constraints:
   canonical names such as `gitea.pr.close` (never bare `pr.close` /
   `issue.close`, which the production normalizer rejects or drops).
 
+### Post-merge moot-lease cleanup ownership (`gitea.pr.comment`)
+
+Neutralising a reviewer lease left behind on an already-merged/closed PR is
+reconciliation work too. `task_capability_map` maps
+`cleanup_post_merge_moot_lease` — and its tool-name alias
+`gitea_cleanup_post_merge_moot_lease` — to role `reconciler` with permission
+`gitea.pr.comment` (#745). Both names carry the **same** contract.
+
+The permission alone is deliberately not sufficient: author, reviewer and
+merger profiles all hold `gitea.pr.comment` for ordinary PR discussion, so the
+role gate — not the permission gate — is what keeps the terminal lease marker
+reconciler-owned.
+
+`gitea_cleanup_post_merge_moot_lease` splits its two modes on purpose:
+
+- **`apply=false` (assessment) requires only `gitea.read`, with no role gate.**
+  This matches `gitea_cleanup_stale_review_decision_lock` and
+  `gitea_cleanup_obsolete_reviewer_comment_lease`, whose assessment paths are
+  likewise read-gated, so an operator can diagnose a stuck lease from whichever
+  namespace happens to be attached without switching roles. The dry run
+  performs no mutation and records append-only evidence in-session.
+- **`apply=true` (mutation) requires all of the following**, in order: the
+  session must have resolved exactly `cleanup_post_merge_moot_lease` (resolving
+  any other task — including a sibling reconciler task — does not authorize
+  it); the active role must be `reconciler`; the profile must hold
+  `gitea.pr.comment`; the explicit `org`/`repo` must agree with the canonical
+  repository identity, which is derived from the session binding and can never
+  be overridden by request parameters; and matching dry-run evidence must show
+  `lease_moot`, `cleanup_allowed`, and the same PR, lease session, candidate
+  head and lease marker id that are live at apply time.
+
+Everything else fails closed: a live lease on an open PR, an already-terminal
+(idempotent) lease, a lease superseded between the dry run and the apply, a
+malformed lease missing session/head/marker, and any foreign-repository target.
+The cleanup only ever appends a terminal `phase: released` marker
+(`blocker: post-merge-moot`) — it never edits or deletes another session's
+comment, and it never merges or adopts a lease.
+
 Launch a static `gitea-reconciler` MCP namespace with
 `GITEA_MCP_PROFILE=prgs-reconciler`. Profile shape is validated by
 `reconciler_profile.assess_reconciler_profile` (#304). Use the

gitea_mcp_server.py

@@ -1829,6 +1829,7 @@ def _seed_session_context(
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
+import post_merge_moot_lease_gate  # noqa: E402  # #745 reconciler cleanup gate
 import merger_lease_adoption  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import worktree_cleanup_audit  # noqa: E402
@@ -8859,13 +8860,14 @@ def gitea_review_pr(
         return out
 
 
-def _delete_branch_repository_binding_block(
+def _repository_binding_block(
     remote: str | None,
     *,
     org: str | None,
     repo: str | None,
+    required_permission: str = "gitea.branch.delete",
 ) -> dict | None:
-    """#733: validate an explicit delete target against the workspace binding.
+    """#733: validate an explicit mutation target against the workspace binding.
 
     The trusted repository identity comes only from the verified,
     workspace-aligned git remote (never ``REMOTES`` defaults, never
@@ -8898,7 +8900,7 @@ def _delete_branch_repository_binding_block(
         return {
             "success": False,
             "performed": False,
-            "required_permission": "gitea.branch.delete",
+            "required_permission": required_permission,
             "reasons": canonical_reasons,
             "blocker_kind": "repository_binding",
         }
@@ -8917,9 +8919,9 @@ def _delete_branch_repository_binding_block(
             return {
                 "success": False,
                 "performed": False,
-                "required_permission": "gitea.branch.delete",
+                "required_permission": required_permission,
                 "reasons": list(override.get("reasons") or [
-                    "delete target repository does not match the workspace "
+                    "target repository does not match the workspace "
                     "binding (fail closed)"
                 ]),
                 "blocker_kind": "repository_binding",
@@ -8929,17 +8931,51 @@ def _delete_branch_repository_binding_block(
         return {
             "success": False,
             "performed": False,
-            "required_permission": "gitea.branch.delete",
+            "required_permission": required_permission,
             "reasons": [
                 "repository binding unverified: no workspace repository "
                 "identity could be established to corroborate the explicit "
-                "delete target (fail closed)"
+                "target (fail closed)"
             ],
             "blocker_kind": "repository_binding",
         }
     return None
 
 
+def _delete_branch_repository_binding_block(
+    remote: str | None,
+    *,
+    org: str | None,
+    repo: str | None,
+) -> dict | None:
+    """Delete-branch view of :func:`_repository_binding_block` (#733).
+
+    Retained as the named entry point for the delete path so #733/#739
+    regression coverage keeps exercising the exact permission label that
+    ``gitea_delete_branch`` reports.
+    """
+    return _repository_binding_block(
+        remote, org=org, repo=repo,
+        required_permission="gitea.branch.delete",
+    )
+
+
+def _bound_repository_slug(remote: str | None) -> str | None:
+    """Canonical repository slug for the active session, or None (#745).
+
+    Same trust order as ``_repository_binding_block``: the configured canonical
+    root when the namespace declares one, otherwise the workspace-derived
+    identity. Request parameters are never consulted, so they cannot redirect a
+    mutation at a foreign repository.
+    """
+    canonical_slug, canonical_reasons = _canonical_repository_slug(
+        get_profile(), remote
+    )
+    if canonical_reasons:
+        return None
+    return canonical_slug or _workspace_repository_slug(remote)
+
+
 @mcp.tool()
 def gitea_delete_branch(
     branch: str,
@@ -12352,6 +12388,10 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
 def gitea_cleanup_post_merge_moot_lease(
     pr_number: int,
     apply: bool = False,
+    expected_session_id: str | None = None,
+    expected_candidate_head: str | None = None,
+    expected_lease_comment_id: int | None = None,
+    worktree_path: str | None = None,
     remote: str = "dadeschools",
     host: str | None = None,
     org: str | None = None,
@@ -12367,14 +12407,37 @@ def gitea_cleanup_post_merge_moot_lease(
     an append-only comment that neutralises the moot lease without deleting any
     other session's comment.
 
+    Role binding (#745). The two modes are gated differently, on purpose:
+
+    * ``apply=false`` (assessment) stays reachable under ``gitea.read`` for
+      **any** role, matching ``gitea_cleanup_stale_review_decision_lock`` and
+      ``gitea_cleanup_obsolete_reviewer_comment_lease``, so an operator can
+      diagnose a stuck lease from whichever namespace is attached. It performs
+      no mutation and records append-only dry-run evidence in-session.
+    * ``apply=true`` (mutation) additionally requires, in order: the session to
+      have resolved exactly ``cleanup_post_merge_moot_lease``; the
+      ``reconciler`` role; ``gitea.pr.comment``; a validated repository /
+      canonical-root / workspace binding; and matching dry-run evidence proving
+      ``lease_moot`` and ``cleanup_allowed`` for the same PR, lease session,
+      candidate head and lease marker. Author, reviewer and merger fail closed
+      here even though their profiles carry ``gitea.pr.comment``.
+
     Args:
         pr_number: The PR whose lingering lease to assess/clean.
         apply: When false (default) report only (read-only). When true, post the
-            terminal released marker if — and only if — cleanup is allowed.
+            terminal released marker if — and only if — cleanup is authorized.
+        expected_session_id: Optional lease session the caller expects; a
+            mismatch against the live lease fails closed.
+        expected_candidate_head: Optional leased head the caller expects; a
+            mismatch against the live lease fails closed.
+        expected_lease_comment_id: Optional lease marker id the caller expects;
+            a mismatch against the live lease fails closed.
+        worktree_path: Reconciler worktree to bind the apply-path preflight to.
         remote: Known instance — 'dadeschools' or 'prgs'.
         host: Override the Gitea host.
-        org: Override the owner/organization.
-        repo: Override the repository name.
+        org: Override the owner/organization. Validated against the canonical
+            repository identity; it can never redirect the mutation.
+        repo: Override the repository name. Validated as above.
 
     Returns:
         dict reporting PR merged/closed state, merge_commit_sha, linked-issue
@@ -12434,14 +12497,62 @@ def gitea_cleanup_post_merge_moot_lease(
         "reasons": assessment.get("reasons") or [],
     }
 
+    # The canonical repository identity comes from the session binding only —
+    # never from the org/repo request parameters (#745 requirement 10).
+    repository_slug = _bound_repository_slug(remote)
+    report["repository_slug"] = repository_slug
+    report["required_task"] = post_merge_moot_lease_gate.CLEANUP_TASK
+    report["required_role_kind"] = post_merge_moot_lease_gate.REQUIRED_ROLE
+
     if not apply:
+        # Append-only dry-run evidence. Recorded for every assessment, allowed
+        # or not, so a later apply can prove the lease it saw is the lease that
+        # is still there. Prior entries are never rewritten.
+        evidence = post_merge_moot_lease_gate.record_dry_run(
+            pr_number=pr_number,
+            repository_slug=repository_slug,
+            lease_moot=bool(assessment.get("is_moot")),
+            cleanup_allowed=bool(assessment.get("cleanup_allowed")),
+            session_id=active.get("session_id"),
+            candidate_head=active.get("candidate_head"),
+            lease_comment_id=active.get("comment_id"),
+        )
+        report["dry_run_evidence"] = evidence
         return report
+
+    # ---------------------------------------------------------------- apply --
+    # Preserve the existing dry-run safety assessment: a lease that is not moot
+    # (open PR, already-terminal, absent) is refused here exactly as before, and
+    # no mutation is attempted.
     if not assessment.get("cleanup_allowed"):
         report["cleanup_skipped_reason"] = (
             assessment.get("reasons") or ["cleanup not allowed"]
         )
         return report
 
+    # 1 + 2 + 6: exact resolved cleanup task, reconciler role, and matching
+    # append-only dry-run evidence. Permission alone is deliberately not enough.
+    authorization = post_merge_moot_lease_gate.assess_apply_authorization(
+        pr_number=pr_number,
+        repository_slug=repository_slug,
+        resolved_task=_preflight_resolved_task,
+        active_role_kind=_profile_role_kind(get_profile()),
+        assessment=assessment,
+        evidence=post_merge_moot_lease_gate.latest_dry_run(
+            pr_number=pr_number, repository_slug=repository_slug
+        ),
+        expected_session_id=expected_session_id,
+        expected_candidate_head=expected_candidate_head,
+        expected_lease_comment_id=expected_lease_comment_id,
+    )
+    report["authorization"] = authorization
+    if not authorization["allowed"]:
+        report["success"] = False
+        report["reasons"] = authorization["reasons"]
+        report["blocker_kind"] = authorization["blocker_kind"]
+        return report
+
+    # 3: the dedicated mutation permission.
     comment_block = _profile_operation_gate("gitea.pr.comment")
     if comment_block:
         report["success"] = False
@@ -12449,7 +12560,28 @@ def gitea_cleanup_post_merge_moot_lease(
         report["permission_report"] = _permission_block_report("gitea.pr.comment")
         return report
 
-    verify_preflight_purity(remote)
+    # 4: explicit target must agree with the canonical repository identity
+    # before any preflight or mutation (#733/#739).
+    repo_binding_block = _repository_binding_block(
+        remote, org=org, repo=repo,
+        required_permission="gitea.pr.comment",
+    )
+    if repo_binding_block:
+        report["success"] = False
+        report["reasons"] = repo_binding_block["reasons"]
+        report["blocker_kind"] = repo_binding_block["blocker_kind"]
+        return report
+
+    # 4 (cont.): canonical root, workspace binding and the resolved-task match
+    # are enforced by the shared preflight, with the explicit target forwarded
+    # so the #604 anti-stomp resolution validates the targeted repository.
+    verify_preflight_purity(
+        remote,
+        worktree_path=worktree_path,
+        task=post_merge_moot_lease_gate.CLEANUP_TASK,
+        org=org,
+        repo=repo,
+    )
     body = assessment["release_body"]
     comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
     with _audited(

post_merge_moot_lease_gate.py

@@ -0,0 +1,279 @@
+"""Reconciler authorization gate for post-merge moot-lease cleanup (#745).
+
+``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase:
+released`` lease marker — a real, durable mutation of the PR lease ledger.
+Before #745 it was gated on permissions alone (``gitea.read`` to enter,
+``gitea.pr.comment`` to apply) with no canonical task and no role binding, so
+any profile carrying ``gitea.pr.comment`` reached the mutation path while the
+reconciler could not satisfy the operator-required resolve-exact-task ->
+mutation sequence.
+
+This module holds the pure half of that gate:
+
+* the canonical task name and its tool-name alias;
+* an **append-only** in-process ledger of read-only dry-run assessments;
+* ``assess_apply_authorization``, which decides whether an apply may proceed.
+
+Apply is authorized only when all of the following hold:
+
+* the session resolved exactly the cleanup task (no other task substitutes);
+* the active profile role is ``reconciler``;
+* a prior dry run in this session recorded ``lease_moot`` and
+  ``cleanup_allowed`` for the *same* repository, PR, lease session, candidate
+  head and lease marker id;
+* the live assessment still agrees with that evidence, so a lease superseded
+  between the dry run and the apply fails closed;
+* any caller-supplied expectations match the live lease exactly.
+
+Everything else fails closed. The ledger is only ever appended to — a
+superseded dry run stays visible as history instead of being rewritten — which
+keeps the cleanup audit trail append-only end to end.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+CLEANUP_TASK = "cleanup_post_merge_moot_lease"
+CLEANUP_TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease"
+REQUIRED_ROLE = "reconciler"
+REQUIRED_PERMISSION = "gitea.pr.comment"
+
+# The read-only assessment stays reachable under gitea.read for every role —
+# the convention shared with cleanup_stale_review_decision_lock and
+# cleanup_obsolete_reviewer_comment_lease — so any namespace can diagnose a
+# stuck lease. Only the apply path demands CLEANUP_TASK + REQUIRED_ROLE.
+ASSESSMENT_PERMISSION = "gitea.read"
+
+_DRY_RUN_LEDGER: list[dict[str, Any]] = []
+
+
+def _norm(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _norm_comment_id(value: Any) -> int | None:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def record_dry_run(
+    *,
+    pr_number: int,
+    repository_slug: str | None,
+    lease_moot: bool,
+    cleanup_allowed: bool,
+    session_id: str | None,
+    candidate_head: str | None,
+    lease_comment_id: Any,
+    recorded_at: datetime | None = None,
+) -> dict[str, Any]:
+    """Append one read-only assessment to the dry-run ledger.
+
+    Never rewrites or removes a prior entry: repeated dry runs accumulate and
+    ``latest_dry_run`` returns the newest matching one.
+    """
+    entry = {
+        "task": CLEANUP_TASK,
+        "pr_number": int(pr_number),
+        "repository_slug": _norm(repository_slug) or None,
+        "lease_moot": bool(lease_moot),
+        "cleanup_allowed": bool(cleanup_allowed),
+        "session_id": _norm(session_id) or None,
+        "candidate_head": _norm(candidate_head) or None,
+        "lease_comment_id": _norm_comment_id(lease_comment_id),
+        "recorded_at": (recorded_at or datetime.now(timezone.utc)).isoformat(),
+    }
+    _DRY_RUN_LEDGER.append(entry)
+    return dict(entry)
+
+
+def dry_run_history() -> tuple[dict[str, Any], ...]:
+    """Immutable view of every recorded dry run, oldest first."""
+    return tuple(dict(entry) for entry in _DRY_RUN_LEDGER)
+
+
+def latest_dry_run(
+    *, pr_number: int, repository_slug: str | None
+) -> dict[str, Any] | None:
+    """Newest dry-run evidence for this repository + PR, or None."""
+    wanted_repo = _norm(repository_slug)
+    for entry in reversed(_DRY_RUN_LEDGER):
+        if entry["pr_number"] != int(pr_number):
+            continue
+        if _norm(entry.get("repository_slug")) != wanted_repo:
+            continue
+        return dict(entry)
+    return None
+
+
+def _reset_for_testing() -> None:
+    """Drop ledger state between tests. Never called by production paths."""
+    _DRY_RUN_LEDGER.clear()
+
+
+def assess_apply_authorization(
+    *,
+    pr_number: int,
+    repository_slug: str | None,
+    resolved_task: str | None,
+    active_role_kind: str | None,
+    assessment: dict[str, Any],
+    evidence: dict[str, Any] | None,
+    expected_session_id: str | None = None,
+    expected_candidate_head: str | None = None,
+    expected_lease_comment_id: Any = None,
+) -> dict[str, Any]:
+    """Decide whether a moot-lease cleanup apply is authorized (fail closed).
+
+    Returns ``{"allowed", "reasons", "blocker_kind", "evidence_matched", ...}``.
+    ``allowed`` is True only when every check passes; each failure contributes a
+    reason so the caller can report all of them together.
+    """
+    reasons: list[str] = []
+    blocker_kind: str | None = None
+
+    def _block(kind: str, reason: str) -> None:
+        nonlocal blocker_kind
+        reasons.append(reason)
+        if blocker_kind is None:
+            blocker_kind = kind
+
+    # 1. Exact resolved cleanup task. Resolving any other task — including a
+    #    sibling reconciler task — does not authorize this mutation.
+    if _norm(resolved_task) != CLEANUP_TASK:
+        _block(
+            "unresolved_cleanup_task",
+            "post-merge moot-lease cleanup requires the session to resolve "
+            f"task '{CLEANUP_TASK}' immediately before apply; resolved task is "
+            f"{resolved_task!r} (fail closed)",
+        )
+
+    # 2. Dedicated reconciler role, enforced independently of the permission.
+    if _norm(active_role_kind) != REQUIRED_ROLE:
+        _block(
+            "wrong_role",
+            f"profile role {active_role_kind!r} cannot apply post-merge "
+            f"moot-lease cleanup; required role is {REQUIRED_ROLE} even when "
+            f"{REQUIRED_PERMISSION} is present (fail closed)",
+        )
+
+    # 3. Canonical repository identity must be established, never inferred from
+    #    request parameters.
+    if not _norm(repository_slug):
+        _block(
+            "repository_binding",
+            "no canonical repository identity could be established for the "
+            "cleanup target (fail closed)",
+        )
+
+    # 4. The live safety assessment must still say the lease is moot/cleanable.
+    if not assessment.get("is_moot") or not assessment.get("cleanup_allowed"):
+        _block(
+            "lease_not_moot",
+            "live assessment does not report a moot, cleanable lease on PR "
+            f"#{pr_number} (lease_moot={bool(assessment.get('is_moot'))}, "
+            f"cleanup_allowed={bool(assessment.get('cleanup_allowed'))}) "
+            "(fail closed)",
+        )
+
+    live = assessment.get("active_lease") or {}
+    live_session = _norm(live.get("session_id"))
+    live_head = _norm(live.get("candidate_head"))
+    live_comment_id = _norm_comment_id(live.get("comment_id"))
+
+    # 5. A lease missing identifying fields is malformed and unsafe to act on.
+    if not live_session or not live_head or live_comment_id is None:
+        _block(
+            "malformed_lease",
+            "active lease is malformed: session_id / candidate_head / "
+            "comment_id must all be present to authorize cleanup "
+            f"(session_id={live.get('session_id')!r}, "
+            f"candidate_head={live.get('candidate_head')!r}, "
+            f"comment_id={live.get('comment_id')!r}) (fail closed)",
+        )
+
+    # 6. Caller expectations, when supplied, must match the live lease exactly.
+    if expected_session_id is not None and _norm(expected_session_id) != live_session:
+        _block(
+            "lease_mismatch",
+            f"expected lease session {expected_session_id!r} does not match the "
+            f"live lease session {live.get('session_id')!r} (fail closed)",
+        )
+    if (
+        expected_candidate_head is not None
+        and _norm(expected_candidate_head) != live_head
+    ):
+        _block(
+            "lease_mismatch",
+            f"expected candidate head {expected_candidate_head!r} does not "
+            f"match the live lease head {live.get('candidate_head')!r} "
+            "(fail closed)",
+        )
+    if expected_lease_comment_id is not None and (
+        _norm_comment_id(expected_lease_comment_id) != live_comment_id
+    ):
+        _block(
+            "lease_mismatch",
+            f"expected lease marker {expected_lease_comment_id!r} does not "
+            f"match the live lease marker {live.get('comment_id')!r} "
+            "(fail closed)",
+        )
+
+    # 7. Matching dry-run evidence recorded earlier in this session.
+    evidence_matched = False
+    if evidence is None:
+        _block(
+            "missing_dry_run_evidence",
+            "no read-only dry run recorded for this repository and PR; run the "
+            "tool with apply=false and confirm lease_moot / cleanup_allowed "
+            "before applying (fail closed)",
+        )
+    elif not evidence.get("lease_moot") or not evidence.get("cleanup_allowed"):
+        _block(
+            "dry_run_not_allowed",
+            "recorded dry run did not report an allowed cleanup "
+            f"(lease_moot={bool(evidence.get('lease_moot'))}, "
+            f"cleanup_allowed={bool(evidence.get('cleanup_allowed'))}) "
+            "(fail closed)",
+        )
+    elif int(evidence.get("pr_number") or -1) != int(pr_number) or _norm(
+        evidence.get("repository_slug")
+    ) != _norm(repository_slug):
+        _block(
+            "dry_run_mismatch",
+            "recorded dry run targets a different repository or PR "
+            f"({evidence.get('repository_slug')}#{evidence.get('pr_number')} vs "
+            f"{repository_slug}#{pr_number}) (fail closed)",
+        )
+    elif (
+        _norm(evidence.get("session_id")) != live_session
+        or _norm(evidence.get("candidate_head")) != live_head
+        or _norm_comment_id(evidence.get("lease_comment_id")) != live_comment_id
+    ):
+        _block(
+            "superseded_lease",
+            "the lease changed after the recorded dry run (dry run: "
+            f"session={evidence.get('session_id')!r}, "
+            f"head={evidence.get('candidate_head')!r}, "
+            f"marker={evidence.get('lease_comment_id')!r}; live: "
+            f"session={live.get('session_id')!r}, "
+            f"head={live.get('candidate_head')!r}, "
+            f"marker={live.get('comment_id')!r}); re-run the dry run "
+            "(fail closed)",
+        )
+    else:
+        evidence_matched = True
+
+    return {
+        "allowed": not reasons,
+        "reasons": reasons,
+        "blocker_kind": blocker_kind,
+        "evidence_matched": evidence_matched,
+        "required_task": CLEANUP_TASK,
+        "required_role_kind": REQUIRED_ROLE,
+        "required_permission": REQUIRED_PERMISSION,
+    }

role_session_router.py

@@ -87,6 +87,11 @@ RECONCILER_TASKS = frozenset({
     # only to the reconciler profile). Raw gitea_delete_branch redirects here to
     # the guarded gitea_cleanup_merged_pr_branch path (#514/#687).
     "delete_branch",
+    # #745: post-merge moot reviewer-lease cleanup is reconciler-owned; the
+    # apply path posts a terminal lease marker. Kept in step with
+    # task_capability_map so map and router cannot disagree (#723 defect A).
+    "cleanup_post_merge_moot_lease",
+    "gitea_cleanup_post_merge_moot_lease",
     "reconcile_already_landed_pr",
     "reconcile_already_landed",
     "reconcile-landed-pr",
@@ -118,6 +123,9 @@ TASK_REQUIRED_ROLE = {
     "reconcile_already_landed": "reconciler",
     "reconcile-landed-pr": "reconciler",
     "cleanup_merged_pr_branch": "reconciler",
+    # #745: post-merge moot reviewer-lease cleanup (canonical task + tool alias).
+    "cleanup_post_merge_moot_lease": "reconciler",
+    "gitea_cleanup_post_merge_moot_lease": "reconciler",
     # #309: reconciler tasks close already-landed PRs/issues only.
     "reconcile_close_landed_pr": "reconciler",
     "reconcile_close_landed_issue": "reconciler",

task_capability_map.py

@@ -158,6 +158,23 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.comment",
         "role": "reviewer",
     },
+    # #745: post-merge moot reviewer-lease cleanup is reconciler-owned. The
+    # apply path posts an append-only terminal `phase: released` lease marker
+    # (gitea.pr.comment), so holding the comment permission alone must not
+    # authorize it — author, reviewer and merger fail closed on the role gate
+    # even though their profiles carry gitea.pr.comment. The read-only
+    # `apply=false` assessment deliberately stays reachable under gitea.read
+    # inside the tool (the same convention as cleanup_stale_review_decision_lock
+    # below), so any namespace can diagnose a stuck lease; only apply requires
+    # this task plus the reconciler role.
+    "cleanup_post_merge_moot_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "reconciler",
+    },
+    "gitea_cleanup_post_merge_moot_lease": {
+        "permission": "gitea.pr.comment",
+        "role": "reconciler",
+    },
     "blind_pr_queue_review": {
         "permission": "gitea.pr.review",
         "role": "reviewer",

tests/test_issue_745_moot_lease_reconciler_gate.py

@@ -0,0 +1,658 @@
+"""Reconciler role binding for post-merge moot-lease cleanup (#745).
+
+``gitea_cleanup_post_merge_moot_lease`` (#515) posts a terminal ``phase:
+released`` lease marker but had no capability-map entry and no role gate: entry
+required only ``gitea.read``, apply required only ``gitea.pr.comment``, so any
+profile holding the comment permission reached the mutation while the
+reconciler could not satisfy resolve-exact-task -> mutation.
+
+These tests pin the fixed contract:
+
+- the canonical task and its tool-name alias resolve identically
+  (``gitea.pr.comment`` + ``reconciler``);
+- only a reconciler profile satisfies permission AND role;
+- ``apply=false`` assessment stays reachable under ``gitea.read`` for any role
+  and mutates nothing (documented, deliberate divergence from the apply path);
+- ``apply=true`` requires the exact resolved task, the reconciler role, the
+  comment permission, a validated repository binding and matching dry-run
+  evidence;
+- live/non-moot/superseded/mismatched/malformed leases fail closed;
+- the dry-run ledger is append-only and cleanup stays idempotent.
+
+Every fixture is synthetic. No production PR, lease session or marker is used
+anywhere in this module (see ``TestNoProductionLeaseTouched``).
+"""
+
+import sys as _sys
+from pathlib import Path as _Path
+
+_sys.path.insert(0, str(_Path(__file__).resolve().parent.parent))
+
+import os  # noqa: E402
+import unittest  # noqa: E402
+from datetime import datetime, timezone  # noqa: E402
+from unittest.mock import patch  # noqa: E402
+
+import mcp_server  # noqa: E402
+import post_merge_moot_lease_gate as gate  # noqa: E402
+import reviewer_pr_lease as leases  # noqa: E402
+from mcp_server import gitea_cleanup_post_merge_moot_lease  # noqa: E402
+from role_session_router import RECONCILER_TASKS, TASK_REQUIRED_ROLE  # noqa: E402
+from task_capability_map import required_permission, required_role  # noqa: E402
+
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
+PR = 487
+ISSUE = 485
+SESSION = "97274-676d20a825c4"
+HEAD_A = "a" * 40
+HEAD_B = "d" * 40
+LEASE_COMMENT_ID = 6603
+
+CANONICAL_TASK = "cleanup_post_merge_moot_lease"
+TOOL_ALIAS = "gitea_cleanup_post_merge_moot_lease"
+
+_BASE_OPS = "gitea.read,gitea.pr.comment"
+RECONCILER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-reconciler",
+    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.close,gitea.branch.delete",
+}
+AUTHOR_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-author",
+    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.create,gitea.issue.create",
+}
+REVIEWER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-reviewer",
+    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.review,gitea.pr.approve",
+}
+MERGER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-merger",
+    "GITEA_ALLOWED_OPERATIONS": _BASE_OPS + ",gitea.pr.merge",
+}
+
+# Permission shape of the configured role profiles (mirrors
+# tests/test_task_capability_role_invariants.py CANONICAL_ROLE_PROFILES).
+ROLE_PROFILE_PERMISSIONS = {
+    "author": {"gitea.read", "gitea.pr.comment", "gitea.pr.create",
+               "gitea.issue.create", "gitea.issue.comment", "gitea.issue.close",
+               "gitea.branch.create", "gitea.branch.push", "gitea.repo.commit"},
+    "reviewer": {"gitea.read", "gitea.pr.comment", "gitea.pr.review",
+                 "gitea.pr.approve", "gitea.pr.request_changes",
+                 "gitea.issue.comment"},
+    "merger": {"gitea.read", "gitea.pr.comment", "gitea.pr.merge",
+               "gitea.issue.comment"},
+    "reconciler": {"gitea.read", "gitea.pr.comment", "gitea.pr.close",
+                   "gitea.issue.close", "gitea.branch.delete"},
+}
+
+
+def _lease_comment(pr_number=PR, session_id=SESSION, *, phase="claimed",
+                   candidate_head=HEAD_A, comment_id=LEASE_COMMENT_ID):
+    body = leases.format_lease_body(
+        repo=SLUG,
+        pr_number=pr_number,
+        issue_number=ISSUE,
+        reviewer_identity="sysadmin",
+        profile="prgs-reviewer",
+        session_id=session_id,
+        worktree="branches/review-pr487",
+        phase=phase,
+        candidate_head=candidate_head,
+        target_branch="master",
+        target_branch_sha="b" * 40,
+        last_activity=datetime.now(timezone.utc),
+    )
+    return {"id": comment_id, "body": body, "user": {"login": "sysadmin"}}
+
+
+def _api_side_effect(*, pr_state, pr_merged, comments, posted_id=9999):
+    """api_request side effect keyed on method + url; records POSTs."""
+    calls = {"post": []}
+
+    def _side(method, url, auth=None, payload=None, *a, **k):
+        if (method or "").upper() == "POST":
+            calls["post"].append({"url": url, "payload": payload})
+            return {"id": posted_id}
+        if "/comments" in url:
+            return list(comments)
+        if "/pulls/" in url:
+            pr = {"state": pr_state, "number": PR, "merge_commit_sha": "c" * 40}
+            if pr_merged:
+                pr["merged"] = True
+                pr["merged_at"] = "2026-07-08T07:46:04Z"
+            return pr
+        if "/issues/" in url:
+            return {"state": "closed" if pr_merged else "open", "number": ISSUE}
+        return {}
+
+    return _side, calls
+
+
+def _assessment(comments, *, pr_merged=True, pr_state="closed"):
+    return leases.assess_post_merge_moot_lease(
+        comments, pr_number=PR, pr_merged=pr_merged, pr_state=pr_state,
+        merge_commit_sha="c" * 40,
+    )
+
+
+# --------------------------------------------------------------------------- #
+# 1-5. Capability map / router contract
+# --------------------------------------------------------------------------- #
+class TestCleanupTaskContract(unittest.TestCase):
+    def test_canonical_task_is_reconciler_owned(self):
+        """1. The reconciler is the role that can resolve the cleanup task."""
+        self.assertEqual(required_permission(CANONICAL_TASK), "gitea.pr.comment")
+        self.assertEqual(required_role(CANONICAL_TASK), "reconciler")
+
+    def test_tool_alias_resolves_to_identical_contract(self):
+        """5. Alias and canonical task must not diverge."""
+        self.assertEqual(
+            (required_permission(TOOL_ALIAS), required_role(TOOL_ALIAS)),
+            (required_permission(CANONICAL_TASK), required_role(CANONICAL_TASK)),
+        )
+
+    def test_author_reviewer_merger_cannot_resolve_the_task(self):
+        """2-4. No non-reconciler role satisfies permission AND role."""
+        for role in ("author", "reviewer", "merger"):
+            with self.subTest(role=role):
+                self.assertNotEqual(required_role(CANONICAL_TASK), role)
+                # They hold the permission — which is exactly why the role gate
+                # is required rather than optional.
+                self.assertIn(
+                    "gitea.pr.comment", ROLE_PROFILE_PERMISSIONS[role],
+                    "test premise: non-reconciler roles do hold pr.comment",
+                )
+
+    def test_reconciler_profile_satisfies_permission_and_role(self):
+        self.assertIn(
+            required_permission(CANONICAL_TASK),
+            ROLE_PROFILE_PERMISSIONS[required_role(CANONICAL_TASK)],
+        )
+
+    def test_router_agrees_with_capability_map(self):
+        for task in (CANONICAL_TASK, TOOL_ALIAS):
+            with self.subTest(task=task):
+                self.assertIn(task, RECONCILER_TASKS)
+                self.assertEqual(TASK_REQUIRED_ROLE[task], required_role(task))
+
+    def test_unknown_alias_still_rejected(self):
+        for bogus in ("cleanup_post_merge_moot_leases", "cleanup_moot_lease", ""):
+            with self.subTest(task=bogus):
+                with self.assertRaises(KeyError):
+                    required_role(bogus)
+                with self.assertRaises(KeyError):
+                    required_permission(bogus)
+
+
+# --------------------------------------------------------------------------- #
+# Authorization gate unit tests
+# --------------------------------------------------------------------------- #
+class TestApplyAuthorizationGate(unittest.TestCase):
+    def setUp(self):
+        gate._reset_for_testing()
+        self.addCleanup(gate._reset_for_testing)
+        self.assessment = _assessment([_lease_comment()])
+
+    def _evidence(self, **over):
+        base = dict(
+            pr_number=PR, repository_slug=SLUG, lease_moot=True,
+            cleanup_allowed=True, session_id=SESSION, candidate_head=HEAD_A,
+            lease_comment_id=LEASE_COMMENT_ID,
+        )
+        base.update(over)
+        return gate.record_dry_run(**base)
+
+    def _assess(self, **over):
+        kwargs = dict(
+            pr_number=PR, repository_slug=SLUG, resolved_task=CANONICAL_TASK,
+            active_role_kind="reconciler", assessment=self.assessment,
+            evidence=self._evidence(),
+        )
+        kwargs.update(over)
+        return gate.assess_apply_authorization(**kwargs)
+
+    def test_reconciler_with_matching_evidence_is_authorized(self):
+        result = self._assess()
+        self.assertTrue(result["allowed"], result["reasons"])
+        self.assertTrue(result["evidence_matched"])
+
+    def test_apply_without_exact_task_resolution_fails(self):
+        """8. Apply without exact task resolution fails preflight."""
+        result = self._assess(resolved_task=None)
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
+
+    def test_resolving_another_task_does_not_authorize_cleanup(self):
+        """9. A sibling reconciler task is not a substitute."""
+        for other in ("delete_branch", "reconcile_already_landed_pr",
+                      "cleanup_merged_pr_branch", "comment_pr"):
+            with self.subTest(task=other):
+                result = self._assess(resolved_task=other)
+                self.assertFalse(result["allowed"])
+                self.assertEqual(
+                    result["blocker_kind"], "unresolved_cleanup_task")
+
+    def test_non_reconciler_roles_fail_closed(self):
+        """10. Author/reviewer/merger cannot apply despite pr.comment."""
+        for role in ("author", "reviewer", "merger", None, ""):
+            with self.subTest(role=role):
+                result = self._assess(active_role_kind=role)
+                self.assertFalse(result["allowed"])
+                self.assertEqual(result["blocker_kind"], "wrong_role")
+
+    def test_missing_repository_identity_fails_closed(self):
+        result = self._assess(repository_slug=None)
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "repository_binding")
+
+    def test_missing_dry_run_evidence_fails_closed(self):
+        result = self._assess(evidence=None)
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence")
+
+    def test_dry_run_that_disallowed_cleanup_fails_closed(self):
+        result = self._assess(evidence=self._evidence(cleanup_allowed=False))
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "dry_run_not_allowed")
+
+    def test_dry_run_for_another_pr_or_repo_fails_closed(self):
+        """11. Wrong PR / repository fails closed."""
+        for over in ({"pr_number": PR + 1}, {"repository_slug": "Other/Repo"}):
+            with self.subTest(**over):
+                result = self._assess(evidence=self._evidence(**over))
+                self.assertFalse(result["allowed"])
+                self.assertEqual(result["blocker_kind"], "dry_run_mismatch")
+
+    def test_superseded_lease_fails_closed(self):
+        """12. Head/session/marker drift since the dry run fails closed."""
+        for over in ({"session_id": "other-session"},
+                     {"candidate_head": HEAD_B},
+                     {"lease_comment_id": 7777}):
+            with self.subTest(**over):
+                result = self._assess(evidence=self._evidence(**over))
+                self.assertFalse(result["allowed"])
+                self.assertEqual(result["blocker_kind"], "superseded_lease")
+
+    def test_expectation_mismatch_fails_closed(self):
+        """11. Wrong session / head / marker expectations fail closed."""
+        for over in ({"expected_session_id": "nope"},
+                     {"expected_candidate_head": HEAD_B},
+                     {"expected_lease_comment_id": 7777}):
+            with self.subTest(**over):
+                result = self._assess(**over)
+                self.assertFalse(result["allowed"])
+                self.assertEqual(result["blocker_kind"], "lease_mismatch")
+
+    def test_matching_expectations_are_authorized(self):
+        result = self._assess(
+            expected_session_id=SESSION,
+            expected_candidate_head=HEAD_A,
+            expected_lease_comment_id=LEASE_COMMENT_ID,
+        )
+        self.assertTrue(result["allowed"], result["reasons"])
+
+    def test_non_moot_lease_fails_closed(self):
+        """12. A live lease on an open PR is never cleanable."""
+        open_pr = _assessment(
+            [_lease_comment()], pr_merged=False, pr_state="open")
+        result = self._assess(assessment=open_pr)
+        self.assertFalse(result["allowed"])
+        self.assertIn(
+            result["blocker_kind"], ("lease_not_moot", "malformed_lease"))
+
+    def test_malformed_lease_fails_closed(self):
+        malformed = dict(self.assessment)
+        malformed["active_lease"] = {
+            "session_id": "", "candidate_head": None, "comment_id": None}
+        result = self._assess(assessment=malformed)
+        self.assertFalse(result["allowed"])
+        self.assertEqual(result["blocker_kind"], "malformed_lease")
+
+    def test_ledger_is_append_only(self):
+        """14. Recording never rewrites or drops prior entries."""
+        first = self._evidence()
+        second = self._evidence(candidate_head=HEAD_B, lease_comment_id=7777)
+        history = gate.dry_run_history()
+        self.assertEqual(len(history), 2)
+        self.assertEqual(history[0]["candidate_head"], first["candidate_head"])
+        self.assertEqual(history[1]["candidate_head"], HEAD_B)
+        # Newest-wins for lookup, but the older entry survives in history.
+        latest = gate.latest_dry_run(pr_number=PR, repository_slug=SLUG)
+        self.assertEqual(latest["lease_comment_id"], second["lease_comment_id"])
+        self.assertEqual(gate.dry_run_history()[0]["lease_comment_id"],
+                         LEASE_COMMENT_ID)
+
+    def test_history_view_cannot_mutate_the_ledger(self):
+        self._evidence()
+        snapshot = gate.dry_run_history()
+        snapshot[0]["pr_number"] = 999999
+        self.assertEqual(
+            gate.dry_run_history()[0]["pr_number"], PR,
+            "dry_run_history must hand out copies, not live rows")
+
+
+# --------------------------------------------------------------------------- #
+# Tool-level behavior
+# --------------------------------------------------------------------------- #
+class _ToolCase(unittest.TestCase):
+    def setUp(self):
+        leases.clear_session_lease()
+        gate._reset_for_testing()
+        self.addCleanup(gate._reset_for_testing)
+
+    def _run(self, env, *, apply, comments, pr_state="closed", pr_merged=True,
+             resolved_task=CANONICAL_TASK, slug=SLUG, **kwargs):
+        side, calls = _api_side_effect(
+            pr_state=pr_state, pr_merged=pr_merged, comments=comments)
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", return_value=None), \
+                patch("mcp_server._bound_repository_slug", return_value=slug), \
+                patch("mcp_server._repository_binding_block", return_value=None), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             resolved_task), \
+                patch.dict(os.environ, env, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=apply, remote="prgs", **kwargs)
+        return result, calls
+
+
+class TestDryRunOpenToEveryRole(_ToolCase):
+    """7. Dry run performs no mutation and stays under the read capability."""
+
+    def test_dry_run_reports_moot_and_mutates_nothing_for_every_role(self):
+        for name, env in (("reconciler", RECONCILER_ENV), ("author", AUTHOR_ENV),
+                          ("reviewer", REVIEWER_ENV), ("merger", MERGER_ENV)):
+            with self.subTest(role=name):
+                gate._reset_for_testing()
+                result, calls = self._run(
+                    env, apply=False, comments=[_lease_comment()],
+                    resolved_task=None)
+                self.assertTrue(result["success"])
+                self.assertTrue(result["lease_moot"])
+                self.assertTrue(result["cleanup_allowed"])
+                self.assertFalse(result["cleanup_performed"])
+                self.assertEqual(result["mode"], "read_only")
+                self.assertEqual(calls["post"], [], "dry run must not mutate")
+
+    def test_dry_run_records_evidence(self):
+        result, _ = self._run(
+            RECONCILER_ENV, apply=False, comments=[_lease_comment()])
+        evidence = result["dry_run_evidence"]
+        self.assertEqual(evidence["pr_number"], PR)
+        self.assertEqual(evidence["repository_slug"], SLUG)
+        self.assertEqual(evidence["session_id"], SESSION)
+        self.assertEqual(evidence["candidate_head"], HEAD_A)
+        self.assertEqual(evidence["lease_comment_id"], LEASE_COMMENT_ID)
+        self.assertTrue(evidence["lease_moot"])
+        self.assertTrue(evidence["cleanup_allowed"])
+
+
+class TestApplyRequiresReconciler(_ToolCase):
+    def test_reconciler_apply_succeeds_after_matching_dry_run(self):
+        """7 (apply). Allowed dry run then apply posts exactly one marker."""
+        comments = [_lease_comment()]
+        dry, dry_calls = self._run(
+            RECONCILER_ENV, apply=False, comments=comments)
+        self.assertTrue(dry["cleanup_allowed"])
+        self.assertEqual(dry_calls["post"], [])
+
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=comments)
+        self.assertTrue(result["success"], result.get("reasons"))
+        self.assertTrue(result["cleanup_performed"])
+        self.assertEqual(result["released_comment_id"], 9999)
+        self.assertEqual(len(calls["post"]), 1)
+        body = calls["post"][0]["payload"]["body"]
+        self.assertIn("phase: released", body)
+        self.assertIn("post-merge-moot", body)
+
+    def test_apply_without_dry_run_fails_closed(self):
+        """6. Apply must be preceded by a matching dry run."""
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=[_lease_comment()])
+        self.assertFalse(result["success"])
+        self.assertFalse(result["cleanup_performed"])
+        self.assertEqual(result["blocker_kind"], "missing_dry_run_evidence")
+        self.assertEqual(calls["post"], [])
+
+    def test_apply_without_exact_task_resolution_fails_closed(self):
+        """8. No resolved cleanup task -> no mutation."""
+        comments = [_lease_comment()]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=comments, resolved_task=None)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
+        self.assertEqual(calls["post"], [])
+
+    def test_resolving_a_different_task_does_not_authorize_apply(self):
+        """9. Another resolved task is not a substitute."""
+        comments = [_lease_comment()]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=comments,
+            resolved_task="delete_branch")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "unresolved_cleanup_task")
+        self.assertEqual(calls["post"], [])
+
+    def test_author_reviewer_merger_cannot_apply(self):
+        """10. Permission-only roles are refused at the role gate."""
+        for name, env in (("author", AUTHOR_ENV), ("reviewer", REVIEWER_ENV),
+                          ("merger", MERGER_ENV)):
+            with self.subTest(role=name):
+                gate._reset_for_testing()
+                comments = [_lease_comment()]
+                self._run(env, apply=False, comments=comments)
+                result, calls = self._run(env, apply=True, comments=comments)
+                self.assertFalse(result["success"])
+                self.assertFalse(result["cleanup_performed"])
+                self.assertEqual(result["blocker_kind"], "wrong_role")
+                self.assertEqual(
+                    calls["post"], [],
+                    f"{name} must not post a terminal lease marker")
+
+
+class TestApplyFailsClosedOnLeaseState(_ToolCase):
+    def test_open_pr_lease_is_never_force_cleaned(self):
+        """12. Non-moot: an active lease on an open PR stays untouched."""
+        comments = [_lease_comment()]
+        result, calls = self._run(
+            RECONCILER_ENV, apply=True, comments=comments,
+            pr_state="open", pr_merged=False)
+        self.assertFalse(result["cleanup_performed"])
+        self.assertFalse(result["pr_merged_or_closed"])
+        self.assertEqual(calls["post"], [], "never force-clean an open PR lease")
+        self.assertTrue(any(
+            "still open" in r for r in result.get("cleanup_skipped_reason", [])))
+
+    def test_superseded_lease_between_dry_run_and_apply_fails_closed(self):
+        """12. The lease moved on after the dry run -> refuse."""
+        self._run(RECONCILER_ENV, apply=False, comments=[_lease_comment()])
+        moved = [_lease_comment(session_id="fresh-session",
+                                candidate_head=HEAD_B, comment_id=7777)]
+        result, calls = self._run(RECONCILER_ENV, apply=True, comments=moved)
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "superseded_lease")
+        self.assertEqual(calls["post"], [])
+
+    def test_expectation_mismatch_fails_closed(self):
+        """11. Wrong session / head / marker expectations refuse the apply."""
+        comments = [_lease_comment()]
+        for kwargs in ({"expected_session_id": "wrong-session"},
+                       {"expected_candidate_head": HEAD_B},
+                       {"expected_lease_comment_id": 7777}):
+            with self.subTest(**kwargs):
+                gate._reset_for_testing()
+                self._run(RECONCILER_ENV, apply=False, comments=comments)
+                result, calls = self._run(
+                    RECONCILER_ENV, apply=True, comments=comments, **kwargs)
+                self.assertFalse(result["success"])
+                self.assertEqual(result["blocker_kind"], "lease_mismatch")
+                self.assertEqual(calls["post"], [])
+
+    def test_already_terminal_cleanup_is_idempotent(self):
+        """13. A released lease reports nothing to clean and posts nothing."""
+        first = _assessment([_lease_comment()])
+        released = {"id": 7000, "body": first["release_body"],
+                    "user": {"login": "sysadmin"}}
+        comments = [_lease_comment(), released]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        result, calls = self._run(RECONCILER_ENV, apply=True, comments=comments)
+        self.assertFalse(result["cleanup_performed"])
+        self.assertFalse(result["lease_moot"])
+        self.assertEqual(calls["post"], [], "no second terminal marker")
+        self.assertTrue(any(
+            "already released/terminal" in r
+            for r in result.get("cleanup_skipped_reason", [])))
+
+    def test_apply_is_append_only_never_deletes(self):
+        """14. The only write is a POST; nothing is edited or deleted."""
+        comments = [_lease_comment()]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        side, _calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=comments)
+        seen = []
+
+        def _recording(method, url, auth=None, payload=None, *a, **k):
+            seen.append((method or "").upper())
+            return side(method, url, auth, payload, *a, **k)
+
+        with patch("mcp_server.api_request", side_effect=_recording), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", return_value=None), \
+                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
+                patch("mcp_server._repository_binding_block", return_value=None), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             CANONICAL_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs")
+        self.assertTrue(result["cleanup_performed"])
+        self.assertNotIn("DELETE", seen)
+        self.assertNotIn("PATCH", seen)
+        self.assertNotIn("PUT", seen)
+        self.assertEqual(seen.count("POST"), 1)
+
+
+class TestRepositoryBinding(_ToolCase):
+    """11. Foreign-repository targets fail closed before any mutation."""
+
+    def test_explicit_foreign_repository_is_rejected(self):
+        comments = [_lease_comment()]
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=comments)
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", return_value=None), \
+                patch("mcp_server._canonical_repository_slug",
+                      return_value=(None, [])), \
+                patch("mcp_server._workspace_repository_slug",
+                      return_value=SLUG), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             CANONICAL_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs",
+                org="Some-Other-Org", repo="Some-Other-Repo")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "repository_binding")
+        self.assertEqual(calls["post"], [])
+
+    def test_unresolvable_canonical_root_fails_closed(self):
+        comments = [_lease_comment()]
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=comments)
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", return_value=None), \
+                patch("mcp_server._canonical_repository_slug",
+                      return_value=(None, ["canonical root unresolvable"])), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             CANONICAL_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertEqual(result["blocker_kind"], "repository_binding")
+        self.assertEqual(calls["post"], [])
+
+
+class TestPreflightBinding(_ToolCase):
+    """4. The apply path binds the shared preflight to the exact task."""
+
+    def test_apply_forwards_task_and_target_to_preflight(self):
+        comments = [_lease_comment()]
+        self._run(RECONCILER_ENV, apply=False, comments=comments)
+        side, _calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=comments)
+        seen = {}
+
+        def _purity(remote=None, worktree_path=None, task=None, **kw):
+            seen.update({"remote": remote, "worktree_path": worktree_path,
+                         "task": task, **kw})
+            return None
+
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", _purity), \
+                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
+                patch("mcp_server._repository_binding_block", return_value=None), \
+                patch.object(mcp_server, "_preflight_resolved_task",
+                             CANONICAL_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs",
+                worktree_path="/tmp/branches/reconciler-745",
+                org="Scaled-Tech-Consulting", repo="Gitea-Tools")
+        self.assertTrue(result["cleanup_performed"])
+        self.assertEqual(seen["task"], CANONICAL_TASK)
+        self.assertEqual(seen["worktree_path"], "/tmp/branches/reconciler-745")
+        self.assertEqual(seen["org"], "Scaled-Tech-Consulting")
+        self.assertEqual(seen["repo"], "Gitea-Tools")
+
+    def test_dry_run_does_not_require_preflight(self):
+        def _boom(*a, **k):
+            raise AssertionError("dry run must not run mutation preflight")
+
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=[_lease_comment()])
+        with patch("mcp_server.api_request", side_effect=side), \
+                patch("mcp_server.get_auth_header", return_value=FAKE_AUTH), \
+                patch("mcp_server.verify_preflight_purity", _boom), \
+                patch("mcp_server._bound_repository_slug", return_value=SLUG), \
+                patch.dict(os.environ, AUTHOR_ENV, clear=True):
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+        self.assertTrue(result["success"])
+        self.assertEqual(calls["post"], [])
+
+
+class TestNoProductionLeaseTouched(unittest.TestCase):
+    """16. No real production lease or PR is referenced by these tests."""
+
+    PRODUCTION_PR = 744
+    PRODUCTION_SESSION = "33673-1d54887a0415"
+    PRODUCTION_MARKER = 12452
+
+    def test_fixtures_are_synthetic(self):
+        self.assertNotEqual(PR, self.PRODUCTION_PR)
+        self.assertNotEqual(SESSION, self.PRODUCTION_SESSION)
+        self.assertNotEqual(LEASE_COMMENT_ID, self.PRODUCTION_MARKER)
+
+    def test_module_source_never_names_the_production_lease(self):
+        source = _Path(__file__).read_text()
+        for token in (self.PRODUCTION_SESSION, str(self.PRODUCTION_MARKER)):
+            self.assertEqual(
+                source.count(token), 1,
+                f"{token!r} must appear only in this guard's own constants",
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_post_merge_moot_lease.py

@@ -19,6 +19,8 @@ from unittest.mock import patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
+import mcp_server  # noqa: E402
+import post_merge_moot_lease_gate as moot_gate  # noqa: E402
 import reviewer_pr_lease as leases  # noqa: E402
 from mcp_server import (  # noqa: E402
     gitea_acquire_reviewer_pr_lease,
@@ -30,6 +32,13 @@ MERGER_ENV = {
     "GITEA_PROFILE_NAME": "prgs-merger",
     "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment",
 }
+# #745: applying the terminal marker is reconciler-owned.
+RECONCILER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-reconciler",
+    "GITEA_ALLOWED_OPERATIONS": "gitea.read,gitea.pr.comment,gitea.pr.close",
+}
+CLEANUP_TASK = moot_gate.CLEANUP_TASK
+REPO_SLUG = "Scaled-Tech-Consulting/Gitea-Tools"
 PR = 487
 ISSUE = 485
 SESSION = "97274-676d20a825c4"
@@ -204,6 +213,8 @@ class TestAcquireToolRefusesMergedPR(unittest.TestCase):
 class TestCleanupTool(unittest.TestCase):
     def setUp(self):
         leases.clear_session_lease()
+        moot_gate._reset_for_testing()
+        self.addCleanup(moot_gate._reset_for_testing)
 
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")
@@ -223,23 +234,53 @@ class TestCleanupTool(unittest.TestCase):
         self.assertEqual(calls["post"], [])
 
     @patch("mcp_server.verify_preflight_purity", return_value=None)
+    @patch("mcp_server._repository_binding_block", return_value=None)
+    @patch("mcp_server._bound_repository_slug", return_value=REPO_SLUG)
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")
     def test_apply_posts_released_marker_on_merged_pr(
-            self, mock_api, _auth, _purity):
+            self, mock_api, _auth, _slug, _binding, _purity):
+        """#745: apply is reconciler-only and needs a matching dry run first."""
         side, calls = _api_side_effect(
             pr_state="closed", pr_merged=True, comments=[_lease_comment()])
         mock_api.side_effect = side
-        with patch.dict(os.environ, MERGER_ENV, clear=True):
+        with patch.object(mcp_server, "_preflight_resolved_task",
+                          CLEANUP_TASK), \
+                patch.dict(os.environ, RECONCILER_ENV, clear=True):
+            gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
             result = gitea_cleanup_post_merge_moot_lease(
                 pr_number=PR, apply=True, remote="prgs")
-        self.assertTrue(result["success"])
+        self.assertTrue(result["success"], result.get("reasons"))
         self.assertTrue(result["cleanup_performed"])
         self.assertEqual(result["released_comment_id"], 9999)
         self.assertEqual(len(calls["post"]), 1)
         self.assertIn("phase: released", calls["post"][0]["payload"]["body"])
         self.assertIn("post-merge-moot", calls["post"][0]["payload"]["body"])
 
+    @patch("mcp_server.verify_preflight_purity", return_value=None)
+    @patch("mcp_server._repository_binding_block", return_value=None)
+    @patch("mcp_server._bound_repository_slug", return_value=REPO_SLUG)
+    @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
+    @patch("mcp_server.api_request")
+    def test_merger_can_no_longer_apply(
+            self, mock_api, _auth, _slug, _binding, _purity):
+        """#745: holding gitea.pr.comment is no longer sufficient to apply."""
+        side, calls = _api_side_effect(
+            pr_state="closed", pr_merged=True, comments=[_lease_comment()])
+        mock_api.side_effect = side
+        with patch.object(mcp_server, "_preflight_resolved_task",
+                          CLEANUP_TASK), \
+                patch.dict(os.environ, MERGER_ENV, clear=True):
+            gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=False, remote="prgs")
+            result = gitea_cleanup_post_merge_moot_lease(
+                pr_number=PR, apply=True, remote="prgs")
+        self.assertFalse(result["success"])
+        self.assertFalse(result["cleanup_performed"])
+        self.assertEqual(result["blocker_kind"], "wrong_role")
+        self.assertEqual(calls["post"], [])
+
     @patch("mcp_server.verify_preflight_purity", return_value=None)
     @patch("mcp_server.get_auth_header", return_value=FAKE_AUTH)
     @patch("mcp_server.api_request")