Merge pull request 'fix(mcp): bind post-merge moot-lease cleanup to the reconciler capability (Closes #745)' (#746) from fix/issue-745-reconciler-moot-lease-gate into master

#753 added the dead-session author-lock recovery assessor, but no sanctioned
recovery could ever reach the lock write. A dead-session lock is by construction
a lock for work that already has an open PR, and the #400 duplicate-work gate
blocked unconditionally on any linked open PR. gitea_lock_issue computed
recovery_sanctioned, then discarded it one gate later:

    open PR #750 already covers issue #749 (fail closed)

Because the recovered lock was never persisted, it stayed non-live, so
_prove_author_ownership_for_pr found no live author lock and
gitea_update_pr_branch_by_merge failed closed too. Both blockers had a single
cause, so only the first one is fixed here.

issue_lock_recovery.owning_pr_recovery_evidence distils a granted assessment
into the minimum evidence the duplicate gate needs: issue, branch, owning PR
number, and head. It returns None unless the outcome is RECOVERY_SANCTIONED and
the assessment's own evidence agrees that local head == remote head == PR head,
so a partial or hand-built assessment cannot authorize anything.

The duplicate gate takes that evidence and re-checks every element against the
live PR list it was handed: exactly one linked open PR, matching number, head
branch, head SHA, and locked branch. Only that exact self-owned PR is exempt.
A different PR, several linked PRs, a different branch or head, or evidence for
another issue all keep failing closed, with a diagnostic naming which element
disagreed. The competing-branch, claim-lease and commit/push/create_pr arms are
untouched, and with no evidence the gate behaves exactly as before.

Ownership proof needed no change: once the recovered lock persists under the
live session pid, _prove_author_ownership_for_pr matches it as before.

Out of scope: the PHASE_COMMIT/PHASE_PUSH/PHASE_CREATE_PR rechecks still block
on a linked open PR for every author, recovered or not. That is pre-existing
#400 behavior, unrelated to lock recovery, and #755 does not cover it.

Tests
- tests/test_issue_755_owning_pr_recovery.py (new, 31 tests) drives the real
  mcp_server.gitea_lock_issue handler, not only the pure assessor: sanctioned
  recovery relocks, persists a live lease with truthful dead_session_recovery
  provenance, and satisfies _prove_author_ownership_for_pr; competing PR,
  multiple linked PRs, mismatched head/branch/worktree, dirty worktree, live
  prior pid, and a fresh claim with no prior lock all stay blocked.
- Neutralizing owning_pr_recovery_evidence regresses all three success tests to
  the exact production error above, so the coverage is load-bearing.
- Focused suites (755, 753, duplicate gates, lock registration, provenance) —
  102 passed.
- Lock/ownership/worktree/handoff suites — 172 passed, 16 subtests.
- Full suite tests/ — 3529 passed, 6 skipped, 2 failed, 365 subtests.
- The same 2 failures reproduce on a pristine detached worktree at
  08f67007c5 (3498 passed, 6 skipped, 2 failed):
  test_issue_702_review_findings_f1_f6 F1 recovery-before-probe and
  test_reconciler_supersession_close org/repo forwarding. Pre-existing.
- git diff --check clean.

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

create_issue_bootstrap.py

@@ -0,0 +1,227 @@
+"""Sanctioned pre-issue bootstrap for ``create_issue`` (#749).
+
+``gitea_create_issue`` is a pure remote mutation: it creates a tracking issue
+and writes nothing to the local working tree. The issue-first gate forbids
+creating ``branches/issue-<N>-*`` before the issue number exists, while the
+#274 branches-only guard previously demanded that worktree first — a deadlock.
+
+This module defines a **narrow, phase-scoped** exemption:
+
+* Only tasks in :data:`CREATE_ISSUE_TASKS` may use it.
+* Only the **canonical control checkout** may be used (never an arbitrary
+  directory, unrelated worktree, or foreign clone).
+* The control checkout must be clean, on an accepted base branch, and
+  base-equivalent to live master when a remote tip is known.
+* Every post-creation author mutation keeps the ordinary ``branches/`` rule.
+
+The exemption cannot widen: unknown tasks, dirty roots, drifted HEADs, non-base
+branches, and non-control workspaces fall through to the existing fail-closed
+guards.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any
+
+from author_mutation_worktree import BASE_BRANCHES, is_path_under_branches
+from reviewer_worktree import parse_dirty_tracked_files
+
+CREATE_ISSUE_TASKS = frozenset({"create_issue", "gitea_create_issue"})
+
+# Satisfiable before an issue number exists — never names issue-<N>.
+EXACT_NEXT_ACTION_BOOTSTRAP = (
+    "Restore the canonical control checkout to a clean accepted base branch "
+    "(master/main/dev) that matches live master, with no tracked local edits "
+    "and no detached HEAD. Re-resolve the exact create_issue task, then re-run "
+    "gitea_create_issue from that clean control checkout. Do not create "
+    "branches/issue-<N>-* worktrees, dummy directories, or borrow unrelated "
+    "worktrees before the issue exists."
+)
+
+EXACT_NEXT_ACTION_POST_CREATE = (
+    "After the issue exists: create a registered worktree under "
+    "branches/issue-<N>-* from clean master, claim/lock the issue, set "
+    "GITEA_AUTHOR_WORKTREE / worktree_path to that path, then continue author "
+    "mutations from the issue-backed worktree only."
+)
+
+
+def is_create_issue_task(task: str | None) -> bool:
+    """True when *task* is the create_issue mutation (or tool alias)."""
+    return (task or "").strip() in CREATE_ISSUE_TASKS
+
+
+def assess_create_issue_bootstrap(
+    *,
+    workspace_path: str,
+    canonical_repo_root: str,
+    current_branch: str | None = None,
+    head_sha: str | None = None,
+    porcelain_status: str = "",
+    remote_master_sha: str | None = None,
+    task: str | None = None,
+) -> dict[str, Any]:
+    """Assess whether create_issue may proceed from the control checkout.
+
+    Returns a structured assessment:
+
+    * ``not_applicable`` — not a create_issue task, or workspace is already a
+      ``branches/`` worktree (use ordinary guards).
+    * ``allowed`` — create_issue bootstrap may proceed from this control root.
+    * ``block`` — create_issue was attempted from control checkout but gates
+      failed (dirty, wrong branch, base race, etc.).
+    """
+    reasons: list[str] = []
+    root = os.path.realpath(canonical_repo_root or "")
+    workspace = os.path.realpath(workspace_path or root or ".")
+    branch = (current_branch or "").strip()
+    dirty = parse_dirty_tracked_files(porcelain_status or "")
+    under_branches = is_path_under_branches(workspace, root) if root else False
+
+    if not is_create_issue_task(task):
+        return _result(
+            not_applicable=True,
+            allowed=False,
+            block=False,
+            reasons=["task is not create_issue"],
+            workspace=workspace,
+            root=root,
+            branch=branch,
+            dirty=dirty,
+            under_branches=under_branches,
+        )
+
+    # Registered branches/ worktrees keep the normal path (no bootstrap).
+    if under_branches:
+        return _result(
+            not_applicable=True,
+            allowed=False,
+            block=False,
+            reasons=["workspace is under branches/; ordinary #274 path applies"],
+            workspace=workspace,
+            root=root,
+            branch=branch,
+            dirty=dirty,
+            under_branches=True,
+        )
+
+    # Only the exact canonical control checkout is eligible.
+    if not root or workspace != root:
+        reasons.append(
+            "create_issue bootstrap requires the canonical control checkout; "
+            f"workspace '{workspace}' is not the repository root '{root or '(unknown)'}'"
+        )
+        return _result(
+            not_applicable=False,
+            allowed=False,
+            block=True,
+            reasons=reasons,
+            workspace=workspace,
+            root=root,
+            branch=branch,
+            dirty=dirty,
+            under_branches=False,
+            exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
+        )
+
+    if dirty:
+        reasons.append(
+            "create_issue bootstrap blocked: control checkout has tracked local "
+            f"edits (dirty files: {', '.join(dirty)})"
+        )
+
+    if not branch:
+        reasons.append(
+            "create_issue bootstrap blocked: control checkout is detached HEAD; "
+            "expected an accepted base branch (master/main/dev)"
+        )
+    elif branch not in BASE_BRANCHES:
+        reasons.append(
+            f"create_issue bootstrap blocked: control checkout branch '{branch}' "
+            f"is not an accepted base branch ({'/'.join(sorted(BASE_BRANCHES))})"
+        )
+
+    remote_tip = (remote_master_sha or "").strip() or None
+    local_tip = (head_sha or "").strip() or None
+    if remote_tip and local_tip and remote_tip != local_tip:
+        reasons.append(
+            "create_issue bootstrap blocked: control checkout HEAD does not match "
+            f"live master (HEAD {local_tip[:12]}, live master {remote_tip[:12]})"
+        )
+
+    if reasons:
+        return _result(
+            not_applicable=False,
+            allowed=False,
+            block=True,
+            reasons=reasons,
+            workspace=workspace,
+            root=root,
+            branch=branch or None,
+            dirty=dirty,
+            under_branches=False,
+            exact_next_action=EXACT_NEXT_ACTION_BOOTSTRAP,
+        )
+
+    return _result(
+        not_applicable=False,
+        allowed=True,
+        block=False,
+        reasons=[],
+        workspace=workspace,
+        root=root,
+        branch=branch or None,
+        dirty=dirty,
+        under_branches=False,
+        exact_next_action=EXACT_NEXT_ACTION_POST_CREATE,
+        bootstrap_path="clean_canonical_control_checkout",
+    )
+
+
+def format_create_issue_bootstrap_error(assessment: dict[str, Any]) -> str:
+    """RuntimeError / typed-block message for a failed bootstrap assessment."""
+    reasons = "; ".join(
+        assessment.get("reasons") or ["create_issue bootstrap failed"]
+    )
+    next_action = (
+        assessment.get("exact_next_action") or EXACT_NEXT_ACTION_BOOTSTRAP
+    )
+    root = assessment.get("canonical_repo_root") or "(unknown)"
+    workspace = assessment.get("workspace_path") or "(unknown)"
+    return (
+        f"Create-issue bootstrap guard (#749): {reasons}. "
+        f"canonical repository root: {root}; workspace: {workspace}. "
+        f"exact_next_action: {next_action}"
+    )
+
+
+def _result(
+    *,
+    not_applicable: bool,
+    allowed: bool,
+    block: bool,
+    reasons: list[str],
+    workspace: str,
+    root: str,
+    branch: str | None,
+    dirty: list[str],
+    under_branches: bool,
+    exact_next_action: str | None = None,
+    bootstrap_path: str | None = None,
+) -> dict[str, Any]:
+    return {
+        "not_applicable": not_applicable,
+        "allowed": allowed,
+        "block": block,
+        "proven": allowed and not block,
+        "reasons": list(reasons),
+        "workspace_path": workspace,
+        "canonical_repo_root": root,
+        "current_branch": branch,
+        "dirty_files": list(dirty),
+        "under_branches": under_branches,
+        "exact_next_action": exact_next_action,
+        "bootstrap_path": bootstrap_path,
+        "task_scope": "create_issue_only",
+    }

docs/canonical-repository-root.md

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

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

docs/llm-workflow-runbooks.md

@@ -48,6 +48,16 @@ It extracts the issue-first, isolated-worktree, no-self-review, profile-safety,
 merge-cleanup, fail-closed, and recovery rules into a reusable package that can
 be adapted to other repositories.
 
+### Sanctioned first mutation: `create_issue` from clean control (#749)
+
+Creating a tracking issue has no issue number yet, so no `branches/issue-<N>-*`
+worktree can exist. The sanctioned path is: clean canonical control checkout
+(accepted base branch, base-equivalent to live master, no tracked dirt) →
+resolve exact `create_issue` → `gitea_create_issue`. After the issue exists,
+all further author mutations require a registered issue-backed worktree and
+lock. Do not improvise with dummy directories, borrowed worktrees, or pre-issue
+worktrees. See `skills/llm-project-workflow/workflows/create-issue.md` §18a.
+
 ## Principle: the profile is the role, not the LLM
 
 ```text

gitea_config.py

@@ -272,6 +272,15 @@ def load_config(path=None):
         )
     if not isinstance(data.get("profiles"), dict):
         raise ConfigError(f"{path} must be a JSON object with a 'profiles' object")
+    # #741: the v1 path returns `data` unflattened, so nothing else validates
+    # the cross-repository binding before gitea_auth.get_profile() reads it.
+    # Validate it here so every supported loader treats the field identically
+    # (a relative or blank path must never reach the runtime guard).
+    for _name, _profile in data["profiles"].items():
+        if isinstance(_profile, dict):
+            _validate_canonical_repository_root(
+                _name, _profile.get("canonical_repository_root")
+            )
     return data
 
 
@@ -358,6 +367,14 @@ def _flatten_identity(env_name, svc_name, svc, ident_name, ident):
     for key in ("role", "username", "execution_profile", "audit_label"):
         if ident.get(key):
             profile[key] = ident[key]
+    # #741: the cross-repository binding must survive flattening. Previously
+    # this key was silently dropped here, so a v2-environments namespace that
+    # declared canonical_repository_root fell back to the *installation* root
+    # and mutated Gitea-Tools instead of its target repository — a fail-open.
+    # Validate it exactly as the v2-contexts loader does before propagating.
+    _validate_canonical_repository_root(addr, ident.get("canonical_repository_root"))
+    if ident.get("canonical_repository_root"):
+        profile["canonical_repository_root"] = ident["canonical_repository_root"]
     return addr, profile
 
 

issue_lock_adoption.py

@@ -85,6 +85,15 @@ def _branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
     return re.search(pattern, name) is not None
 
 
+def branch_carries_issue_marker(branch_name: str, issue_number: int) -> bool:
+    """Public accessor for the exact issue-marker match (#753).
+
+    Dead-session lock recovery needs the same word-boundary matcher to detect
+    ambiguous branch claims, so it is exposed rather than reached into.
+    """
+    return _branch_carries_issue_marker(branch_name, issue_number)
+
+
 def assess_own_branch_adoption(
     *,
     issue_number: int,

issue_lock_recovery.py

@@ -0,0 +1,451 @@
+"""Dead-session author issue-lock recovery (#753).
+
+A durable author issue lock records the PID of the MCP session that took it.
+When that process exits, ``issue_lock_store.assess_lock_freshness`` classifies
+the lock as ``stale`` (``live=False``) even while its lease is still within TTL,
+so every ownership check that requires a *live* lock fails closed.
+
+Re-taking the lock through ``gitea_lock_issue`` is unreachable for real work:
+``issue_lock_worktree.assess_issue_lock_worktree`` demands the worktree be
+base-equivalent to ``master``/``main``/``dev``, and a branch that already
+carries commits is ahead of its base by construction. The existing
+``assess_expired_lock_reclaim`` affordance does not apply either, because
+``assess_same_issue_lease_conflict`` only consults it once the lease has
+*expired* — a dead PID under an unexpired lease never reaches it.
+
+This module is the pure evidence assessor for that one narrow case. It grants
+recovery only when every element of durable ownership still matches exactly and
+the recorded process is demonstrably dead. It never trusts caller assertions:
+every field is compared against durable lock state or live observation supplied
+by the caller. It performs no mutation and no network I/O.
+
+Recovery deliberately does **not** relax base-equivalence for brand-new issue
+claims — only for a lock whose own prior record already proves the branch,
+worktree, head, and author.
+"""
+
+from __future__ import annotations
+
+import os
+from typing import Any, Iterable, Mapping, Sequence
+
+from issue_lock_store import is_process_alive
+from reviewer_worktree import parse_dirty_tracked_files
+
+# Outcome values
+RECOVERY_SANCTIONED = "RECOVERY_SANCTIONED"
+NO_CANDIDATE = "NO_CANDIDATE"
+REFUSED = "REFUSED"
+
+# Durable fields a lock must carry before it can be considered at all.
+REQUIRED_LOCK_FIELDS = ("issue_number", "branch_name", "worktree_path")
+
+
+def _same_realpath(left: str | None, right: str | None) -> bool:
+    if not left or not right:
+        return False
+    try:
+        return os.path.realpath(left) == os.path.realpath(right)
+    except OSError:
+        return left == right
+
+
+def _text(value: Any) -> str:
+    return str(value or "").strip()
+
+
+def _lock_claimant(lock: Mapping[str, Any]) -> dict[str, Any]:
+    claimant = lock.get("claimant")
+    if not isinstance(claimant, Mapping):
+        lease = lock.get("work_lease")
+        claimant = lease.get("claimant") if isinstance(lease, Mapping) else None
+    return dict(claimant) if isinstance(claimant, Mapping) else {}
+
+
+def _recorded_pid(lock: Mapping[str, Any]) -> Any:
+    pid = lock.get("session_pid")
+    if pid is None:
+        pid = lock.get("pid")
+    return pid
+
+
+def _malformed_reasons(lock: Mapping[str, Any]) -> list[str]:
+    """Names of durable fields that are missing or unusable."""
+    missing: list[str] = []
+    for field in REQUIRED_LOCK_FIELDS:
+        if not _text(lock.get(field)):
+            missing.append(field)
+    pid = _recorded_pid(lock)
+    if pid is None or _text(pid) == "":
+        missing.append("session_pid/pid")
+    else:
+        try:
+            if int(pid) <= 0:
+                missing.append("session_pid/pid")
+        except (TypeError, ValueError):
+            missing.append("session_pid/pid")
+    return missing
+
+
+def assess_dead_session_lock_recovery(
+    existing_lock: Mapping[str, Any] | None,
+    *,
+    issue_number: int,
+    branch_name: str,
+    worktree_path: str,
+    remote: str,
+    org: str,
+    repo: str,
+    identity: str | None,
+    profile: str | None,
+    current_branch: str | None,
+    porcelain_status: str,
+    head_sha: str | None,
+    remote_head_sha: str | None,
+    pr_head_sha: str | None = None,
+    pr_number: int | None = None,
+    competing_live_locks: Sequence[Mapping[str, Any]] | None = None,
+    candidate_branches: Iterable[str] | None = None,
+    current_pid: int | None = None,
+) -> dict[str, Any]:
+    """Decide whether a dead-session author lock may be natively recovered.
+
+    Returns a dict with ``recovery_sanctioned`` (bool), ``outcome``, ``reasons``
+    (why it was refused, or the positive proof when sanctioned), and
+    ``evidence`` (a redaction-safe record for auditing).
+
+    ``NO_CANDIDATE`` means no recovery was attempted at all — there is no
+    existing lock, or the lock does not describe this issue. The caller must
+    treat that exactly as it treated the pre-#753 world. ``REFUSED`` means a
+    candidate existed but the evidence did not agree; the caller fails closed.
+    """
+    reasons: list[str] = []
+    evidence: dict[str, Any] = {
+        "issue_number": issue_number,
+        "branch_name": branch_name,
+        "worktree_path": worktree_path,
+        "remote": remote,
+        "org": org,
+        "repo": repo,
+    }
+
+    if not existing_lock:
+        return _result(
+            NO_CANDIDATE, False, ["no existing durable lock for this issue"], evidence
+        )
+
+    lock = dict(existing_lock)
+
+    # ── Candidate identification ────────────────────────────────────────────
+    # Recovery only ever applies to a lock that already claims THIS issue.
+    # Anything else is not a recovery candidate and must not be reinterpreted.
+    if lock.get("issue_number") != issue_number:
+        return _result(
+            NO_CANDIDATE,
+            False,
+            [
+                f"existing lock targets issue #{lock.get('issue_number')}, "
+                f"not #{issue_number}; not a recovery candidate"
+            ],
+            evidence,
+        )
+
+    # A malformed/incomplete durable record can never prove ownership.
+    missing = _malformed_reasons(lock)
+    if missing:
+        return _result(
+            REFUSED,
+            False,
+            [
+                "durable lock record is incomplete and cannot prove ownership "
+                f"(missing/unusable: {', '.join(missing)})"
+            ],
+            evidence,
+        )
+
+    recorded_pid = _recorded_pid(lock)
+    evidence["prior_session_pid"] = recorded_pid
+    evidence["replacement_session_pid"] = (
+        current_pid if current_pid is not None else os.getpid()
+    )
+
+    # ── Repository scope ────────────────────────────────────────────────────
+    for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
+        actual = _text(lock.get(field))
+        if actual != _text(expected):
+            reasons.append(
+                f"lock {field} '{actual}' does not match requested '{_text(expected)}'"
+            )
+
+    # ── Branch identity ─────────────────────────────────────────────────────
+    locked_branch = _text(lock.get("branch_name"))
+    if locked_branch != _text(branch_name):
+        reasons.append(
+            f"lock branch '{locked_branch}' does not match requested "
+            f"'{_text(branch_name)}'"
+        )
+    evidence["locked_branch"] = locked_branch
+
+    # The worktree must actually be sitting on the locked branch. Without this
+    # a clean worktree parked elsewhere could stand in for the real work.
+    checked_out = _text(current_branch)
+    if not checked_out:
+        reasons.append(
+            "worktree is not on a named branch (detached HEAD); locked-branch "
+            "occupancy could not be proven"
+        )
+    elif checked_out != locked_branch:
+        reasons.append(
+            f"worktree is on branch '{checked_out}', not the locked branch "
+            f"'{locked_branch}'"
+        )
+
+    # ── Worktree identity ───────────────────────────────────────────────────
+    locked_worktree = _text(lock.get("worktree_path"))
+    if not _same_realpath(locked_worktree, worktree_path):
+        reasons.append(
+            f"lock worktree '{locked_worktree}' does not match declared "
+            f"'{_text(worktree_path)}'"
+        )
+    evidence["locked_worktree_path"] = locked_worktree
+
+    # ── Cleanliness (never waived) ──────────────────────────────────────────
+    dirty_files = parse_dirty_tracked_files(porcelain_status)
+    if dirty_files:
+        reasons.append(
+            "worktree has tracked local edits; recovery requires a clean "
+            f"worktree (dirty files: {', '.join(dirty_files)})"
+        )
+    evidence["dirty_files"] = dirty_files
+
+    # ── Head agreement: local == remote == PR ───────────────────────────────
+    local_head = _text(head_sha)
+    remote_head = _text(remote_head_sha)
+    if not local_head:
+        reasons.append("local head SHA could not be determined")
+    if not remote_head:
+        reasons.append(
+            f"remote head for branch '{locked_branch}' could not be determined"
+        )
+    if local_head and remote_head and local_head != remote_head:
+        reasons.append(
+            f"local head {local_head} does not match remote branch head {remote_head}"
+        )
+    evidence["local_head"] = local_head or None
+    evidence["remote_head"] = remote_head or None
+
+    pr_head = _text(pr_head_sha)
+    if pr_head:
+        evidence["pr_head"] = pr_head
+        evidence["pr_number"] = pr_number
+        if local_head and pr_head != local_head:
+            reasons.append(
+                f"open PR #{pr_number} head {pr_head} does not match local head "
+                f"{local_head}"
+            )
+
+    # ── Author identity ─────────────────────────────────────────────────────
+    claimant = _lock_claimant(lock)
+    locked_identity = _text(claimant.get("username"))
+    locked_profile = _text(claimant.get("profile"))
+    evidence["locked_identity"] = locked_identity or None
+    evidence["locked_profile"] = locked_profile or None
+    if not locked_identity or not locked_profile:
+        reasons.append(
+            "durable lock does not record a claimant identity/profile; "
+            "author ownership could not be proven"
+        )
+    if not _text(identity) or not _text(profile):
+        reasons.append(
+            "active session identity/profile is unknown; author ownership "
+            "could not be proven"
+        )
+    if locked_identity and _text(identity) and locked_identity != _text(identity):
+        reasons.append(
+            f"lock claimant '{locked_identity}' does not match active identity "
+            f"'{_text(identity)}'"
+        )
+    if locked_profile and _text(profile) and locked_profile != _text(profile):
+        reasons.append(
+            f"lock profile '{locked_profile}' does not match active profile "
+            f"'{_text(profile)}'"
+        )
+
+    # ── The defining condition: the recorded owner must be dead ─────────────
+    prior_alive = is_process_alive(recorded_pid)
+    evidence["prior_pid_alive"] = prior_alive
+    if prior_alive:
+        reasons.append(
+            f"prior owner pid {recorded_pid} is still alive; this is not a "
+            "dead-session recovery"
+        )
+    if current_pid is not None and recorded_pid is not None:
+        try:
+            if int(recorded_pid) == int(current_pid):
+                reasons.append(
+                    "recorded pid is the current session; nothing to recover"
+                )
+        except (TypeError, ValueError):
+            pass
+
+    # ── Competing ownership ─────────────────────────────────────────────────
+    competing: list[dict[str, Any]] = []
+    for entry in competing_live_locks or ():
+        if not isinstance(entry, Mapping):
+            continue
+        same_issue = entry.get("issue_number") == issue_number
+        same_branch = _text(entry.get("branch_name")) == locked_branch
+        if not (same_issue or same_branch):
+            continue
+        # The lock we are recovering is not competition with itself.
+        if (
+            same_issue
+            and same_branch
+            and _same_realpath(_text(entry.get("worktree_path")), worktree_path)
+        ):
+            continue
+        competing.append(
+            {
+                "issue_number": entry.get("issue_number"),
+                "branch_name": entry.get("branch_name"),
+                "worktree_path": entry.get("worktree_path"),
+                "pid": entry.get("pid"),
+            }
+        )
+    if competing:
+        described = ", ".join(
+            f"issue #{c['issue_number']} branch '{c['branch_name']}'" for c in competing
+        )
+        reasons.append(f"competing live lock or lease exists ({described})")
+    evidence["competing_live_locks"] = competing
+
+    # ── Ambiguous branch claims ─────────────────────────────────────────────
+    others = [
+        name
+        for name in (candidate_branches or ())
+        if _text(name) and _text(name) != locked_branch
+    ]
+    if others:
+        reasons.append(
+            "multiple branches claim this issue "
+            f"({', '.join(sorted(set(others)))}); ownership is ambiguous"
+        )
+    evidence["other_candidate_branches"] = sorted(set(others))
+
+    if reasons:
+        return _result(REFUSED, False, reasons, evidence)
+
+    return _result(
+        RECOVERY_SANCTIONED,
+        True,
+        [
+            f"durable lock for issue #{issue_number} matches branch "
+            f"'{locked_branch}', worktree '{locked_worktree}', head {local_head}, "
+            f"and claimant '{locked_identity}'; recorded pid {recorded_pid} is dead"
+        ],
+        evidence,
+    )
+
+
+def _result(
+    outcome: str,
+    sanctioned: bool,
+    reasons: list[str],
+    evidence: dict[str, Any],
+) -> dict[str, Any]:
+    return {
+        "outcome": outcome,
+        "recovery_sanctioned": sanctioned,
+        "is_candidate": outcome != NO_CANDIDATE,
+        "reasons": reasons,
+        "evidence": evidence,
+    }
+
+
+def owning_pr_recovery_evidence(
+    assessment: Mapping[str, Any] | None,
+) -> dict[str, Any] | None:
+    """Server-derived proof of the open PR a sanctioned recovery already owns (#755).
+
+    A dead-session recovery is, by construction, recovery of work that already
+    has an open PR — so the duplicate-work gate's linked-open-PR blocker would
+    otherwise discard every sanctioned recovery. This distils the completed
+    assessment into the minimum evidence that gate needs to tell "the PR this
+    lock already owns" apart from "a competing duplicate PR".
+
+    Returns ``None`` unless recovery was actually granted and the assessment's
+    own evidence names exactly one owning PR whose head agrees with the local
+    and remote heads. Nothing here is caller-supplied: every field is copied
+    from evidence the assessor built out of durable lock state plus live
+    git/Gitea observation, so a caller cannot manufacture an exemption.
+    """
+    if not isinstance(assessment, Mapping):
+        return None
+    if assessment.get("outcome") != RECOVERY_SANCTIONED:
+        return None
+    if not assessment.get("recovery_sanctioned"):
+        return None
+
+    evidence = assessment.get("evidence") or {}
+    branch_name = _text(evidence.get("locked_branch"))
+    pr_head = _text(evidence.get("pr_head"))
+    local_head = _text(evidence.get("local_head"))
+    remote_head = _text(evidence.get("remote_head"))
+    raw_pr_number = evidence.get("pr_number")
+
+    if raw_pr_number is None or not branch_name or not pr_head:
+        return None
+    # The assessor already required these to agree. Re-check, so a truncated or
+    # hand-built evidence map can never authorize an exemption.
+    if pr_head != local_head or pr_head != remote_head:
+        return None
+    try:
+        pr_number = int(raw_pr_number)
+        issue_number = int(evidence.get("issue_number"))
+    except (TypeError, ValueError):
+        return None
+
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch_name,
+        "head_sha": pr_head,
+    }
+
+
+def build_recovery_record(
+    assessment: Mapping[str, Any],
+    *,
+    recovered_at: str,
+) -> dict[str, Any]:
+    """Durable, secret-free provenance for a completed recovery (#753 AC2/AC6)."""
+    evidence = dict(assessment.get("evidence") or {})
+    return {
+        "recovered": True,
+        "reason": "owning MCP session exited; durable ownership evidence matched",
+        "recovered_at": recovered_at,
+        "prior_session_pid": evidence.get("prior_session_pid"),
+        "replacement_session_pid": evidence.get("replacement_session_pid"),
+        "prior_pid_alive": evidence.get("prior_pid_alive"),
+        "branch_name": evidence.get("locked_branch"),
+        "worktree_path": evidence.get("locked_worktree_path"),
+        "local_head": evidence.get("local_head"),
+        "remote_head": evidence.get("remote_head"),
+        "pr_head": evidence.get("pr_head"),
+        "pr_number": evidence.get("pr_number"),
+        "identity": evidence.get("locked_identity"),
+        "profile": evidence.get("locked_profile"),
+        "proof": list(assessment.get("reasons") or []),
+    }
+
+
+def format_recovery_refusal(assessment: Mapping[str, Any]) -> str:
+    """Single fail-closed message for a refused recovery attempt."""
+    reasons = list(assessment.get("reasons") or []) or [
+        "dead-session lock recovery evidence did not agree"
+    ]
+    return (
+        "Dead-session issue-lock recovery refused: "
+        + "; ".join(reasons)
+        + " (fail closed)"
+    )

issue_lock_worktree.py

@@ -92,8 +92,19 @@ def assess_issue_lock_worktree(
     inspected_git_root: str | None = None,
     base_branch: str | None = None,
     base_branches: frozenset[str] | None = None,
+    recovery_sanctioned: bool = False,
 ) -> dict:
-    """Fail closed when lock preconditions are not met on the declared worktree."""
+    """Fail closed when lock preconditions are not met on the declared worktree.
+
+    ``recovery_sanctioned`` is set only when ``issue_lock_recovery`` has already
+    proven, from the durable lock itself, that this is a dead-session recovery of
+    an existing claim (#753): same issue, branch, worktree, author, and head, with
+    the recording process dead. In that one case the base-equivalence requirement
+    is waived, because a branch that already carries the work is ahead of its base
+    by construction and could never satisfy it. Every other precondition —
+    notably worktree cleanliness — still applies unchanged, and brand-new issue
+    claims keep the full base-equivalence requirement.
+    """
     bases = base_branches or BASE_BRANCHES
     reasons: list[str] = []
     path = (worktree_path or "").strip()
@@ -111,7 +122,11 @@ def assess_issue_lock_worktree(
             f"(dirty files: {', '.join(dirty_files)})"
         )
 
-    if base_equivalent is False:
+    if recovery_sanctioned:
+        # Base-equivalence intentionally not evaluated: ownership was proven
+        # against the durable lock record instead (#753).
+        pass
+    elif base_equivalent is False:
         reasons.append(
             "issue lock worktree must be base-equivalent to one of "
             f"{_base_list(bases)} before implementation work; inspected "
@@ -139,6 +154,7 @@ def assess_issue_lock_worktree(
         inspected_git_root=inspected_git_root,
         base_branch=base_branch,
         base_equivalent=base_equivalent,
+        recovery_sanctioned=recovery_sanctioned,
     )
 
 
@@ -197,6 +213,7 @@ def _assessment(
     inspected_git_root: str | None = None,
     base_branch: str | None = None,
     base_equivalent: bool | None = None,
+    recovery_sanctioned: bool = False,
 ) -> dict:
     return {
         "proven": proven,
@@ -208,6 +225,8 @@ def _assessment(
         "dirty_files": dirty_files,
         "base_branch": base_branch,
         "base_equivalent": base_equivalent,
+        "recovery_sanctioned": recovery_sanctioned,
+        "base_equivalence_waived": bool(recovery_sanctioned),
     }
 
 

issue_work_duplicate_gate.py

@@ -2,7 +2,7 @@
 
 from __future__ import annotations
 
-from typing import Any
+from typing import Any, Mapping
 
 import issue_claim_heartbeat as claim_hb
 
@@ -27,6 +27,116 @@ def _linked_open_pr(issue_number: int, open_prs: list[dict]) -> dict | None:
     return claim_hb._linked_open_pr(issue_number, open_prs)
 
 
+def _pr_links_issue(issue_number: int, pr: Mapping[str, Any]) -> bool:
+    """Same linkage rule ``claim_hb._linked_open_pr`` applies, per PR.
+
+    ``_linked_open_pr`` only yields the *first* match, which cannot answer
+    "is there exactly one linked PR?" — a question the owning-PR exemption
+    below must answer before it can trust any of them.
+    """
+    pattern = _issue_pattern(issue_number)
+    head = (pr.get("head") or {}).get("ref") or ""
+    text = f"{pr.get('title', '')} {pr.get('body', '')}".lower()
+    if pattern in head.lower():
+        return True
+    return (
+        f"closes #{int(issue_number)}" in text
+        or f"fixes #{int(issue_number)}" in text
+    )
+
+
+def _all_linked_open_prs(
+    issue_number: int, open_prs: list[dict]
+) -> list[Mapping[str, Any]]:
+    return [pr for pr in (open_prs or []) if _pr_links_issue(issue_number, pr)]
+
+
+def _assess_owning_pr_exemption(
+    issue_number: int,
+    *,
+    linked_open_prs: list[Mapping[str, Any]],
+    locked_branch: str | None,
+    recovered_owning_pr: Mapping[str, Any] | None,
+) -> tuple[bool, list[str]]:
+    """Is the linked open PR provably the one a sanctioned recovery owns (#755)?
+
+    ``recovered_owning_pr`` is produced by
+    ``issue_lock_recovery.owning_pr_recovery_evidence`` from a completed
+    server-side recovery assessment — it is never a caller-supplied field.
+    Every element is re-checked here against the live PR list this gate was
+    given, so a stale or partial token cannot widen the exemption.
+
+    Returns ``(exempt, diagnostic_reasons)``. Diagnostics are only emitted when
+    a token was offered and rejected, so a blocked caller can see which element
+    of ownership disagreed.
+    """
+    if not recovered_owning_pr:
+        return False, []
+
+    notes: list[str] = []
+    token_issue = recovered_owning_pr.get("issue_number")
+    token_pr = recovered_owning_pr.get("pr_number")
+    token_branch = str(recovered_owning_pr.get("branch_name") or "").strip()
+    token_head = str(recovered_owning_pr.get("head_sha") or "").strip()
+    locked = (locked_branch or "").strip()
+
+    if token_issue is not None and int(token_issue) != int(issue_number):
+        notes.append(
+            f"recovery evidence is for issue #{token_issue}, not "
+            f"#{issue_number} (no owning-PR exemption)"
+        )
+        return False, notes
+    if not locked or not token_branch or locked != token_branch:
+        notes.append(
+            f"recovery evidence branch '{token_branch or 'unknown'}' does not "
+            f"match the branch being locked '{locked or 'unknown'}' "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+    if len(linked_open_prs) != 1:
+        numbers = ", ".join(
+            f"#{pr.get('number')}" for pr in linked_open_prs
+        ) or "none"
+        notes.append(
+            f"{len(linked_open_prs)} open PRs link issue #{issue_number} "
+            f"({numbers}); recovery may only own exactly one "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+
+    only = linked_open_prs[0]
+    head_obj = only.get("head") or {}
+    only_number = only.get("number")
+    only_ref = str(head_obj.get("ref") or "").strip()
+    only_sha = str(head_obj.get("sha") or "").strip()
+
+    if token_pr is None or only_number is None or int(only_number) != int(token_pr):
+        notes.append(
+            f"linked open PR #{only_number} is not the recovered owning PR "
+            f"#{token_pr} (no owning-PR exemption)"
+        )
+        return False, notes
+    if only_ref != token_branch:
+        notes.append(
+            f"open PR #{only_number} head branch '{only_ref}' does not match "
+            f"the recovered branch '{token_branch}' (no owning-PR exemption)"
+        )
+        return False, notes
+    if not token_head or not only_sha or only_sha != token_head:
+        notes.append(
+            f"open PR #{only_number} head {only_sha or 'unknown'} does not "
+            f"match the recovered head {token_head or 'unknown'} "
+            "(no owning-PR exemption)"
+        )
+        return False, notes
+
+    return True, [
+        f"open PR #{only_number} is the exact PR already owned by the "
+        f"recovering lock for issue #{issue_number} (branch '{token_branch}', "
+        f"head {token_head}); not duplicate work"
+    ]
+
+
 def _matching_branches(
     issue_number: int,
     branch_names: list[str],
@@ -52,8 +162,15 @@ def assess_work_issue_duplicate_gate(
     claim_entry: dict | None = None,
     locked_branch: str | None = None,
     phase: str = PHASE_LOCK,
+    recovered_owning_pr: Mapping[str, Any] | None = None,
 ) -> dict[str, Any]:
-    """Fail closed when duplicate work is already in flight for an issue."""
+    """Fail closed when duplicate work is already in flight for an issue.
+
+    ``recovered_owning_pr`` (#755) is server-derived evidence that a sanctioned
+    dead-session lock recovery already owns one specific open PR. It exempts
+    *only* that exact PR from the linked-open-PR blocker; every other duplicate
+    signal, and every mismatch, keeps failing closed.
+    """
     reasons: list[str] = []
     outcome = OUTCOME_DUPLICATE_WORK_NOT_PREVENTED
     prs = list(open_prs or [])
@@ -61,12 +178,23 @@ def assess_work_issue_duplicate_gate(
     pattern = _issue_pattern(issue_number)
 
     linked = _linked_open_pr(issue_number, prs)
+    linked_open_prs = _all_linked_open_prs(issue_number, prs)
+    owning_pr_exempted = False
+    exemption_notes: list[str] = []
     if linked:
-        reasons.append(
+        owning_pr_exempted, exemption_notes = _assess_owning_pr_exemption(
-            f"open PR #{linked.get('number')} already covers issue "
+            issue_number,
-            f"#{issue_number} (fail closed)"
+            linked_open_prs=linked_open_prs,
+            locked_branch=locked_branch,
+            recovered_owning_pr=recovered_owning_pr,
         )
-        outcome = OUTCOME_DUPLICATE_PR_PREVENTED
+        if not owning_pr_exempted:
+            reasons.append(
+                f"open PR #{linked.get('number')} already covers issue "
+                f"#{issue_number} (fail closed)"
+            )
+            reasons.extend(exemption_notes)
+            outcome = OUTCOME_DUPLICATE_PR_PREVENTED
 
     conflicting_branches = _matching_branches(
         issue_number, branches, locked_branch=locked_branch
@@ -122,6 +250,9 @@ def assess_work_issue_duplicate_gate(
         "phase": phase,
         "outcome": outcome,
         "linked_open_pr": linked.get("number") if linked else entry.get("linked_open_pr"),
+        "linked_open_pr_count": len(linked_open_prs),
+        "owning_pr_recovery_exempted": owning_pr_exempted,
+        "owning_pr_recovery_notes": list(exemption_notes),
         "conflicting_branches": conflicting_branches,
         "claim_status": status or None,
         "reasons": reasons,

merger_lease_adoption.py

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

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,
+    }

pr_sync_status.py

@@ -41,6 +41,186 @@ _DENIED_UPDATE_ROLES = frozenset({"reviewer", "merger", "reconciler", "mixed", "
 UPDATE_STYLE_MERGE = "merge"
 _FORBIDDEN_UPDATE_STYLES = frozenset({"rebase", "rebase-merge", "squash", "force"})
 
+# ── Commit check classifications (#751) ──────────────────────────────────
+# Gitea's *combined* commit status reports ``state: pending`` both when a real
+# check is executing and when the status-context collection is empty. Reading
+# ``state`` alone therefore cannot distinguish "CI is running" from "no CI
+# exists", which permanently blocks a merge-ready PR that no check will ever
+# report on. These classifications are derived from the actual context
+# collection plus the live branch-protection policy.
+CHECKS_SUCCESS = "success"
+CHECKS_FAILURE = "failure"
+CHECKS_PENDING = "pending"
+CHECKS_NONE = "none"                          # configured/produced nothing
+CHECKS_NOT_REQUIRED = "not_required"          # protection does not require checks
+CHECKS_MISSING_REQUIRED = "missing_required"  # required contexts have no result
+CHECKS_UNKNOWN = "unknown"                    # indeterminable — fail closed
+
+# Values that permit merge_now when checks are required.
+_CHECKS_OK = frozenset({"success", "passed", "ok", "skipped", "not_required"})
+
+# Raw per-context state vocabularies reported by Gitea.
+_CTX_SUCCESS = frozenset({"success", "passed", "ok"})
+_CTX_FAILURE = frozenset({"failure", "failed", "error", "cancelled", "canceled"})
+_CTX_PENDING = frozenset({"pending", "running", "queued", "expected"})
+_CTX_SKIPPED = frozenset({"skipped", "neutral"})
+
+
+def _normalize_context_rows(statuses: Any) -> list[dict[str, str]]:
+    """Reduce a raw status collection to newest-wins ``{context, state}`` rows.
+
+    Gitea returns the status collection newest-first, so the first row seen for
+    a context wins. Rows without a usable state are discarded rather than being
+    silently treated as passing.
+    """
+    rows: list[dict[str, str]] = []
+    seen: set[str] = set()
+    if not isinstance(statuses, list):
+        return rows
+    for raw in statuses:
+        if not isinstance(raw, dict):
+            continue
+        context = (raw.get("context") or raw.get("name") or "").strip()
+        state = (raw.get("status") or raw.get("state") or "").strip().lower()
+        if not state:
+            continue
+        key = context or f"__unnamed__{len(rows)}"
+        if key in seen:
+            continue
+        seen.add(key)
+        rows.append({"context": context, "state": state})
+    return rows
+
+
+def _aggregate_context_states(rows: list[dict[str, str]]) -> str:
+    """Fail-closed aggregate: failure > pending > unknown-state > success."""
+    states = {row["state"] for row in rows}
+    if states & _CTX_FAILURE:
+        return CHECKS_FAILURE
+    if states & _CTX_PENDING:
+        return CHECKS_PENDING
+    unresolved = states - _CTX_SUCCESS - _CTX_SKIPPED
+    if unresolved:
+        # An unrecognized context state must never read as success.
+        return CHECKS_UNKNOWN
+    return CHECKS_SUCCESS
+
+
+def classify_commit_checks(
+    *,
+    combined_state: str | None = None,
+    statuses: Any = None,
+    checks_enabled: bool | None = None,
+    required_contexts: Any = None,
+    policy_determinable: bool = True,
+    status_determinable: bool = True,
+) -> dict[str, Any]:
+    """Classify head checks from live evidence (#751).
+
+    ``combined_state`` is deliberately **not** authoritative: it is recorded for
+    observability but never used to infer that CI is executing. The context
+    collection and the live protection policy decide.
+
+    Returns ``checks_status`` (one of the ``CHECKS_*`` values), the derived
+    ``checks_required`` flag, and structured ``reasons``.
+    """
+    reasons: list[str] = []
+    rows = _normalize_context_rows(statuses)
+    required = [
+        str(ctx).strip()
+        for ctx in (required_contexts or [])
+        if str(ctx or "").strip()
+    ]
+    observed_combined = (combined_state or "").strip().lower() or None
+
+    result: dict[str, Any] = {
+        "checks_status": CHECKS_UNKNOWN,
+        "checks_required": True,
+        "combined_state": observed_combined,
+        "context_count": len(rows),
+        "observed_contexts": [row["context"] for row in rows],
+        "required_contexts": required,
+        "missing_required_contexts": [],
+        "policy_determinable": bool(policy_determinable),
+        "status_determinable": bool(status_determinable),
+        "reasons": reasons,
+    }
+
+    # Policy unreadable → never assume checks are optional.
+    if not policy_determinable:
+        reasons.append(
+            "branch-protection check policy could not be read; cannot prove "
+            "whether status checks are required (fail closed)"
+        )
+        return result
+
+    if checks_enabled is False:
+        result["checks_required"] = False
+        result["checks_status"] = CHECKS_NOT_REQUIRED
+        reasons.append(
+            "live branch protection does not require status checks for the base "
+            "branch; head status contexts do not gate merge"
+        )
+        return result
+
+    if checks_enabled is None:
+        reasons.append(
+            "branch-protection status-check requirement is indeterminate "
+            "(fail closed)"
+        )
+        return result
+
+    # Checks are required from here on.
+    if not status_determinable:
+        reasons.append(
+            "head commit status collection could not be read while branch "
+            "protection requires status checks (fail closed)"
+        )
+        return result
+
+    if required:
+        by_context = {row["context"]: row["state"] for row in rows if row["context"]}
+        missing = [ctx for ctx in required if ctx not in by_context]
+        if missing:
+            result["missing_required_contexts"] = missing
+            result["checks_status"] = CHECKS_MISSING_REQUIRED
+            reasons.append(
+                "branch protection requires status context(s) "
+                f"{', '.join(missing)} but no matching status result exists at "
+                "the head commit (fail closed)"
+            )
+            return result
+        matched = [
+            {"context": ctx, "state": by_context[ctx]} for ctx in required
+        ]
+        result["checks_status"] = _aggregate_context_states(matched)
+        reasons.append(
+            f"evaluated {len(matched)} required status context(s) from live "
+            "branch protection; unrelated contexts were ignored"
+        )
+        return result
+
+    # Status checks enabled with no specific required contexts configured.
+    if not rows:
+        result["checks_status"] = CHECKS_NONE
+        reasons.append(
+            "branch protection enables status checks but no status context was "
+            "produced for the head commit"
+        )
+        if observed_combined in _CTX_PENDING:
+            reasons.append(
+                f"combined commit state '{observed_combined}' does not indicate "
+                "executing CI because the status-context collection is empty"
+            )
+        return result
+
+    result["checks_status"] = _aggregate_context_states(rows)
+    reasons.append(
+        f"aggregated {len(rows)} reported status context(s); branch protection "
+        "configures no explicit required-context list"
+    )
+    return result
+
 
 def _normalize_sha(value: str | None) -> str | None:
     text = (value or "").strip().lower()
@@ -120,6 +300,7 @@ def assess_pr_sync_status(
         "branch_protection_requires_current_base": requires_current,
         "approval_at_current_head": approval_ok if approval_at_current_head is not None else None,
         "checks_status": checks,
+        "checks_required": bool(checks_required),
         "active_locks_and_leases": {
             "author_lock": bool(active_author_lock) if active_author_lock is not None else None,
             "reviewer_lease": bool(active_reviewer_lease) if active_reviewer_lease is not None else None,
@@ -258,21 +439,41 @@ def assess_pr_sync_status(
         result["recommended_next_action"] = ACTION_BLOCKED
         return result
 
-    # ── Checks gate for merge_now ────────────────────────────────────────
+    # ── Checks gate for merge_now (#751) ─────────────────────────────────
-    if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
+    # ``checks_required`` is derived from the live branch-protection policy by
-        if checks in ("pending", "running", "queued"):
+    # the production caller. When protection does not require status checks,
+    # head contexts cannot gate the merge and this whole gate is skipped.
+    if not checks_required:
+        reasons.append(
+            "live branch protection does not require status checks; head check "
+            f"state ({checks}) does not gate merge"
+        )
+    elif checks not in _CHECKS_OK:
+        if checks in _CTX_PENDING:
             reasons.append(f"required checks are not finished (status={checks})")
-            result["recommended_next_action"] = ACTION_BLOCKED
+        elif checks in _CTX_FAILURE:
-            return result
-        if checks in ("failure", "failed", "error", "cancelled"):
             reasons.append(f"required checks failed (status={checks})")
-            result["recommended_next_action"] = ACTION_BLOCKED
+        elif checks == CHECKS_MISSING_REQUIRED:
-            return result
+            reasons.append(
-        # unknown — fail closed when checks_required
+                "branch protection configures required status context(s) but no "
-        if checks == "unknown":
+                "matching status result exists at the current head (fail closed)"
+            )
+        elif checks == CHECKS_NONE:
+            reasons.append(
+                "branch protection requires status checks but no status context "
+                "was produced for the current head (fail closed); an empty "
+                "status collection is not executing CI"
+            )
+        elif checks == CHECKS_UNKNOWN:
             reasons.append("checks status unknown (fail closed)")
-            result["recommended_next_action"] = ACTION_BLOCKED
+        else:
-            return result
+            # Unrecognized vocabulary must never fall through to merge_now.
+            reasons.append(
+                f"unrecognized checks status '{checks}' cannot prove required "
+                "checks passed (fail closed)"
+            )
+        result["recommended_next_action"] = ACTION_BLOCKED
+        return result
 
     # ── Ready to merge without update ────────────────────────────────────
     # Includes: current with approval; outdated when update is NOT required.

pr_work_lease.py

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

reviewer_pr_lease.py

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

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",

skills/llm-project-workflow/workflows/create-issue.md

@@ -450,6 +450,35 @@ If any gate fails, do not create the issue.
 
 Produce a recovery handoff or duplicate report.
 
+### 18a. Sanctioned first-mutation path (#749)
+
+`gitea_create_issue` is a **pure remote mutation** (no local tree write). The
+issue-first gate forbids creating `branches/issue-<N>-*` before the issue
+number exists. Therefore the **only sanctioned first mutation** is:
+
+1. Read-only identity + capability + duplicate search from the control checkout.
+2. Ensure the **canonical control checkout** is:
+   * the configured repository root for the requested remote/org/repo;
+   * on an accepted base branch (`master` / `main` / `dev`);
+   * base-equivalent to live master;
+   * clean (no tracked local edits);
+   * in runtime/master parity.
+3. Resolve exact task `create_issue`, then call `gitea_create_issue` **from that
+   clean control checkout** (no `worktree_path` required for this step alone).
+4. After the issue number exists: create a **registered** worktree under
+   `branches/issue-<N>-*`, claim/lock, and perform every subsequent author
+   mutation from that worktree only.
+
+**Forbidden improvisations (fail closed):**
+
+* `mkdir` dummy directories under `branches/` (#713)
+* borrowing an unrelated pre-existing worktree
+* creating a pre-issue worktree in violation of issue-first
+* running create_issue from a dirty, drifted, detached, or non-canonical root
+
+Post-creation mutations (`lock_issue`, commit, push, `create_pr`, etc.) **never**
+receive this bootstrap exemption.
+
 ## 19. Issue commenting gate
 
 Before commenting on an existing issue, verify:

task_capability_map.py

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

@@ -0,0 +1,367 @@
+"""Regression tests for create_issue bootstrap (#749).
+
+TDD: these tests define the sanctioned first-mutation path for
+``gitea_create_issue`` from a clean canonical control checkout, and prove
+the exemption cannot widen to dirty roots, foreign clones, arbitrary
+``branches/`` directories, or post-creation author mutations.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import create_issue_bootstrap as cib  # noqa: E402
+import gitea_mcp_server as srv  # noqa: E402
+import workflow_scope_guard as wsg  # noqa: E402
+
+FAKE_AUTH = {"Authorization": "token test-token"}
+MASTER_SHA = "a" * 40
+STALE_SHA = "b" * 40
+
+current_file_path = Path(__file__).resolve()
+if "branches" in current_file_path.parts:
+    CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[3])
+else:
+    CONTROL_CHECKOUT_ROOT = str(current_file_path.parents[1])
+
+
+class TestCreateIssueBootstrapAssessor(unittest.TestCase):
+    ROOT = "/repo/Gitea-Tools"
+
+    def test_non_create_issue_task_not_applicable(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=self.ROOT,
+            canonical_repo_root=self.ROOT,
+            current_branch="master",
+            head_sha=MASTER_SHA,
+            porcelain_status="",
+            remote_master_sha=MASTER_SHA,
+            task="lock_issue",
+        )
+        self.assertTrue(res["not_applicable"])
+        self.assertFalse(res["allowed"])
+        self.assertFalse(res["block"])
+
+    def test_branches_worktree_not_applicable(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=f"{self.ROOT}/branches/issue-1-x",
+            canonical_repo_root=self.ROOT,
+            current_branch="fix/issue-1-x",
+            task="create_issue",
+        )
+        self.assertTrue(res["not_applicable"])
+        self.assertFalse(res["allowed"])
+
+    def test_clean_control_checkout_allowed(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=self.ROOT,
+            canonical_repo_root=self.ROOT,
+            current_branch="master",
+            head_sha=MASTER_SHA,
+            porcelain_status="",
+            remote_master_sha=MASTER_SHA,
+            task="create_issue",
+        )
+        self.assertFalse(res["not_applicable"])
+        self.assertTrue(res["allowed"])
+        self.assertFalse(res["block"])
+        self.assertEqual(res["bootstrap_path"], "clean_canonical_control_checkout")
+        # Post-create next action must name issue-backed worktree after N exists.
+        self.assertIn("branches/issue-<N>-*", res["exact_next_action"])
+
+    def test_tool_alias_gitea_create_issue_allowed(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=self.ROOT,
+            canonical_repo_root=self.ROOT,
+            current_branch="main",
+            head_sha=MASTER_SHA,
+            porcelain_status="",
+            remote_master_sha=MASTER_SHA,
+            task="gitea_create_issue",
+        )
+        self.assertTrue(res["allowed"])
+
+    def test_dirty_control_checkout_blocked(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=self.ROOT,
+            canonical_repo_root=self.ROOT,
+            current_branch="master",
+            head_sha=MASTER_SHA,
+            porcelain_status=" M gitea_mcp_server.py\n",
+            remote_master_sha=MASTER_SHA,
+            task="create_issue",
+        )
+        self.assertTrue(res["block"])
+        self.assertFalse(res["allowed"])
+        self.assertTrue(any("tracked local edits" in r for r in res["reasons"]))
+        # Pre-issue phase: next action must be satisfiable without inventing <N>.
+        next_a = res["exact_next_action"] or ""
+        self.assertIn("clean accepted base branch", next_a)
+        self.assertIn("before the issue exists", next_a)
+        # Must not prescribe "bind branches/issue-<N>" as the recovery step.
+        self.assertNotIn("Bind an issue-backed worktree", next_a)
+
+    def test_stale_base_blocked(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=self.ROOT,
+            canonical_repo_root=self.ROOT,
+            current_branch="master",
+            head_sha=STALE_SHA,
+            porcelain_status="",
+            remote_master_sha=MASTER_SHA,
+            task="create_issue",
+        )
+        self.assertTrue(res["block"])
+        self.assertTrue(any("live master" in r for r in res["reasons"]))
+
+    def test_non_base_branch_blocked(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=self.ROOT,
+            canonical_repo_root=self.ROOT,
+            current_branch="feat/something",
+            head_sha=MASTER_SHA,
+            porcelain_status="",
+            remote_master_sha=MASTER_SHA,
+            task="create_issue",
+        )
+        self.assertTrue(res["block"])
+
+    def test_detached_head_blocked(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=self.ROOT,
+            canonical_repo_root=self.ROOT,
+            current_branch="",
+            head_sha=MASTER_SHA,
+            porcelain_status="",
+            remote_master_sha=MASTER_SHA,
+            task="create_issue",
+        )
+        self.assertTrue(res["block"])
+        self.assertTrue(any("detached" in r for r in res["reasons"]))
+
+    def test_foreign_workspace_blocked(self):
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path="/other/clone",
+            canonical_repo_root=self.ROOT,
+            current_branch="master",
+            head_sha=MASTER_SHA,
+            porcelain_status="",
+            remote_master_sha=MASTER_SHA,
+            task="create_issue",
+        )
+        self.assertTrue(res["block"])
+        self.assertTrue(any("canonical control checkout" in r for r in res["reasons"]))
+
+
+class TestCreateIssueBootstrapIntegration(unittest.TestCase):
+    def setUp(self):
+        srv._preflight_whoami_called = True
+        srv._preflight_capability_called = True
+        srv._preflight_resolved_role = "author"
+        srv._preflight_resolved_task = "create_issue"
+        srv._preflight_whoami_violation = False
+        srv._preflight_capability_violation = False
+        self._orig_in_test = srv._preflight_in_test_mode
+        srv._preflight_in_test_mode = lambda: False
+
+    def tearDown(self):
+        srv._preflight_in_test_mode = self._orig_in_test
+        srv._preflight_resolved_task = None
+
+    def _git_state(self, branch="master", head=MASTER_SHA, porcelain=""):
+        return {
+            "current_branch": branch,
+            "head_sha": head,
+            "porcelain_status": porcelain,
+        }
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch(
+        "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+        return_value=(True, []),
+    )
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    @patch(
+        "gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
+        return_value=MASTER_SHA,
+    )
+    def test_clean_control_checkout_create_issue_succeeds(
+        self, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth
+    ):
+        mock_api.return_value = {
+            "number": 99,
+            "html_url": "https://gitea.example.com/issues/99",
+        }
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch(
+                "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+                return_value=self._git_state(),
+            ):
+                with patch(
+                    "gitea_mcp_server._get_workspace_porcelain", return_value=""
+                ):
+                    with patch(
+                        "gitea_mcp_server._enforce_root_checkout_guard"
+                    ):
+                        # Anti-stomp / master parity: keep gates green.
+                        with patch.object(
+                            srv,
+                            "_run_anti_stomp_preflight",
+                            return_value=None,
+                        ):
+                            res = srv.gitea_create_issue(
+                                title="Bootstrap issue from clean control",
+                                body="Body text for content gate.",
+                            )
+        self.assertEqual(res.get("number"), 99)
+        mock_api.assert_called_once()
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch(
+        "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+        return_value=(True, []),
+    )
+    @patch("gitea_mcp_server.api_request")
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
+    def test_dirty_control_checkout_create_issue_fails_closed(
+        self, _get_all, mock_api, _role, _ns, _prof, _auth
+    ):
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch(
+                "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+                return_value=self._git_state(
+                    porcelain=" M author_mutation_worktree.py\n"
+                ),
+            ):
+                with patch(
+                    "gitea_mcp_server._get_workspace_porcelain",
+                    return_value=" M author_mutation_worktree.py\n",
+                ):
+                    res = srv.gitea_create_issue(
+                        title="Should fail on dirty root",
+                        body="Body text for content gate.",
+                    )
+        self.assertFalse(res.get("success", True) and res.get("number"))
+        if isinstance(res, dict) and res.get("success") is False:
+            blob = " ".join(res.get("reasons") or [])
+            self.assertTrue(
+                "tracked local edits" in blob
+                or "dirty" in blob.lower()
+                or "control checkout" in blob.lower()
+                or res.get("blocker_kind")
+            )
+            # Pre-issue phase must not demand issue-<N> worktree.
+            next_a = res.get("exact_next_action") or ""
+            if next_a:
+                self.assertNotIn("issue-<N>-*", next_a)
+        mock_api.assert_not_called()
+
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
+    @patch(
+        "gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop",
+        return_value=(True, []),
+    )
+    def test_lock_issue_still_requires_branches_worktree(self, _role, _ns, _prof, _auth):
+        """Existing issue-backed mutations receive no exemption (#749 AC3/AC7)."""
+        srv._preflight_resolved_task = "lock_issue"
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch(
+                "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+                return_value=self._git_state(),
+            ):
+                with patch(
+                    "gitea_mcp_server._get_workspace_porcelain", return_value=""
+                ):
+                    with patch(
+                        "gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha",
+                        return_value=MASTER_SHA,
+                    ):
+                        with patch.object(
+                            srv, "_enforce_root_checkout_guard"
+                        ):
+                            with self.assertRaises(RuntimeError) as ctx:
+                                srv.verify_preflight_purity(
+                                    remote="prgs",
+                                    task="lock_issue",
+                                )
+        msg = str(ctx.exception)
+        self.assertTrue(
+            "Branches-only mutation guard" in msg
+            or "stable control checkout" in msg,
+            msg,
+        )
+        self.assertIn("control checkout", msg)
+
+    def test_workflow_scope_skips_missing_worktree_for_create_issue_clean_root(self):
+        """#683 root assessor must not block clean-root create_issue bootstrap."""
+        res = wsg.assess_root_source_mutation(
+            workspace_path=CONTROL_CHECKOUT_ROOT,
+            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
+            porcelain_status="",
+            role_kind="author",
+            mutation_task="create_issue",
+        )
+        self.assertFalse(res["block"], res)
+        self.assertTrue(res.get("create_issue_bootstrap") or res["proven"])
+
+    def test_workflow_scope_still_blocks_clean_root_for_lock_issue(self):
+        res = wsg.assess_root_source_mutation(
+            workspace_path=CONTROL_CHECKOUT_ROOT,
+            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
+            porcelain_status="",
+            role_kind="author",
+            mutation_task="lock_issue",
+        )
+        self.assertTrue(res["block"])
+        self.assertEqual(res["blocker_kind"], wsg.BLOCKER_MISSING_WORKTREE)
+
+    def test_arbitrary_branches_directory_not_bootstrap(self):
+        """#713: mkdir fake under branches/ is not the bootstrap path."""
+        fake = os.path.join(CONTROL_CHECKOUT_ROOT, "branches", "fake-mkdir-only")
+        res = cib.assess_create_issue_bootstrap(
+            workspace_path=fake,
+            canonical_repo_root=CONTROL_CHECKOUT_ROOT,
+            current_branch="master",
+            head_sha=MASTER_SHA,
+            porcelain_status="",
+            remote_master_sha=MASTER_SHA,
+            task="create_issue",
+        )
+        # Under branches/ → not bootstrap; ordinary membership/registration applies.
+        self.assertTrue(res["not_applicable"])
+        self.assertFalse(res["allowed"])
+
+
+class TestCreateIssueCapabilityAgreement(unittest.TestCase):
+    def test_map_and_alias_agree_on_create_issue(self):
+        import task_capability_map as tcm
+
+        self.assertEqual(
+            tcm.required_permission("create_issue"),
+            "gitea.issue.create",
+        )
+        # Tool alias must resolve to the same task contract.
+        alias = getattr(tcm, "TOOL_TASK_ALIASES", None) or getattr(
+            tcm, "TASK_ALIASES", None
+        )
+        if alias is not None:
+            mapped = alias.get("gitea_create_issue")
+            if mapped is not None:
+                self.assertEqual(mapped, "create_issue")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_create_issue_workspace_guard.py

@@ -51,29 +51,62 @@ class TestCreateIssueWorkspaceGuard(unittest.TestCase):
             "porcelain_status": "",
         },
     )
-    def test_create_issue_stable_checkout_rejected(
+    def test_create_issue_stable_checkout_bootstrap_allowed_when_clean(
         self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
     ):
-        # Without worktree_path/env hints, workspace resolves to PROJECT_ROOT. When that
+        # #749: clean canonical control checkout is the sanctioned create_issue path.
-        # path is the stable control checkout (not under branches/), mutation must fail.
+        mock_api.return_value = {
+            "number": 77,
+            "html_url": "https://gitea.example.com/issues/77",
+        }
         with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
-            try:
+            with patch("gitea_mcp_server._get_workspace_porcelain", return_value=""):
-                res = srv.gitea_create_issue(title="Test issue", body="body text")
+                with patch.object(srv, "_run_anti_stomp_preflight", return_value=None):
-            except RuntimeError as exc:
+                    with patch.object(srv, "_enforce_root_checkout_guard"):
-                self.assertIn("stable control checkout", str(exc))
+                        res = srv.gitea_create_issue(
-            else:
+                            title="Test issue", body="body text for gate"
-                # #683: production guards return typed blockers at entrypoints
+                        )
-                self.assertFalse(res.get("success"))
+        self.assertEqual(res.get("number"), 77)
-                self.assertFalse(res.get("performed"))
+        mock_api.assert_called_once()
-                blob = " ".join(res.get("reasons") or []) + " " + str(
+
-                    res.get("blocker_kind") or ""
+    @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
-                )
+    @patch("gitea_mcp_server._profile_permission_block", return_value=None)
-                self.assertTrue(
+    @patch("gitea_mcp_server._namespace_mutation_block", return_value=None)
-                    "stable control checkout" in blob
+    @patch("gitea_mcp_server.role_session_router.check_author_mutation_after_reviewer_stop", return_value=(True, []))
-                    or "missing_issue_worktree" in blob
+    @patch("gitea_mcp_server.api_request")
-                    or "control checkout" in blob.lower()
+    @patch("gitea_mcp_server.api_get_all", return_value=[])
-                )
+    @patch("gitea_mcp_server.root_checkout_guard.resolve_remote_master_sha", return_value="a" * 40)
-                self.assertTrue(res.get("exact_next_action"))
+    @patch(
+        "gitea_mcp_server.issue_lock_worktree.read_worktree_git_state",
+        return_value={
+            "current_branch": "master",
+            "head_sha": "a" * 40,
+            "porcelain_status": " M dirty.py\n",
+        },
+    )
+    def test_create_issue_dirty_control_checkout_rejected(
+        self, _git, _remote_sha, _get_all, mock_api, _role, _ns, _prof, _auth,
+    ):
+        # #749: dirty control checkout still fails closed (no bootstrap).
+        with patch.object(srv, "PROJECT_ROOT", CONTROL_CHECKOUT_ROOT):
+            with patch(
+                "gitea_mcp_server._get_workspace_porcelain",
+                return_value=" M dirty.py\n",
+            ):
+                try:
+                    res = srv.gitea_create_issue(title="Test issue", body="body text")
+                except RuntimeError as exc:
+                    self.assertTrue(
+                        "tracked local edits" in str(exc)
+                        or "dirty" in str(exc).lower()
+                        or "bootstrap" in str(exc).lower()
+                        or "control checkout" in str(exc).lower()
+                    )
+                else:
+                    self.assertFalse(res.get("success", True) and res.get("number"))
+                    blob = " ".join(res.get("reasons") or [])
+                    self.assertTrue(blob or res.get("blocker_kind"))
+        mock_api.assert_not_called()
 
     @patch("gitea_mcp_server._auth", return_value=FAKE_AUTH)
     @patch("gitea_mcp_server._profile_permission_block", return_value=None)

tests/test_issue_741_canonical_root_consumers.py

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

tests/test_issue_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_issue_747_sliding_lease_ttl.py

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

tests/test_issue_751_checks_assessor.py

@@ -0,0 +1,602 @@
+"""Regression coverage for the PR checks assessor defect (#751).
+
+Gitea's *combined* commit status reports ``state: pending`` both when a check is
+executing and when the status-context collection is empty. The assessor read
+``state`` alone and defaulted ``checks_required`` to ``True``, so a PR whose head
+had no status contexts — and never would — was routed to ``blocked`` forever.
+
+These tests pin the corrected semantics end to end: live branch protection
+decides whether checks are required, and the actual context collection decides
+what the checks say.
+"""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import pr_sync_status  # noqa: E402
+from pr_sync_status import (  # noqa: E402
+    ACTION_BLOCKED,
+    ACTION_MERGE_NOW,
+    ACTION_UPDATE_BRANCH_BY_MERGE,
+    CHECKS_FAILURE,
+    CHECKS_MISSING_REQUIRED,
+    CHECKS_NONE,
+    CHECKS_NOT_REQUIRED,
+    CHECKS_PENDING,
+    CHECKS_SUCCESS,
+    CHECKS_UNKNOWN,
+    assess_pr_sync_status,
+    classify_commit_checks,
+)
+
+
+def _sha(prefix: str) -> str:
+    return (prefix + "0" * 40)[:40]
+
+
+PR_HEAD = _sha("aaaaaaaa")
+BASE_HEAD = _sha("bbbbbbbb")
+
+
+def _base_kwargs(**overrides):
+    data = {
+        "host": "gitea.prgs.cc",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "pr_number": 751,
+        "pr_state": "open",
+        "source_branch": "fix/issue-751-checks-assessor",
+        "pr_head_sha": PR_HEAD,
+        "base_head_sha": BASE_HEAD,
+        "commits_behind": 0,
+        "mergeable": True,
+        "has_conflicts": False,
+        "branch_protection_requires_current_base": False,
+        "approval_at_current_head": True,
+        "checks_status": "success",
+        "active_author_lock": True,
+        "active_reviewer_lease": False,
+        "active_merger_lease": False,
+    }
+    data.update(overrides)
+    return data
+
+
+def _reasons(result) -> str:
+    return " | ".join(result["reasons"]).lower()
+
+
+class TestClassifyCommitChecks(unittest.TestCase):
+    """Pure classification from live evidence."""
+
+    def test_empty_collection_with_combined_pending_is_not_executing_ci(self):
+        # The exact PR #750 failure mode at the classification layer.
+        result = classify_commit_checks(
+            combined_state="pending",
+            statuses=[],
+            checks_enabled=True,
+            required_contexts=[],
+        )
+        self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
+        self.assertEqual(result["checks_status"], CHECKS_NONE)
+        self.assertEqual(result["context_count"], 0)
+        joined = " ".join(result["reasons"]).lower()
+        self.assertIn("empty", joined)
+        self.assertIn("does not indicate", joined)
+
+    def test_protection_disables_status_checks(self):
+        result = classify_commit_checks(
+            combined_state="pending",
+            statuses=[],
+            checks_enabled=False,
+            required_contexts=[],
+        )
+        self.assertFalse(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_NOT_REQUIRED)
+
+    def test_no_required_contexts_aggregates_reported_contexts(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "build", "status": "success"}],
+            checks_enabled=True,
+            required_contexts=[],
+        )
+        self.assertTrue(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
+
+    def test_required_checks_pending(self):
+        result = classify_commit_checks(
+            combined_state="pending",
+            statuses=[{"context": "build", "status": "pending"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_PENDING)
+
+    def test_required_checks_failed(self):
+        result = classify_commit_checks(
+            combined_state="failure",
+            statuses=[{"context": "build", "status": "failure"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_FAILURE)
+
+    def test_required_checks_successful(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "build", "status": "success"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
+        self.assertTrue(result["checks_required"])
+
+    def test_required_context_configured_with_no_matching_result(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "lint", "status": "success"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
+        self.assertEqual(result["missing_required_contexts"], ["build"])
+
+    def test_mixed_required_and_unrelated_contexts_ignores_unrelated(self):
+        # An unrelated failing context must not fail a satisfied required set.
+        result = classify_commit_checks(
+            combined_state="failure",
+            statuses=[
+                {"context": "build", "status": "success"},
+                {"context": "optional-scan", "status": "failure"},
+            ],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
+        # ...and a failing *required* context still fails despite passing extras.
+        failing = classify_commit_checks(
+            combined_state="success",
+            statuses=[
+                {"context": "build", "status": "failure"},
+                {"context": "optional-scan", "status": "success"},
+            ],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(failing["checks_status"], CHECKS_FAILURE)
+
+    def test_policy_unreadable_fails_closed(self):
+        result = classify_commit_checks(
+            combined_state=None,
+            statuses=[],
+            checks_enabled=None,
+            required_contexts=[],
+            policy_determinable=False,
+        )
+        self.assertTrue(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
+        self.assertIn("fail closed", " ".join(result["reasons"]).lower())
+
+    def test_malformed_policy_indeterminate_flag_fails_closed(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "build", "status": "success"}],
+            checks_enabled=None,
+            required_contexts=[],
+            policy_determinable=True,
+        )
+        self.assertTrue(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
+
+    def test_status_collection_unreadable_fails_closed_when_required(self):
+        result = classify_commit_checks(
+            checks_enabled=True,
+            required_contexts=["build"],
+            status_determinable=False,
+        )
+        self.assertTrue(result["checks_required"])
+        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
+
+    def test_unrecognized_context_state_never_reads_as_success(self):
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[{"context": "build", "status": "banana"}],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_UNKNOWN)
+
+    def test_newest_wins_per_context(self):
+        # Gitea returns newest-first; the stale failure must not win.
+        result = classify_commit_checks(
+            combined_state="success",
+            statuses=[
+                {"context": "build", "status": "success"},
+                {"context": "build", "status": "failure"},
+            ],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_SUCCESS)
+
+    def test_malformed_status_rows_are_discarded_not_treated_as_passing(self):
+        result = classify_commit_checks(
+            combined_state="pending",
+            statuses=[{"context": "build"}, "not-a-dict", None],
+            checks_enabled=True,
+            required_contexts=["build"],
+        )
+        self.assertEqual(result["checks_status"], CHECKS_MISSING_REQUIRED)
+
+
+class TestChecksGateSemantics(unittest.TestCase):
+    """``assess_pr_sync_status`` routing and blocker reasons."""
+
+    def test_not_required_allows_merge_now_with_empty_checks(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED),
+            checks_required=False,
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertTrue(result["approval_valid_for_merge"])
+        self.assertFalse(result["checks_required"])
+        self.assertIn("does not require status checks", _reasons(result))
+
+    def test_none_blocks_when_checks_required(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_NONE),
+            checks_required=True,
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("no status context", _reasons(result))
+        self.assertIn("not executing ci", _reasons(result))
+
+    def test_missing_required_blocks(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_MISSING_REQUIRED),
+            checks_required=True,
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("no matching status result", _reasons(result))
+
+    def test_required_pending_blocks(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_PENDING), checks_required=True
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("not finished", _reasons(result))
+
+    def test_required_failure_blocks(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_FAILURE), checks_required=True
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("failed", _reasons(result))
+
+    def test_required_success_merges(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_SUCCESS), checks_required=True
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+
+    def test_unknown_blocks_when_required(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_UNKNOWN), checks_required=True
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("unknown", _reasons(result))
+
+    def test_unrecognized_status_does_not_fall_through_to_merge(self):
+        # Regression: the previous gate only handled a fixed vocabulary and let
+        # anything else reach merge_now.
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status="totally-unexpected"),
+            checks_required=True,
+        )
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertIn("unrecognized checks status", _reasons(result))
+
+    def test_checks_gate_does_not_bypass_approval_requirement(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                checks_status=CHECKS_NOT_REQUIRED, approval_at_current_head=False
+            ),
+            checks_required=False,
+        )
+        self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+        self.assertFalse(result["approval_valid_for_merge"])
+
+    def test_checks_gate_does_not_bypass_current_base_protection(self):
+        # Existing current-base behavior stays intact (#727).
+        result = assess_pr_sync_status(
+            **_base_kwargs(
+                checks_status=CHECKS_NOT_REQUIRED,
+                commits_behind=3,
+                branch_protection_requires_current_base=True,
+            ),
+            checks_required=False,
+        )
+        self.assertEqual(
+            result["recommended_next_action"], ACTION_UPDATE_BRANCH_BY_MERGE
+        )
+
+    def test_checks_gate_does_not_bypass_conflicts(self):
+        result = assess_pr_sync_status(
+            **_base_kwargs(checks_status=CHECKS_NOT_REQUIRED, mergeable=False),
+            checks_required=False,
+        )
+        self.assertNotEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+
+
+class TestBranchProtectionPolicy(unittest.TestCase):
+    """Live protection reader derives the status-check requirement."""
+
+    def setUp(self):
+        import gitea_mcp_server as gms
+
+        self.gms = gms
+
+    def _policy(self, responses):
+        def fake_api_request(method, url, auth, *args, **kwargs):
+            for fragment, payload in responses.items():
+                if fragment in url:
+                    if isinstance(payload, Exception):
+                        raise payload
+                    return payload
+            return None
+
+        with patch.object(self.gms, "api_request", side_effect=fake_api_request):
+            return self.gms._branch_protection_policy(
+                "https://example/api/v1/repos/o/r", {"h": "1"}, base_branch="master"
+            )
+
+    def test_status_checks_enabled_with_contexts(self):
+        policy = self._policy({
+            "branch_protections": [{
+                "branch_name": "master",
+                "block_on_outdated_branch": True,
+                "enable_status_check": True,
+                "status_check_contexts": ["ci/build", " "],
+            }],
+        })
+        self.assertTrue(policy["determinable"])
+        self.assertTrue(policy["checks_enabled"])
+        self.assertTrue(policy["requires_current_base"])
+        self.assertEqual(policy["required_contexts"], ["ci/build"])
+
+    def test_status_checks_disabled(self):
+        policy = self._policy({
+            "branch_protections": [{
+                "branch_name": "master",
+                "enable_status_check": False,
+            }],
+        })
+        self.assertTrue(policy["determinable"])
+        self.assertFalse(policy["checks_enabled"])
+
+    def test_no_protection_rule_means_checks_not_required(self):
+        # PR #750's repository shape: protection list readable but empty.
+        policy = self._policy({"branch_protections": [], "branches/master": {}})
+        self.assertTrue(policy["determinable"])
+        self.assertFalse(policy["protection_found"])
+        self.assertFalse(policy["checks_enabled"])
+        self.assertIsNone(policy["requires_current_base"])
+
+    def test_protection_without_status_field_means_not_enabled(self):
+        policy = self._policy({
+            "branch_protections": [{
+                "branch_name": "master",
+                "block_on_outdated_branch": True,
+            }],
+        })
+        self.assertTrue(policy["determinable"])
+        self.assertFalse(policy["checks_enabled"])
+        self.assertTrue(policy["requires_current_base"])
+
+    def test_api_failure_is_not_determinable(self):
+        policy = self._policy({
+            "branch_protections": RuntimeError("boom"),
+        })
+        self.assertFalse(policy["determinable"])
+        self.assertIsNone(policy["checks_enabled"])
+
+    def test_missing_branch_is_not_determinable(self):
+        with patch.object(self.gms, "api_request", return_value=[]):
+            policy = self.gms._branch_protection_policy(
+                "https://example/api/v1/repos/o/r", {}, base_branch="  "
+            )
+        self.assertFalse(policy["determinable"])
+
+    def test_requires_current_base_accessor_preserved(self):
+        def fake(method, url, auth, *a, **k):
+            if "branch_protections" in url:
+                return [{"branch_name": "master",
+                         "block_on_outdated_branch": True}]
+            return None
+
+        with patch.object(self.gms, "api_request", side_effect=fake):
+            value = self.gms._branch_protection_requires_current_base(
+                "https://example/api/v1/repos/o/r", {}, base_branch="master"
+            )
+        self.assertTrue(value)
+
+
+class TestCommitChecksSnapshot(unittest.TestCase):
+    def setUp(self):
+        import gitea_mcp_server as gms
+
+        self.gms = gms
+
+    def test_reads_state_and_context_collection(self):
+        payload = {"state": "pending", "statuses": [{"context": "b",
+                                                     "status": "pending"}]}
+        with patch.object(self.gms, "api_request", return_value=payload):
+            snap = self.gms._commit_checks_snapshot(
+                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
+            )
+        self.assertTrue(snap["determinable"])
+        self.assertEqual(snap["combined_state"], "pending")
+        self.assertEqual(len(snap["statuses"]), 1)
+
+    def test_empty_collection_recorded_as_determinable(self):
+        with patch.object(
+            self.gms, "api_request", return_value={"state": "pending", "statuses": []}
+        ):
+            snap = self.gms._commit_checks_snapshot(
+                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
+            )
+        self.assertTrue(snap["determinable"])
+        self.assertEqual(snap["statuses"], [])
+
+    def test_api_failure_is_not_determinable(self):
+        with patch.object(self.gms, "api_request", side_effect=RuntimeError("x")):
+            snap = self.gms._commit_checks_snapshot(
+                "https://example/api/v1/repos/o/r", {}, sha=PR_HEAD
+            )
+        self.assertFalse(snap["determinable"])
+
+
+class TestMcpWrapperForwarding(unittest.TestCase):
+    """The production MCP path must reach the derived ``checks_required``."""
+
+    def setUp(self):
+        import gitea_mcp_server as gms
+
+        self.gms = gms
+        self.base = (
+            "https://gitea.prgs.cc/api/v1/repos/Scaled-Tech-Consulting/Gitea-Tools"
+        )
+
+    def _patches(self, fake_api_request, spy):
+        return [
+            patch.object(self.gms, "_profile_operation_gate", return_value=None),
+            patch.object(self.gms, "_resolve", return_value=(
+                "gitea.prgs.cc", "Scaled-Tech-Consulting", "Gitea-Tools")),
+            patch.object(self.gms, "_auth", return_value={"Authorization": "x"}),
+            patch.object(self.gms, "repo_api_url", return_value=self.base),
+            patch.object(self.gms, "api_request", side_effect=fake_api_request),
+            patch.object(self.gms, "gitea_get_pr_review_feedback", return_value={
+                "success": True, "approval_at_current_head": True}),
+            patch.object(self.gms, "_prove_author_ownership_for_pr", return_value={
+                "has_author_lock": True}),
+            patch.object(pr_sync_status, "assess_pr_sync_status", side_effect=spy),
+        ]
+
+    def _run(self, *, protections, status_payload, pr_number=750,
+             caller_checks_status=None):
+        captured = {}
+        real_assess = pr_sync_status.assess_pr_sync_status
+
+        def spy(**kwargs):
+            captured.update(kwargs)
+            return real_assess(**kwargs)
+
+        def fake_api_request(method, url, auth, *args, **kwargs):
+            if f"/pulls/{pr_number}" in url:
+                return {
+                    "number": pr_number,
+                    "state": "open",
+                    "title": "t",
+                    "body": "b",
+                    "mergeable": True,
+                    "head": {"sha": PR_HEAD, "ref": "fix/x"},
+                    "base": {"sha": BASE_HEAD, "ref": "master"},
+                }
+            if "/branch_protections" in url:
+                if isinstance(protections, Exception):
+                    raise protections
+                return protections
+            if "/branches/master" in url:
+                return {"commit": {"id": BASE_HEAD}}
+            if "/status" in url:
+                if isinstance(status_payload, Exception):
+                    raise status_payload
+                return status_payload
+            if "/compare/" in url:
+                return {"total_commits": 0}
+            if "/comments" in url:
+                return []
+            return None
+
+        stack = self._patches(fake_api_request, spy)
+        for p in stack:
+            p.start()
+        try:
+            result = self.gms.gitea_assess_pr_sync_status(
+                pr_number=pr_number, remote="prgs",
+                checks_status=caller_checks_status,
+            )
+        finally:
+            for p in reversed(stack):
+                p.stop()
+        return result, captured
+
+    def test_derived_checks_required_is_forwarded(self):
+        result, captured = self._run(
+            protections=[{"branch_name": "master", "enable_status_check": True,
+                          "status_check_contexts": ["ci/build"]}],
+            status_payload={"state": "pending", "statuses": [
+                {"context": "ci/build", "status": "pending"}]},
+        )
+        self.assertIn("checks_required", captured)
+        self.assertTrue(captured["checks_required"])
+        self.assertEqual(captured["checks_status"], CHECKS_PENDING)
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_pr750_empty_status_with_no_protection_is_merge_ready(self):
+        # The exact reported failure: combined pending, zero contexts, and no
+        # branch protection requiring checks.
+        result, captured = self._run(
+            protections=[],
+            status_payload={"state": "pending", "statuses": []},
+        )
+        self.assertFalse(captured["checks_required"])
+        self.assertEqual(captured["checks_status"], CHECKS_NOT_REQUIRED)
+        self.assertNotEqual(result["checks_status"], CHECKS_PENDING)
+        self.assertEqual(result["recommended_next_action"], ACTION_MERGE_NOW)
+
+    def test_protection_read_failure_blocks(self):
+        result, captured = self._run(
+            protections=RuntimeError("protection unavailable"),
+            status_payload={"state": "success", "statuses": []},
+        )
+        self.assertTrue(captured["checks_required"])
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+
+    def test_caller_supplied_status_cannot_mask_live_required_failure(self):
+        # A caller-declared "success" must not override live evidence.
+        result, captured = self._run(
+            protections=[{"branch_name": "master", "enable_status_check": True,
+                          "status_check_contexts": ["ci/build"]}],
+            status_payload={"state": "failure", "statuses": [
+                {"context": "ci/build", "status": "failure"}]},
+            caller_checks_status="success",
+        )
+        self.assertEqual(captured["checks_status"], CHECKS_FAILURE)
+        self.assertEqual(result["recommended_next_action"], ACTION_BLOCKED)
+        self.assertEqual(
+            result["checks_evidence"]["caller_supplied_checks_status"], "success"
+        )
+
+    def test_checks_evidence_is_reported_without_secrets(self):
+        result, _ = self._run(
+            protections=[],
+            status_payload={"state": "pending", "statuses": []},
+        )
+        evidence = result["checks_evidence"]
+        self.assertEqual(evidence["context_count"], 0)
+        self.assertEqual(evidence["combined_state"], "pending")
+        self.assertFalse(evidence["protection_found"])
+        blob = repr(result).lower()
+        self.assertNotIn("authorization", blob)
+        self.assertNotIn("token", blob)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_753_dead_pid_lock_recovery.py

@@ -0,0 +1,364 @@
+"""Dead-session author issue-lock recovery (#753).
+
+Covers the narrow recovery path that lets an author re-acquire a durable lock
+after the MCP session that recorded it exits, plus every rejection condition
+that must keep failing closed.
+"""
+
+import os
+import subprocess
+import sys
+import unittest
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import issue_lock_recovery  # noqa: E402
+import issue_lock_store  # noqa: E402
+import issue_lock_worktree  # noqa: E402
+
+ISSUE = 4242
+BRANCH = f"fix/issue-{ISSUE}-demo"
+WORKTREE = "/scratch/wt"
+HEAD = "a" * 40
+OTHER_SHA = "b" * 40
+IDENTITY = "example-user"
+PROFILE = "example-author"
+
+
+def dead_pid() -> int:
+    """A PID that has certainly exited (spawned, then reaped)."""
+    proc = subprocess.Popen([sys.executable, "-c", "pass"])
+    proc.wait()
+    return proc.pid
+
+
+def future_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def make_lock(**overrides):
+    lock = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": "prgs",
+        "org": "ExampleOrg",
+        "repo": "ExampleRepo",
+        "session_pid": dead_pid(),
+        "work_lease": {
+            "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+            "issue_number": ISSUE,
+            "branch": BRANCH,
+            "worktree_path": WORKTREE,
+            "claimant": {"username": IDENTITY, "profile": PROFILE},
+            "expires_at": future_ts(),
+        },
+    }
+    lock.update(overrides)
+    return lock
+
+
+def assess(lock=None, **overrides):
+    kwargs = {
+        "issue_number": ISSUE,
+        "branch_name": BRANCH,
+        "worktree_path": WORKTREE,
+        "remote": "prgs",
+        "org": "ExampleOrg",
+        "repo": "ExampleRepo",
+        "identity": IDENTITY,
+        "profile": PROFILE,
+        "current_branch": BRANCH,
+        "porcelain_status": "",
+        "head_sha": HEAD,
+        "remote_head_sha": HEAD,
+        "pr_head_sha": HEAD,
+        "pr_number": 99,
+        "competing_live_locks": [],
+        "candidate_branches": [BRANCH],
+        "current_pid": os.getpid(),
+    }
+    kwargs.update(overrides)
+    return issue_lock_recovery.assess_dead_session_lock_recovery(
+        make_lock() if lock is None else lock, **kwargs
+    )
+
+
+class TestDeadSessionRecoveryGranted(unittest.TestCase):
+    def test_dead_pid_with_exact_evidence_recovers(self):
+        result = assess()
+        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
+        self.assertEqual(result["outcome"], issue_lock_recovery.RECOVERY_SANCTIONED)
+
+    def test_recovery_still_granted_when_no_open_pr_exists(self):
+        # A locked branch need not have a PR yet; absence must not block.
+        result = assess(pr_head_sha=None, pr_number=None)
+        self.assertTrue(result["recovery_sanctioned"], result["reasons"])
+
+    def test_lease_expiry_is_not_required_for_recovery(self):
+        # The defining condition is PID death, not TTL expiry (the #601 gap).
+        lock = make_lock()
+        self.assertFalse(issue_lock_store.is_lease_expired(lock))
+        self.assertFalse(issue_lock_store.assess_lock_freshness(lock)["live"])
+        self.assertTrue(assess(lock)["recovery_sanctioned"])
+
+
+class TestDeadSessionRecoveryRefused(unittest.TestCase):
+    def assert_refused(self, result, needle):
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertEqual(result["outcome"], issue_lock_recovery.REFUSED)
+        self.assertTrue(
+            any(needle in reason for reason in result["reasons"]),
+            f"expected {needle!r} in {result['reasons']}",
+        )
+
+    def test_live_prior_pid_refused(self):
+        lock = make_lock(session_pid=os.getpid(), pid=os.getpid())
+        # Distinct current pid so the refusal is attributable to liveness.
+        self.assert_refused(assess(lock, current_pid=os.getpid() + 1), "still alive")
+
+    def test_different_author_identity_refused(self):
+        self.assert_refused(
+            assess(identity="someone-else"), "does not match active identity"
+        )
+
+    def test_different_profile_refused(self):
+        self.assert_refused(
+            assess(profile="other-profile"), "does not match active profile"
+        )
+
+    def test_different_branch_refused(self):
+        lock = make_lock(branch_name=f"fix/issue-{ISSUE}-other")
+        self.assert_refused(assess(lock), "does not match requested")
+
+    def test_worktree_parked_on_another_branch_refused(self):
+        self.assert_refused(assess(current_branch="master"), "not the locked branch")
+
+    def test_detached_head_worktree_refused(self):
+        self.assert_refused(assess(current_branch=None), "detached HEAD")
+
+    def test_different_worktree_refused(self):
+        self.assert_refused(
+            assess(worktree_path="/scratch/elsewhere"), "does not match declared"
+        )
+
+    def test_dirty_worktree_refused(self):
+        self.assert_refused(
+            assess(porcelain_status=" M gitea_mcp_server.py\n"), "requires a clean"
+        )
+
+    def test_local_head_differing_from_remote_refused(self):
+        self.assert_refused(
+            assess(remote_head_sha=OTHER_SHA), "does not match remote branch head"
+        )
+
+    def test_pr_head_differing_refused(self):
+        self.assert_refused(assess(pr_head_sha=OTHER_SHA), "does not match local head")
+
+    def test_missing_remote_head_refused(self):
+        self.assert_refused(assess(remote_head_sha=None), "remote head")
+
+    def test_competing_live_lock_refused(self):
+        competing = [
+            {
+                "issue_number": ISSUE,
+                "branch_name": BRANCH,
+                "worktree_path": "/scratch/other-wt",
+                "pid": os.getpid(),
+            }
+        ]
+        self.assert_refused(
+            assess(competing_live_locks=competing), "competing live lock"
+        )
+
+    def test_unrelated_live_lock_does_not_block(self):
+        unrelated = [
+            {
+                "issue_number": 999,
+                "branch_name": "fix/issue-999-unrelated",
+                "worktree_path": "/scratch/unrelated",
+                "pid": os.getpid(),
+            }
+        ]
+        self.assertTrue(assess(competing_live_locks=unrelated)["recovery_sanctioned"])
+
+    def test_multiple_candidate_branches_refused(self):
+        self.assert_refused(
+            assess(candidate_branches=[BRANCH, f"feat/issue-{ISSUE}-rival"]),
+            "multiple branches claim this issue",
+        )
+
+    def test_repository_scope_mismatch_refused(self):
+        self.assert_refused(assess(repo="OtherRepo"), "does not match requested")
+
+    def test_malformed_lock_missing_worktree_refused(self):
+        lock = make_lock()
+        lock.pop("worktree_path")
+        self.assert_refused(assess(lock), "incomplete")
+
+    def test_malformed_lock_missing_pid_refused(self):
+        lock = make_lock()
+        lock.pop("session_pid", None)
+        lock.pop("pid", None)
+        self.assert_refused(assess(lock), "incomplete")
+
+    def test_lock_without_claimant_refused(self):
+        lock = make_lock()
+        lock["work_lease"] = dict(lock["work_lease"])
+        lock["work_lease"].pop("claimant")
+        self.assert_refused(assess(lock), "claimant identity/profile")
+
+
+class TestNotACandidate(unittest.TestCase):
+    def test_absent_lock_is_not_a_candidate(self):
+        result = issue_lock_recovery.assess_dead_session_lock_recovery(
+            None,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=WORKTREE,
+            remote="prgs",
+            org="ExampleOrg",
+            repo="ExampleRepo",
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            head_sha=HEAD,
+            remote_head_sha=HEAD,
+        )
+        self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE)
+        self.assertFalse(result["recovery_sanctioned"])
+        self.assertFalse(result["is_candidate"])
+
+    def test_lock_for_a_different_issue_is_not_a_candidate(self):
+        result = assess(make_lock(issue_number=7777))
+        self.assertEqual(result["outcome"], issue_lock_recovery.NO_CANDIDATE)
+        self.assertFalse(result["recovery_sanctioned"])
+
+
+class TestWorktreeGateWaiver(unittest.TestCase):
+    def test_new_issue_claim_still_requires_base_equivalence(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path=WORKTREE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            base_equivalent=False,
+        )
+        self.assertTrue(result["block"])
+        self.assertFalse(result["base_equivalence_waived"])
+
+    def test_sanctioned_recovery_waives_base_equivalence(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path=WORKTREE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            base_equivalent=False,
+            recovery_sanctioned=True,
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+        self.assertTrue(result["base_equivalence_waived"])
+
+    def test_recovery_never_waives_cleanliness(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path=WORKTREE,
+            current_branch=BRANCH,
+            porcelain_status=" M gitea_mcp_server.py\n",
+            base_equivalent=False,
+            recovery_sanctioned=True,
+        )
+        self.assertTrue(result["block"])
+        self.assertTrue(
+            any("tracked file edits" in reason for reason in result["reasons"])
+        )
+
+    def test_unproven_base_equivalence_still_blocks_without_recovery(self):
+        result = issue_lock_worktree.assess_issue_lock_worktree(
+            worktree_path=WORKTREE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            base_equivalent=None,
+        )
+        self.assertTrue(result["block"])
+
+
+class TestRecoveryRecordAndDownstream(unittest.TestCase):
+    def test_recovery_record_preserves_truthful_provenance(self):
+        assessment = assess()
+        prior = assessment["evidence"]["prior_session_pid"]
+        record = issue_lock_recovery.build_recovery_record(
+            assessment, recovered_at="2026-07-18T23:21:40Z"
+        )
+        self.assertTrue(record["recovered"])
+        self.assertEqual(record["prior_session_pid"], prior)
+        self.assertEqual(record["replacement_session_pid"], os.getpid())
+        self.assertNotEqual(
+            record["prior_session_pid"], record["replacement_session_pid"]
+        )
+        self.assertFalse(record["prior_pid_alive"])
+        self.assertEqual(record["recovered_at"], "2026-07-18T23:21:40Z")
+        self.assertEqual(record["branch_name"], BRANCH)
+        self.assertEqual(record["local_head"], HEAD)
+        self.assertEqual(record["identity"], IDENTITY)
+        self.assertTrue(record["proof"])
+
+    def test_recovery_record_carries_no_secret_material(self):
+        record = issue_lock_recovery.build_recovery_record(
+            assess(), recovered_at="2026-07-18T23:21:40Z"
+        )
+        blob = repr(record).lower()
+        for banned in ("token", "password", "authorization", "secret", "api_key"):
+            self.assertNotIn(banned, blob)
+
+    def test_recovered_lock_satisfies_update_by_merge_ownership(self):
+        # After recovery the lock is rebound to the live session, so the
+        # ownership re-check used by gitea_update_pr_branch_by_merge passes.
+        assessment = assess()
+        recovered_lock = make_lock(session_pid=os.getpid(), pid=os.getpid())
+        recovered_lock["dead_session_recovery"] = (
+            issue_lock_recovery.build_recovery_record(
+                assessment, recovered_at="2026-07-18T23:21:40Z"
+            )
+        )
+        freshness = issue_lock_store.assess_lock_freshness(recovered_lock)
+        self.assertTrue(freshness["live"], freshness)
+
+        verdict = issue_lock_store.verify_lock_for_mutation(
+            recovered_lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=WORKTREE,
+        )
+        self.assertTrue(verdict["proven"], verdict["reasons"])
+        self.assertFalse(verdict["block"])
+
+    def test_pre_recovery_lock_fails_ownership_check(self):
+        # Guards against a false positive above: the dead-PID lock must fail.
+        verdict = issue_lock_store.verify_lock_for_mutation(
+            make_lock(),
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path=WORKTREE,
+        )
+        self.assertTrue(verdict["block"])
+        self.assertTrue(any("not live" in reason for reason in verdict["reasons"]))
+
+    def test_no_manual_file_seeding_required(self):
+        # The whole decision is reachable from the durable record plus live
+        # observation; nothing is written to disk to reach a verdict.
+        self.assertTrue(assess()["recovery_sanctioned"])
+
+    def test_refusal_message_is_fail_closed(self):
+        message = issue_lock_recovery.format_recovery_refusal(
+            assess(porcelain_status=" M x.py\n")
+        )
+        self.assertIn("fail closed", message)
+        self.assertIn("recovery refused", message.lower())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_issue_755_owning_pr_recovery.py

@@ -0,0 +1,609 @@
+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
+"""Dead-session lock recovery when the issue already owns an open PR (#755).
+
+#753 added the recovery *assessor*, but the production ``gitea_lock_issue``
+path still rejected every sanctioned recovery: a dead-session lock is by
+construction a lock for work that already has an open PR, and the #400
+duplicate-work gate blocked unconditionally on any linked open PR. These tests
+drive the real MCP handler, not just the pure assessor, so that gap cannot
+reopen.
+"""
+import os
+import subprocess
+import sys
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import issue_lock_provenance  # noqa: E402
+import issue_lock_recovery  # noqa: E402
+import issue_lock_store  # noqa: E402
+import mcp_server  # noqa: E402
+from issue_work_duplicate_gate import (  # noqa: E402
+    OUTCOME_DUPLICATE_PR_PREVENTED,
+    OUTCOME_DUPLICATE_WORK_NOT_PREVENTED,
+    PHASE_LOCK,
+    assess_work_issue_duplicate_gate,
+)
+
+ISSUE = 4755
+BRANCH = f"fix/issue-{ISSUE}-owning-pr"
+OTHER_BRANCH = f"fix/issue-{ISSUE}-competing"
+HEAD = "c" * 40
+OTHER_HEAD = "d" * 40
+OWNING_PR = 4756
+OTHER_PR = 4757
+IDENTITY = "example-user"
+PROFILE = "test-author-prgs"
+ORG = "Scaled-Tech-Consulting"
+REPO = "Gitea-Tools"
+
+
+def dead_pid() -> int:
+    """A PID that has certainly exited (spawned, then reaped)."""
+    proc = subprocess.Popen([sys.executable, "-c", "pass"])
+    proc.wait()
+    return proc.pid
+
+
+def shifted_ts(hours: int = 4) -> str:
+    return (
+        (datetime.now(timezone.utc) + timedelta(hours=hours))
+        .isoformat()
+        .replace("+00:00", "Z")
+    )
+
+
+def owning_pr(number=OWNING_PR, ref=BRANCH, sha=HEAD, issue=ISSUE):
+    return {
+        "number": number,
+        "title": f"fix: something (Closes #{issue})",
+        "body": f"Closes #{issue}.",
+        "head": {"ref": ref, "sha": sha},
+    }
+
+
+def sanctioned_token(
+    issue_number=ISSUE, pr_number=OWNING_PR, branch=BRANCH, head=HEAD
+):
+    """The evidence shape the server derives from a granted recovery."""
+    return {
+        "issue_number": issue_number,
+        "pr_number": pr_number,
+        "branch_name": branch,
+        "head_sha": head,
+    }
+
+
+# ───────────────────────── duplicate gate: exemption ─────────────────────────
+
+
+class TestOwningPrExemptionGranted(unittest.TestCase):
+    def test_exact_owning_pr_is_not_duplicate_work(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_WORK_NOT_PREVENTED)
+        self.assertEqual(result["linked_open_pr"], OWNING_PR)
+        self.assertEqual(result["linked_open_pr_count"], 1)
+
+    def test_unrelated_open_pr_alongside_owning_pr_is_ignored(self):
+        unrelated = {
+            "number": 999,
+            "title": "chore: unrelated",
+            "body": "no linkage",
+            "head": {"ref": "chore/unrelated", "sha": OTHER_HEAD},
+        }
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[unrelated, owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["linked_open_pr_count"], 1)
+
+
+class TestOwningPrExemptionRefused(unittest.TestCase):
+    def assert_blocked(self, result):
+        self.assertTrue(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+        self.assertEqual(result["outcome"], OUTCOME_DUPLICATE_PR_PREVENTED)
+
+    def test_no_recovery_evidence_keeps_ordinary_blocker(self):
+        self.assert_blocked(
+            assess_work_issue_duplicate_gate(
+                ISSUE,
+                open_prs=[owning_pr()],
+                branch_names=[BRANCH],
+                claim_entry={"status": "not_claimed"},
+                locked_branch=BRANCH,
+                phase=PHASE_LOCK,
+            )
+        )
+
+    def test_competing_pr_number_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(number=OTHER_PR)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_multiple_linked_open_prs_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+        self.assertEqual(result["linked_open_pr_count"], 2)
+
+    def test_different_branch_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_locked_branch_differing_from_evidence_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=OTHER_BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_different_head_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr(sha=OTHER_HEAD)],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+    def test_evidence_for_another_issue_refused(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(issue_number=ISSUE + 1),
+        )
+        self.assert_blocked(result)
+
+    def test_missing_head_in_live_pr_refused(self):
+        pr = owning_pr()
+        pr["head"] = {"ref": BRANCH}
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[pr],
+            branch_names=[BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        self.assert_blocked(result)
+
+
+class TestOrdinaryDuplicateBehaviorUnchanged(unittest.TestCase):
+    def test_clean_issue_still_passes(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[],
+            branch_names=["feat/other-issue-99"],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+        )
+        self.assertFalse(result["block"])
+        self.assertFalse(result["owning_pr_recovery_exempted"])
+
+    def test_competing_branch_still_blocks_even_with_owning_pr_evidence(self):
+        result = assess_work_issue_duplicate_gate(
+            ISSUE,
+            open_prs=[owning_pr()],
+            branch_names=[BRANCH, OTHER_BRANCH],
+            claim_entry={"status": "not_claimed"},
+            locked_branch=BRANCH,
+            phase=PHASE_LOCK,
+            recovered_owning_pr=sanctioned_token(),
+        )
+        # The owning PR is exempt, but the competing branch is not.
+        self.assertTrue(result["block"])
+        self.assertTrue(result["owning_pr_recovery_exempted"])
+        self.assertIn(OTHER_BRANCH, result["conflicting_branches"])
+
+
+# ─────────────────── server-derived evidence cannot be forged ───────────────────
+
+
+class TestOwningPrEvidenceDerivation(unittest.TestCase):
+    def granted(self, **evidence_overrides):
+        evidence = {
+            "issue_number": ISSUE,
+            "locked_branch": BRANCH,
+            "local_head": HEAD,
+            "remote_head": HEAD,
+            "pr_head": HEAD,
+            "pr_number": OWNING_PR,
+        }
+        evidence.update(evidence_overrides)
+        return {
+            "outcome": issue_lock_recovery.RECOVERY_SANCTIONED,
+            "recovery_sanctioned": True,
+            "is_candidate": True,
+            "reasons": [],
+            "evidence": evidence,
+        }
+
+    def test_granted_recovery_yields_evidence(self):
+        token = issue_lock_recovery.owning_pr_recovery_evidence(self.granted())
+        self.assertEqual(token, sanctioned_token())
+
+    def test_none_assessment_yields_nothing(self):
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(None))
+
+    def test_refused_assessment_yields_nothing(self):
+        refused = self.granted()
+        refused["outcome"] = issue_lock_recovery.REFUSED
+        refused["recovery_sanctioned"] = False
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(refused))
+
+    def test_sanctioned_flag_without_outcome_yields_nothing(self):
+        forged = self.granted()
+        forged["outcome"] = "SOMETHING_ELSE"
+        self.assertIsNone(issue_lock_recovery.owning_pr_recovery_evidence(forged))
+
+    def test_missing_pr_number_yields_nothing(self):
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(pr_number=None)
+            )
+        )
+
+    def test_head_disagreement_in_evidence_yields_nothing(self):
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(remote_head=OTHER_HEAD)
+            )
+        )
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(
+                self.granted(local_head=OTHER_HEAD)
+            )
+        )
+
+    def test_real_refused_assessment_yields_nothing(self):
+        """End-to-end against the real assessor, not a hand-built dict."""
+        lock = {
+            "issue_number": ISSUE,
+            "branch_name": BRANCH,
+            "worktree_path": "/scratch/wt",
+            "remote": "prgs",
+            "org": "Example-Org",
+            "repo": "Example-Repo",
+            "session_pid": os.getpid(),  # alive → must refuse
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": ISSUE,
+                "branch": BRANCH,
+                "worktree_path": "/scratch/wt",
+                "claimant": {"username": IDENTITY, "profile": PROFILE},
+                "expires_at": shifted_ts(),
+            },
+        }
+        assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
+            lock,
+            issue_number=ISSUE,
+            branch_name=BRANCH,
+            worktree_path="/scratch/wt",
+            remote="prgs",
+            org="Example-Org",
+            repo="Example-Repo",
+            identity=IDENTITY,
+            profile=PROFILE,
+            current_branch=BRANCH,
+            porcelain_status="",
+            head_sha=HEAD,
+            remote_head_sha=HEAD,
+            pr_head_sha=HEAD,
+            pr_number=OWNING_PR,
+            competing_live_locks=[],
+            candidate_branches=[BRANCH],
+            current_pid=os.getpid(),
+        )
+        self.assertFalse(assessment["recovery_sanctioned"])
+        self.assertIsNone(
+            issue_lock_recovery.owning_pr_recovery_evidence(assessment)
+        )
+
+
+# ──────────────────── end-to-end: the real gitea_lock_issue ────────────────────
+
+
+class LockIssueEndToEndBase(unittest.TestCase):
+    """Drives ``mcp_server.gitea_lock_issue`` with live git/Gitea observation
+    stubbed at the module boundary — the production gate chain itself runs."""
+
+    def setUp(self):
+        self.lock_dir = tempfile.TemporaryDirectory()
+        self.addCleanup(self.lock_dir.cleanup)
+        self.worktree = os.path.realpath(os.getcwd())
+        # Bind host/org/repo to what the ``test-author-prgs`` fixture profile is
+        # pinned to, so the session-context gate under test is the real one and
+        # not a cross-host denial. The issue number and lock dir stay synthetic.
+        self.remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.prgs.cc",
+                "org": ORG,
+                "repo": REPO,
+            },
+        })
+        self.remotes.start()
+        self.addCleanup(patch.stopall)
+        mcp_server._IDENTITY_CACHE.clear()
+
+    def write_durable_lock(self, *, pid, branch=BRANCH, worktree=None):
+        path = issue_lock_store.lock_file_path(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        claimant = {"username": IDENTITY, "profile": PROFILE}
+        data = {
+            "issue_number": ISSUE,
+            "branch_name": branch,
+            "remote": "prgs",
+            "org": ORG,
+            "repo": REPO,
+            "worktree_path": worktree or self.worktree,
+            "session_pid": pid,
+            "pid": pid,
+            "work_lease": {
+                "operation_type": issue_lock_store.AUTHOR_ISSUE_WORK_LEASE,
+                "issue_number": ISSUE,
+                "branch": branch,
+                "worktree_path": worktree or self.worktree,
+                "claimant": claimant,
+                "created_at": shifted_ts(-1),
+                "last_heartbeat_at": shifted_ts(-1),
+                "expires_at": shifted_ts(),
+            },
+            "lock_provenance": issue_lock_provenance.build_sanctioned_lock_provenance(
+                tool="gitea_lock_issue",
+                claimant=claimant,
+            ),
+        }
+        issue_lock_store.save_lock_file(path, data)
+        return path
+
+    def run_lock(
+        self,
+        *,
+        open_prs,
+        porcelain="",
+        current_branch=BRANCH,
+        base_equivalent=False,
+        branch_names=None,
+        head_sha=HEAD,
+        remote_head=HEAD,
+    ):
+        branch_names = branch_names if branch_names is not None else [BRANCH]
+        branch_entries = [
+            {"name": name, "commit": {"id": remote_head}} for name in branch_names
+        ]
+        env = shared_mutation_env(
+            "test-author-prgs",
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        with patch(
+            "mcp_server.api_get_all", return_value=branch_entries
+        ), patch(
+            "mcp_server._list_open_pulls", return_value=list(open_prs)
+        ), patch(
+            "mcp_server.get_auth_header", return_value="token x"
+        ), patch(
+            "mcp_server._work_lease_claimant",
+            return_value={"username": IDENTITY, "profile": PROFILE},
+        ), patch(
+            "mcp_server.issue_lock_worktree.read_worktree_git_state",
+            return_value={
+                "current_branch": current_branch,
+                "porcelain_status": porcelain,
+                "base_equivalent": base_equivalent,
+                "head_sha": head_sha,
+                "inspected_git_root": self.worktree,
+                "base_branch": "master",
+            },
+        ), patch(
+            "mcp_server.issue_duplicate_context_fetcher",
+            side_effect=lambda h, o, r, auth, issue_number: (
+                list(open_prs), list(branch_names), {"status": "not_claimed"}
+            ),
+        ):
+            with patch.dict(os.environ, env, clear=True):
+                os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+                return mcp_server.gitea_lock_issue(
+                    issue_number=ISSUE,
+                    branch_name=BRANCH,
+                    remote="prgs",
+                    worktree_path=self.worktree,
+                )
+
+
+class TestRecoveryWithOwningPrSucceeds(LockIssueEndToEndBase):
+    def test_dead_session_recovery_with_owning_pr_relocks(self):
+        self.write_durable_lock(pid=dead_pid())
+        result = self.run_lock(open_prs=[owning_pr()])
+
+        self.assertTrue(result["success"])
+        self.assertEqual(result["issue_number"], ISSUE)
+        self.assertEqual(result["branch_name"], BRANCH)
+        self.assertTrue(result["lock_freshness"]["live"])
+        self.assertTrue(result["lock_freshness"]["pid_alive"])
+
+    def test_recovered_lock_records_truthful_provenance(self):
+        prior = dead_pid()
+        self.write_durable_lock(pid=prior)
+        self.run_lock(open_prs=[owning_pr()])
+
+        lock = issue_lock_store.load_issue_lock(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        record = lock.get("dead_session_recovery") or {}
+        self.assertTrue(record.get("recovered"))
+        self.assertEqual(record.get("prior_session_pid"), prior)
+        self.assertEqual(record.get("replacement_session_pid"), os.getpid())
+        self.assertFalse(record.get("prior_pid_alive"))
+        self.assertEqual(record.get("pr_number"), OWNING_PR)
+        self.assertEqual(record.get("branch_name"), BRANCH)
+        self.assertEqual(record.get("identity"), IDENTITY)
+
+    def test_recovered_lock_is_live_and_proves_pr_ownership(self):
+        """AC6: the persisted lock satisfies update-by-merge's ownership prover."""
+        self.write_durable_lock(pid=dead_pid())
+        self.run_lock(open_prs=[owning_pr()])
+
+        lock = issue_lock_store.load_issue_lock(
+            remote="prgs",
+            org=ORG,
+            repo=REPO,
+            issue_number=ISSUE,
+            lock_dir=self.lock_dir.name,
+        )
+        self.assertTrue(issue_lock_store.is_lease_live(lock))
+
+        env = shared_mutation_env(
+            "test-author-prgs",
+            include_example_repo=True,
+            GITEA_ISSUE_LOCK_DIR=self.lock_dir.name,
+        )
+        with patch.dict(os.environ, env, clear=True):
+            os.environ["GITEA_ISSUE_LOCK_DIR"] = self.lock_dir.name
+            ownership = mcp_server._prove_author_ownership_for_pr(
+                pr_number=OWNING_PR,
+                pr_title=f"fix: something (Closes #{ISSUE})",
+                pr_body=f"Closes #{ISSUE}.",
+                source_branch=BRANCH,
+                remote="prgs",
+                host="gitea.prgs.cc",
+                org=ORG,
+                repo=REPO,
+                worktree_path=self.worktree,
+            )
+        self.assertTrue(ownership["proven"], ownership["reasons"])
+        self.assertTrue(ownership["has_author_lock"])
+        self.assertEqual(ownership["matched_issue"], ISSUE)
+
+
+class TestRecoveryRejectionsEndToEnd(LockIssueEndToEndBase):
+    def assert_lock_refused(self, **kwargs):
+        with self.assertRaises((ValueError, RuntimeError)) as ctx:
+            self.run_lock(**kwargs)
+        return str(ctx.exception)
+
+    def test_competing_pr_still_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        message = self.assert_lock_refused(
+            open_prs=[owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+            branch_names=[BRANCH],
+        )
+        self.assertIn("already covers issue", message)
+
+    def test_multiple_linked_open_prs_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        message = self.assert_lock_refused(
+            open_prs=[owning_pr(), owning_pr(number=OTHER_PR, ref=OTHER_BRANCH)],
+        )
+        self.assertIn("already covers issue", message)
+
+    def test_owning_pr_on_a_different_head_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(open_prs=[owning_pr(sha=OTHER_HEAD)])
+
+    def test_lock_registered_to_a_different_worktree_blocked(self):
+        self.write_durable_lock(
+            pid=dead_pid(), worktree=os.path.join(self.worktree, "elsewhere")
+        )
+        self.assert_lock_refused(open_prs=[owning_pr()])
+
+    def test_dirty_worktree_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], porcelain=" M gitea_mcp_server.py"
+        )
+
+    def test_worktree_parked_on_another_branch_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], current_branch="master"
+        )
+
+    def test_local_head_differing_from_remote_blocked(self):
+        self.write_durable_lock(pid=dead_pid())
+        self.assert_lock_refused(
+            open_prs=[owning_pr()], head_sha=OTHER_HEAD
+        )
+
+    def test_live_prior_pid_blocked(self):
+        self.write_durable_lock(pid=os.getpid())
+        self.assert_lock_refused(open_prs=[owning_pr()])
+
+    def test_new_claim_without_prior_lock_still_requires_base_equivalence(self):
+        """AC10: no durable lock → no recovery → base-equivalence still rules."""
+        self.assert_lock_refused(open_prs=[], branch_names=[])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_merger_lease_finalization.py

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

tests/test_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")

tests/test_reviewer_pr_lease.py

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

workflow_scope_guard.py

@@ -278,12 +278,19 @@ def assess_root_source_mutation(
     current_branch: str | None = None,
     locked_issue_number: int | None = None,
     role_kind: str | None = None,
+    mutation_task: str | None = None,
 ) -> dict[str, Any]:
     """Fail closed for diagnostic/source edits on the control/root checkout.
 
     Allowed only when the active workspace is under ``branches/``. Dirty
     tracked source/test files on the control checkout always block, including
     temporary/diagnostic/test-only intent.
+
+    #749: ``create_issue`` is a pure remote mutation with no local tree write.
+    When *mutation_task* is create_issue and the control checkout has no dirty
+    source/test files, the missing-worktree signal is suppressed so the
+    sanctioned bootstrap path can proceed. Dirty roots and every other task
+    still fail closed.
     """
     role = (role_kind or "").strip().lower()
     if role == "reconciler":
@@ -302,6 +309,7 @@ def assess_root_source_mutation(
     dirty_src = dirty_source_files(porcelain_status)
     reasons: list[str] = []
     blocker_kind: str | None = None
+    create_issue_bootstrap = False
 
     if not under_branches and workspace == root and dirty_src:
         # Root workspace with source dirtiness is unattributed root WIP.
@@ -313,32 +321,50 @@ def assess_root_source_mutation(
         )
         blocker_kind = BLOCKER_ROOT_DIAGNOSTIC_EDIT
 
+    # Lazy import keeps workflow_scope_guard free of circular import at module load.
+    try:
+        import create_issue_bootstrap as _cib
+    except Exception:  # pragma: no cover - import always available in-tree
+        _cib = None
+
     if (
         not under_branches
         and workspace == root
         and not dirty_src
         and role == "author"
     ):
-        # Explicit missing-worktree signal for force-on author entrypoints.
+        if _cib is not None and _cib.is_create_issue_task(mutation_task):
-        reasons.append(
+            # #749: clean-root create_issue is the sanctioned bootstrap path.
-            "author source/test mutation from the stable control checkout is "
+            create_issue_bootstrap = True
-            "forbidden; bind an issue-backed worktree under branches/ first"
+        else:
-        )
+            # Explicit missing-worktree signal for force-on author entrypoints.
-        blocker_kind = BLOCKER_MISSING_WORKTREE
+            reasons.append(
+                "author source/test mutation from the stable control checkout is "
+                "forbidden; bind an issue-backed worktree under branches/ first"
+            )
+            blocker_kind = BLOCKER_MISSING_WORKTREE
 
     if reasons:
         kind = blocker_kind or BLOCKER_ROOT_DIAGNOSTIC_EDIT
+        next_action = _NEXT_ACTIONS[kind]
+        if (
+            kind == BLOCKER_ROOT_DIAGNOSTIC_EDIT
+            and _cib is not None
+            and _cib.is_create_issue_task(mutation_task)
+        ):
+            next_action = _cib.EXACT_NEXT_ACTION_BOOTSTRAP
         return {
             "proven": False,
             "block": True,
             "blocker_kind": kind,
-            "exact_next_action": _NEXT_ACTIONS[kind],
+            "exact_next_action": next_action,
             "reasons": reasons,
             "dirty_source_files": dirty_src,
             "workspace_path": workspace,
             "canonical_repo_root": root,
             "under_branches": under_branches,
             "locked_issue_number": locked_issue_number,
+            "create_issue_bootstrap": False,
         }
     return {
         "proven": True,
@@ -351,6 +377,7 @@ def assess_root_source_mutation(
         "canonical_repo_root": root,
         "under_branches": under_branches,
         "locked_issue_number": locked_issue_number,
+        "create_issue_bootstrap": create_issue_bootstrap,
     }
 
 
@@ -365,6 +392,7 @@ def assess_production_mutation_guards(
     role_kind: str | None = None,
     require_author_lock: bool = False,
     in_test_mode: bool = False,
+    mutation_task: str | None = None,
 ) -> dict[str, Any]:
     """Compose root + scope production guards when they must be active (#683)."""
     if not production_guards_active(in_test_mode=in_test_mode):
@@ -385,6 +413,7 @@ def assess_production_mutation_guards(
         current_branch=current_branch,
         locked_issue_number=locked_issue_number,
         role_kind=role_kind,
+        mutation_task=mutation_task,
     )
     if root_assess["block"]:
         return {**root_assess, "skipped": False}
@@ -408,6 +437,7 @@ def assess_production_mutation_guards(
         "skipped": False,
         "root": root_assess,
         "scope": scope_assess,
+        "create_issue_bootstrap": bool(root_assess.get("create_issue_bootstrap")),
     }