fix(mcp): allow author-lock recovery after the owning session exits (Closes #753)

A durable author issue lock records the PID of the MCP session that took it.
When that process exits, assess_lock_freshness marks the lock stale (live=False)
even while its lease is still within TTL, so every ownership check that needs a
live lock fails closed -- including gitea_update_pr_branch_by_merge.

Re-taking the lock was unreachable for real work. assess_issue_lock_worktree
requires the worktree to 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: it is only
consulted once the lease has expired, so a dead PID under an unexpired lease
never reaches it. assess_own_branch_adoption already speaks of "lock recovery",
but it runs after the base-equivalence gate and so was never reached.

This adds issue_lock_recovery, a pure assessor that grants a narrow waiver only
when every element of durable ownership still matches exactly and the recorded
process is demonstrably dead: same remote/org/repo/issue, same branch (and the
worktree actually on it), same registered worktree, clean worktree, local head
== remote head == open PR head, same claimant identity/profile, no competing
live lock or lease, and no ambiguous branch claims. A malformed or incomplete
lock record can never prove ownership.

The waiver suppresses base-equivalence and nothing else. Cleanliness and every
other precondition still apply, and brand-new issue claims keep the full
requirement. A refused assessment never raises: it withholds the waiver and
lets the pre-existing guard fail closed exactly as before, so recovery can only
ever add permission, never remove a guard. Refusal reasons are appended to the
block message so a caller sees the exact missing evidence.

A completed recovery is recorded on the lock as dead_session_recovery with the
prior and replacement session PIDs, heads, and claimant, so the takeover is
auditable and never looks like an original claim. Rebinding sets the live
session PID, so the recovered lock satisfies verify_lock_for_mutation and the
downstream PR update paths.

Validation:
* new tests/test_issue_753_dead_pid_lock_recovery.py -- 33 passed
* issue-lock, adoption, store, provenance, registration, duplicate-gate,
  worktree, create-issue-guard suites -- 120 passed
* MCP server, commit payloads, handoff ledger, PR ownership, branch cleanup
  suites -- 286 passed
* full suite -- 3498 passed, 6 skipped, 2 failed
* the same 2 failures reproduce identically on pristine master 0425bf9a
  (test_issue_702_review_findings_f1_f6 F1 recovery-before-probe and
  test_reconciler_supersession_close org/repo forwarding), so they are
  pre-existing and unrelated
* git diff --check clean

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

docs/canonical-repository-root.md

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

gitea_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
 
 

gitea_mcp_server.py

@@ -393,6 +393,35 @@ def _configured_canonical_root() -> tuple[str | None, str | None]:
     return crr.configured_canonical_root(profile, os.environ)
 
 
+def _canonical_local_git_root() -> str:
+    """Local git working root for *target-repository* operations (#741).
+
+    Two distinct roots exist and must never be conflated:
+
+    * ``PROJECT_ROOT`` — the Gitea-Tools **installation** checkout, i.e. where
+      this server script lives. Server implementation/version parity, workflow
+      and skill file loading, and self-code staleness detection are anchored
+      here and must stay that way.
+    * the **canonical target repository root** — the working root of the
+      repository whose issues/PRs/branches the namespace actually mutates.
+
+    Every local git invocation that reads or proves *target repository* state
+    (remote identity, ancestry, worktree inventory, cleanup) must run here, or
+    a cross-repository namespace silently operates on Gitea-Tools instead.
+
+    Returns the resolved canonical target root when one is configured, else
+    ``PROJECT_ROOT`` — which preserves the single-repository default exactly.
+    A configured-but-invalid root is *not* silently replaced by the install
+    checkout: it is returned as-is so downstream git calls fail rather than
+    succeed against the wrong repository, and
+    :func:`_enforce_canonical_repository_root` fails the mutation closed first.
+    """
+    configured, _source = _configured_canonical_root()
+    if not configured:
+        return PROJECT_ROOT
+    return crr.resolve_repo_toplevel(configured) or os.path.realpath(configured)
+
+
 def _resolve_preflight_workspace_path(worktree_path: str | None = None) -> str:
     """Resolve the namespace-scoped workspace root inspected by pre-flight guards.
 
@@ -1448,6 +1477,12 @@ def _verify_role_mutation_workspace(
     git_state = issue_lock_worktree.read_worktree_git_state(
         _resolve_preflight_workspace_path(worktree_path)
     )
+    # #741: thread the configured canonical root exactly as
+    # _resolve_namespace_mutation_context does. Omitting it here made the two
+    # paths disagree: the #274 branches-only / worktree-membership guards fell
+    # back to the install checkout and validated Gitea-Tools/branches/ instead
+    # of the target repository the namespace is actually bound to.
+    _configured_root, _configured_source = _configured_canonical_root()
     assessment = nwb.assess_namespace_mutation_workspace(
         role_kind=role,
         worktree_path=worktree_path,
@@ -1458,6 +1493,7 @@ def _verify_role_mutation_workspace(
         ),
         profile_name=get_profile().get("profile_name"),
         current_branch=git_state.get("current_branch"),
+        configured_canonical_root=_configured_root,
     )
     if assessment["block"]:
         raise RuntimeError(
@@ -1614,6 +1650,7 @@ import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
 import issue_lock_store  # noqa: E402
 import issue_lock_adoption  # noqa: E402
+import issue_lock_recovery  # noqa: E402
 import stacked_pr_support  # noqa: E402
 import merge_approval_gate  # noqa: E402
 import review_quarantine  # noqa: E402  # #695 contaminated formal-review quarantine
@@ -2403,9 +2440,24 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
     filled_org = False
     filled_repo = False
     if not org_explicit or not repo_explicit:
-        parsed = remote_repo_guard.parse_org_repo_from_remote_url(
-            _local_git_remote_url(remote)
-        )
+        parsed = None
+        # #741: for a cross-repository namespace, prefer the canonical root's
+        # own repository identity. `_local_git_remote_url` looks a remote up by
+        # *name*, but a target checkout commonly names its remote `origin`
+        # rather than `prgs`; that lookup then returns None and the omitted
+        # coordinates silently fell through to the REMOTES default target —
+        # i.e. a completely unrelated repository. `_canonical_repository_slug`
+        # probes the candidate remote names against the canonical root.
+        try:
+            canonical_slug, _reasons = _canonical_repository_slug(get_profile(), remote)
+        except Exception:
+            canonical_slug = None
+        if canonical_slug:
+            parsed = session_ctx.parse_repository_slug(canonical_slug)
+        if not parsed:
+            parsed = remote_repo_guard.parse_org_repo_from_remote_url(
+                _local_git_remote_url(remote)
+            )
         if parsed:
             if not org_explicit:
                 resolved_org = parsed[0]
@@ -2413,6 +2465,40 @@ def _resolve(remote: str, host: str | None, org: str | None, repo: str | None):
             if not repo_explicit:
                 resolved_repo = parsed[1]
                 filled_repo = True
+    # #741: explicit caller coordinates may *confirm* a cross-repository
+    # canonical binding but must never override it. Both-explicit coordinates
+    # short-circuit the #530 guard (remote_repo_guard.assess_remote_repo_match),
+    # so without this check a caller could name any repository and skip
+    # validation entirely. Unconfigured (single-repository) namespaces yield no
+    # canonical slug and are unaffected. An unresolvable configured root is not
+    # blocked here — reads must keep working; the fail-closed decision belongs
+    # to _enforce_canonical_repository_root at the mutation gate.
+    if org_explicit or repo_explicit:
+        try:
+            canonical_slug, _canonical_reasons = _canonical_repository_slug(
+                get_profile(), remote
+            )
+        except Exception:
+            canonical_slug = None
+        bound = (
+            session_ctx.parse_repository_slug(canonical_slug)
+            if canonical_slug
+            else None
+        )
+        if bound:
+            override = session_ctx.assess_repository_override(
+                requested_org=org,
+                requested_repo=repo,
+                bound_org=bound[0],
+                bound_repo=bound[1],
+            )
+            if override.get("block"):
+                raise RuntimeError(
+                    "Canonical repository binding guard (#741): "
+                    + "; ".join(override.get("reasons") or [])
+                    + ". Explicit org/repo may confirm the configured canonical "
+                    "repository binding but can never override it."
+                )
     _enforce_remote_repo_guard(
         remote,
         resolved_org,
@@ -3067,11 +3153,14 @@ def gitea_lock_issue(
         return blocked
 
     resolved_worktree = issue_lock_worktree.resolve_author_worktree_path(
-        worktree_path, PROJECT_ROOT
+        worktree_path, _canonical_local_git_root()
     )
     h, o, r = _resolve(remote, host, org, repo)
+    existing_issue_lock = _load_existing_issue_lock(
+        remote=remote, org=o, repo=r, issue_number=issue_number
+    )
     active_lease_block = issue_lock_store.assess_same_issue_lease_conflict(
-        _load_existing_issue_lock(remote=remote, org=o, repo=r, issue_number=issue_number),
+        existing_issue_lock,
         issue_number=issue_number,
         branch_name=branch_name,
         worktree_path=resolved_worktree,
@@ -3113,6 +3202,72 @@ def gitea_lock_issue(
         org=org,
         repo=repo,
     )
+    # ── Dead-session lock recovery assessment (#753) ──
+    # When the MCP session that took a lock exits, the lock goes non-live
+    # (stale by dead PID) even inside its lease TTL, and the branch it owns is
+    # ahead of its base by construction — so the base-equivalence gate below
+    # makes normal re-lock unreachable for every PR that already exists.
+    #
+    # This grants a waiver ONLY for that case, proven against the durable lock
+    # record plus live git/Gitea observation. A refused assessment never raises:
+    # it simply withholds the waiver, leaving the pre-existing guard to fail
+    # closed exactly as before. Recovery can only ever add permission.
+    recovery_assessment: dict | None = None
+    if (
+        existing_issue_lock
+        and existing_issue_lock.get("issue_number") == issue_number
+        and not issue_lock_store.is_lease_live(existing_issue_lock)
+    ):
+        recovery_auth = _auth(h)
+        try:
+            recovery_branches = api_get_all(
+                f"{repo_api_url(h, o, r)}/branches", recovery_auth
+            )
+        except Exception as exc:
+            raise RuntimeError(
+                f"Could not list branches to verify issue-lock recovery: {exc}"
+            )
+        recovery_remote_head: str | None = None
+        recovery_candidates: list[str] = []
+        for entry in recovery_branches:
+            entry_name = _branch_entry_name(entry)
+            if entry_name == branch_name:
+                recovery_remote_head = _branch_entry_commit_sha(entry)
+            if issue_lock_adoption.branch_carries_issue_marker(entry_name, issue_number):
+                recovery_candidates.append(entry_name)
+        recovery_pr_head: str | None = None
+        recovery_pr_number: int | None = None
+        for pull in _list_open_pulls(h, o, r, recovery_auth):
+            pull_head = pull.get("head") or {}
+            if str(pull_head.get("ref") or "") == branch_name:
+                recovery_pr_head = pull_head.get("sha")
+                recovery_pr_number = pull.get("number")
+                break
+        recovery_claimant = _work_lease_claimant(h)
+        recovery_assessment = issue_lock_recovery.assess_dead_session_lock_recovery(
+            existing_issue_lock,
+            issue_number=issue_number,
+            branch_name=branch_name,
+            worktree_path=resolved_worktree,
+            remote=remote,
+            org=o,
+            repo=r,
+            identity=recovery_claimant.get("username"),
+            profile=recovery_claimant.get("profile"),
+            current_branch=git_state.get("current_branch"),
+            porcelain_status=git_state.get("porcelain_status") or "",
+            head_sha=git_state.get("head_sha"),
+            remote_head_sha=recovery_remote_head,
+            pr_head_sha=recovery_pr_head,
+            pr_number=recovery_pr_number,
+            competing_live_locks=issue_lock_store.list_live_locks(),
+            candidate_branches=recovery_candidates,
+            current_pid=os.getpid(),
+        )
+
+    recovery_sanctioned = bool(
+        recovery_assessment and recovery_assessment.get("recovery_sanctioned")
+    )
     lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
         worktree_path=resolved_worktree,
         current_branch=git_state.get("current_branch"),
@@ -3120,10 +3275,20 @@ def gitea_lock_issue(
         base_equivalent=git_state.get("base_equivalent"),
         inspected_git_root=git_state.get("inspected_git_root"),
         base_branch=git_state.get("base_branch"),
+        recovery_sanctioned=recovery_sanctioned,
     )
     if lock_assessment["block"]:
+        reasons = list(lock_assessment.get("reasons") or [])
+        # Surface why recovery was unavailable, so a blocked caller sees the
+        # exact missing evidence instead of only the base-equivalence text.
+        if recovery_assessment and recovery_assessment.get("is_candidate"):
+            reasons.append(
+                issue_lock_recovery.format_recovery_refusal(recovery_assessment)
+            )
         raise RuntimeError(
-            issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
+            issue_lock_worktree.format_issue_lock_worktree_error(
+                {**lock_assessment, "reasons": reasons}
+            )
         )
 
     auth = _auth(h)
@@ -3186,6 +3351,14 @@ def gitea_lock_issue(
     }
     if stacked_approved:
         data["approved_stacked_base"] = stacked_approved
+    if recovery_sanctioned and recovery_assessment:
+        # #753 AC2/AC6: record that this claim was recovered after session
+        # death, with the prior and replacement session identity, so the
+        # takeover is auditable and never looks like an original claim.
+        data["dead_session_recovery"] = issue_lock_recovery.build_recovery_record(
+            recovery_assessment,
+            recovered_at=_work_lease_timestamp(_work_lease_now()),
+        )
 
     lock_file_path = _save_issue_lock(data)
     lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
@@ -3219,6 +3392,14 @@ def gitea_lock_issue(
         "lock_freshness": freshness,
         "lock_proof": lock_proof,
     }
+    if recovery_sanctioned and recovery_assessment:
+        result["dead_session_recovery"] = data["dead_session_recovery"]
+        result["message"] = (
+            f"Recovered the durable lock for issue #{issue_number} on branch "
+            f"'{branch_name}' after the owning MCP session (pid "
+            f"{recovery_assessment['evidence'].get('prior_session_pid')}) exited; "
+            "ownership evidence matched exactly (fail-closed check complete)."
+        )
     if stacked_approved:
         result["approved_stacked_base"] = stacked_approved
         result["message"] = (
@@ -3369,7 +3550,7 @@ def gitea_create_pr(
     locked_worktree = lock_data.get("worktree_path")
 
     worktree_check = issue_lock_worktree.verify_pr_worktree_matches_lock(
-        locked_worktree, worktree_path, PROJECT_ROOT
+        locked_worktree, worktree_path, _canonical_local_git_root()
     )
     if worktree_check["block"]:
         raise ValueError(worktree_check["reasons"][0])
@@ -8499,13 +8680,24 @@ def gitea_merge_pr(
 
 
 def _local_git_remote_url(remote_name: str) -> str | None:
-    """Best-effort local ``git remote get-url`` for trust-gate corroboration."""
+    """Best-effort local ``git remote get-url`` for trust-gate corroboration.
+
+    #741: this runs in the **canonical target repository root**, not the
+    Gitea-Tools installation checkout. Every consumer of this function derives
+    *target repository identity* from it (``_resolve``, the #530 remote/repo
+    guard, the anti-stomp org/repo fill, ``_workspace_repository_slug``), so
+    running it in ``PROJECT_ROOT`` inverted those guards for a cross-repository
+    namespace: the genuinely bound target was rejected while Gitea-Tools was
+    accepted. Single-repository namespaces are unaffected —
+    :func:`_canonical_local_git_root` returns ``PROJECT_ROOT`` when no
+    cross-repository binding is configured.
+    """
     try:
         proc = subprocess.run(
             ["git", "remote", "get-url", remote_name],
             capture_output=True,
             text=True,
-            cwd=PROJECT_ROOT,
+            cwd=_canonical_local_git_root(),
         )
         if proc.returncode != 0:
             return None
@@ -9075,7 +9267,7 @@ def gitea_cleanup_merged_pr_branch(
         else f"origin/{target_branch}"
     )
     head_on_target = merged_cleanup_reconcile.is_head_ancestor_of_ref(
-        PROJECT_ROOT,
+        _canonical_local_git_root(),
         head_sha,
         target_ref,
     )
@@ -9110,7 +9302,7 @@ def gitea_cleanup_merged_pr_branch(
         repo=r,
         branch=head_branch,
         pr_number=pr_number,
-        project_root=PROJECT_ROOT,
+        project_root=_canonical_local_git_root(),
         auth=auth,
         base_api=base,
     )
@@ -9605,7 +9797,7 @@ def gitea_capture_branches_worktree_snapshot(
             "reasons": read_block,
             "permission_report": _permission_block_report("gitea.read"),
         }
-    root = PROJECT_ROOT
+    root = _canonical_local_git_root()
     if worktree_path:
         root = os.path.realpath(os.path.abspath(worktree_path))
         git_root = _get_git_root(root)
@@ -9715,14 +9907,14 @@ def gitea_reconcile_merged_cleanups(
                 h, o, r, auth, head_ref
             )
         head_on_master[int(pr["number"])] = merged_cleanup_reconcile.is_head_ancestor_of_ref(
-            PROJECT_ROOT, head_sha, master_ref
+            _canonical_local_git_root(), head_sha, master_ref
         )
 
     delete_capability_allowed = not _profile_operation_gate("gitea.branch.delete")
 
     # #534: discover reviewer scratch trees and active leases for those PRs.
     scratch_candidates = merged_cleanup_reconcile.discover_reviewer_scratch_worktrees(
-        PROJECT_ROOT
+        _canonical_local_git_root()
     )
     active_reviewer_leases: dict[int, bool] = {}
     pr_states: dict[int, dict] = {}
@@ -9748,7 +9940,7 @@ def gitea_reconcile_merged_cleanups(
         }
 
     report = merged_cleanup_reconcile.build_reconciliation_report(
-        project_root=PROJECT_ROOT,
+        project_root=_canonical_local_git_root(),
         closed_prs=merged_closed,
         open_prs=open_prs,
         remote_branch_exists=remote_branch_exists,
@@ -9788,7 +9980,7 @@ def gitea_reconcile_merged_cleanups(
                 repo=r,
                 branch=head_branch,
                 pr_number=pr_num_int,
-                project_root=PROJECT_ROOT,
+                project_root=_canonical_local_git_root(),
                 auth=auth,
                 base_api=base,
             )
@@ -9872,7 +10064,7 @@ def gitea_reconcile_merged_cleanups(
 
         if local_assessment.get("safe_to_remove_worktree"):
             result = merged_cleanup_reconcile.remove_local_worktree(
-                PROJECT_ROOT,
+                _canonical_local_git_root(),
                 head_branch,
                 worktree_path=local_assessment.get("worktree_path"),
             )
@@ -9882,7 +10074,7 @@ def gitea_reconcile_merged_cleanups(
         if not scratch.get("safe_to_remove_worktree"):
             continue
         result = merged_cleanup_reconcile.remove_reviewer_scratch_worktree(
-            PROJECT_ROOT, scratch.get("worktree_path") or ""
+            _canonical_local_git_root(), scratch.get("worktree_path") or ""
         )
         actions.append({"action": "remove_reviewer_scratch_worktree", **result})
 
@@ -9980,7 +10172,7 @@ def gitea_assess_already_landed_reconciliation(
 
     assessment = reconciliation_workflow.assess_open_pr_reconciliation(
         pr=pr,
-        project_root=PROJECT_ROOT,
+        project_root=_canonical_local_git_root(),
         remote=remote,
         target_branch=target_branch,
     )
@@ -10076,14 +10268,14 @@ def gitea_scan_already_landed_open_prs(
     }
 
     target_fetch = reconciliation_workflow.fetch_target_branch(
-        PROJECT_ROOT, remote, target_branch
+        _canonical_local_git_root(), remote, target_branch
     )
 
     candidates: list[dict] = []
     for pr in open_prs:
         assessment = reconciliation_workflow.assess_open_pr_reconciliation(
             pr=pr,
-            project_root=PROJECT_ROOT,
+            project_root=_canonical_local_git_root(),
             remote=remote,
             target_branch=target_branch,
             target_fetch=target_fetch,
@@ -10180,7 +10372,7 @@ def gitea_audit_worktree_cleanup(
         active_issue_branches.add(str(lock["branch_name"]).strip())
 
     report = worktree_cleanup_audit.audit_branches_directory(
-        PROJECT_ROOT,
+        _canonical_local_git_root(),
         open_pr_branches=open_pr_branches,
         active_issue_branches=active_issue_branches,
         now=datetime.now(timezone.utc),
@@ -10263,7 +10455,7 @@ def gitea_reconcile_already_landed_pr(
 
     assessment = already_landed_reconcile.assess_already_landed_reconciliation(
         pr=pr,
-        project_root=PROJECT_ROOT,
+        project_root=_canonical_local_git_root(),
         remote=remote,
         target_branch=target_branch,
     )
@@ -10468,7 +10660,7 @@ def gitea_reconcile_superseded_by_merged_pr(
     )
 
     target_fetch = already_landed_reconcile.fetch_target_branch(
-        PROJECT_ROOT, remote, target_branch
+        _canonical_local_git_root(), remote, target_branch
     )
     superseding_head_sha = (
         (superseding_pr.get("head") or {}).get("sha")
@@ -10478,7 +10670,7 @@ def gitea_reconcile_superseded_by_merged_pr(
     ancestor = None
     if target_fetch.get("success"):
         ancestor = already_landed_reconcile.is_head_ancestor_of_ref(
-            PROJECT_ROOT,
+            _canonical_local_git_root(),
             superseding_head_sha,
             target_fetch.get("target_ref") or f"{remote}/{target_branch}",
         )
@@ -11622,6 +11814,11 @@ def gitea_acquire_merger_pr_lease(
     }, lease_provenance=merger_lease_adoption.build_lease_provenance(
         source=merger_lease_adoption.SOURCE_ACQUIRE_MERGER,
         comment_id=posted.get("id"),
+        # #742: bind the lease to this native runtime so a later owner-session
+        # finalization can prove it is the same daemon that acquired it.
+        native_token_fingerprint=(
+            mcp_daemon_guard.native_runtime_status().get("token_fingerprint")
+        ),
     ))
     return {
         "success": True,
@@ -11811,6 +12008,191 @@ def gitea_adopt_merger_pr_lease(
     }
 
 
+@mcp.tool()
+def gitea_release_merger_pr_lease(
+    pr_number: int,
+    worktree: str,
+    candidate_head: str,
+    outcome: str = merger_lease_adoption.OUTCOME_RELEASED,
+    reason: str | None = None,
+    issue_number: int | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Terminally release or abandon this merger session's own PR lease (#742).
+
+    Sanctioned owner-session finalization for a comment-backed merger lease when
+    the merge does **not** occur (non-retryable merge failure, aborted merge, or
+    a blocker that ends the attempt). Without it a failed merger lease can only
+    expire, blocking the queue for the remainder of its TTL.
+
+    Merger-only and owner-only. The caller must hold the exact in-session lease
+    it is finalizing; foreign-session release, reviewer profiles, and mismatched
+    repository / PR / candidate head / identity / profile / native runtime token
+    fingerprint all fail closed. Reviewer sessions use
+    ``gitea_release_reviewer_pr_lease`` instead — that tool is never a merger
+    path.
+
+    Finalization is append-only: it posts a terminal lease marker and never
+    edits or deletes prior ledger comments. It is idempotent — an already
+    terminal lease returns ``already_terminal`` and posts nothing. The PR, its
+    approval, the review decision lock, the branch, and the worktree are all
+    preserved; only the lease is ended.
+
+    Args:
+        pr_number: PR whose merger lease to finalize.
+        outcome: 'released' (clean end) or 'abandoned' (ended under a blocker).
+        reason: Short finalization reason recorded in the marker.
+        candidate_head: Pinned head SHA the lease was acquired against.
+
+    Returns:
+        dict reporting finalization status, reasons, and the marker comment id.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    comment_block = _profile_operation_gate("gitea.pr.comment")
+    if comment_block:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": comment_block,
+            "permission_report": _permission_block_report("gitea.pr.comment"),
+        }
+    # Merger-role proof: reviewer profiles hold gitea.pr.comment but never
+    # gitea.pr.merge, so this gate keeps the tool merger-only (#742).
+    merge_block = _profile_operation_gate("gitea.pr.merge")
+    if merge_block:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": merge_block,
+            "permission_report": _permission_block_report("gitea.pr.merge"),
+        }
+
+    try:
+        _verify_role_mutation_workspace(
+            remote,
+            worktree=worktree,
+            task="release_merger_pr_lease",
+            org=org,
+            repo=repo,
+        )
+    except RuntimeError as e:
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": [str(e)],
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    profile = get_profile()
+    profile_name = profile.get("profile_name") or "unknown"
+    identity = _authenticated_username(h) or profile.get("username") or ""
+    repo_label = f"{o}/{r}"
+
+    session = reviewer_pr_lease.get_session_lease()
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo)
+
+    pr_live = api_request(
+        "GET", f"{repo_api_url(h, o, r)}/pulls/{pr_number}", auth) or {}
+    live_head = str(
+        (pr_live.get("head") or {}).get("sha") or pr_live.get("head_sha") or ""
+    ).strip()
+
+    assessment = merger_lease_adoption.assess_merger_lease_finalization(
+        comments,
+        pr_number=pr_number,
+        session=session,
+        actor_identity=identity,
+        actor_profile=profile_name,
+        actor_session_id=(session or {}).get("session_id"),
+        repo=repo_label,
+        worktree=worktree,
+        candidate_head=candidate_head,
+        live_head_sha=live_head or None,
+        outcome=outcome,
+        reason=reason,
+        runtime_token_fingerprint=(
+            mcp_daemon_guard.native_runtime_status().get("token_fingerprint")
+        ),
+        issue_number=issue_number,
+    )
+
+    if assessment.get("already_terminal"):
+        # Idempotent: the lease is already terminal on the ledger. Clear the
+        # in-session mirror so no stale proof survives, and post nothing.
+        reviewer_pr_lease.clear_session_lease()
+        return {
+            "success": True,
+            "finalized": False,
+            "already_terminal": True,
+            "pr_number": pr_number,
+            "outcome": assessment.get("outcome"),
+            "reasons": [],
+        }
+
+    if not assessment.get("finalize_allowed"):
+        return {
+            "success": False,
+            "finalized": False,
+            "already_terminal": False,
+            "pr_number": pr_number,
+            "reasons": assessment.get("reasons") or [],
+        }
+
+    comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
+    with _audited(
+        "comment_pr",
+        host=h,
+        remote=remote,
+        org=o,
+        repo=r,
+        pr_number=pr_number,
+        request_metadata={"source": "release_merger_pr_lease"},
+    ):
+        posted = api_request(
+            "POST", comment_url, auth, {"body": assessment["finalization_body"]}
+        )
+
+    if not isinstance(posted, dict) or not posted.get("id"):
+        # The lease is still live on the ledger: keep the in-session mirror so
+        # the owner can retry finalization rather than losing its own proof.
+        return {
+            "success": False,
+            "finalized": False,
+            "reasons": [
+                "terminal lease marker POST failed or returned no comment id "
+                "(lease left intact; retry finalization)"
+            ],
+        }
+
+    reviewer_pr_lease.clear_session_lease()
+
+    return {
+        "success": True,
+        "finalized": True,
+        "already_terminal": False,
+        "pr_number": pr_number,
+        "outcome": assessment.get("outcome"),
+        "finalization_reason": assessment.get("finalization_reason"),
+        "finalization_comment_id": posted.get("id"),
+        "finalized_lease_comment_id": assessment.get("lease_comment_id"),
+        "candidate_head": assessment.get("candidate_head"),
+        "repo": repo_label,
+        "reasons": [],
+    }
+
+
 @mcp.tool()
 def gitea_heartbeat_reviewer_pr_lease(
     pr_number: int,
@@ -11846,6 +12228,13 @@ def gitea_heartbeat_reviewer_pr_lease(
     verify_preflight_purity(remote, task="review_pr")
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
+    # Slide the sliding TTL forward (#747): the heartbeat is the liveness proof,
+    # so renewal is stated explicitly rather than inherited from the acquisition
+    # default.
+    beat_at = datetime.now(timezone.utc)
+    renewed_expiry = beat_at + timedelta(
+        minutes=reviewer_pr_lease.LEASE_RENEWAL_MINUTES
+    )
     body = reviewer_pr_lease.format_lease_body(
         repo=f"{o}/{r}",
         pr_number=pr_number,
@@ -11858,6 +12247,8 @@ def gitea_heartbeat_reviewer_pr_lease(
         candidate_head=candidate_head or session.get("candidate_head"),
         target_branch=session.get("target_branch") or "master",
         target_branch_sha=target_branch_sha or session.get("target_branch_sha"),
+        last_activity=beat_at,
+        ttl_minutes=reviewer_pr_lease.LEASE_RENEWAL_MINUTES,
     )
     comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
     with _audited(
@@ -11898,6 +12289,13 @@ def gitea_heartbeat_reviewer_pr_lease(
         "phase": phase,
         "comment_id": posted.get("id"),
         "session_lease": updated,
+        # Report the renewed window so an operator can tell "held and live"
+        # from "held and dying" (#747).
+        "ttl_minutes": reviewer_pr_lease.LEASE_RENEWAL_MINUTES,
+        "expires_at": renewed_expiry.replace(microsecond=0)
+        .isoformat()
+        .replace("+00:00", "Z"),
+        "seconds_remaining": reviewer_pr_lease.LEASE_RENEWAL_MINUTES * 60,
         "reasons": [],
     }
 
@@ -15065,46 +15463,151 @@ def _count_commits_behind(
     return None
 
 
+def _matching_protection_rule(protections: Any, branch: str) -> dict | None:
+    """Select the branch-protection rule governing ``branch`` (#751)."""
+    if not isinstance(protections, list):
+        return None
+    for rule in protections:
+        if not isinstance(rule, dict):
+            continue
+        name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+        # Exact match or glob-ish contains for common patterns.
+        if name == branch or name in ("*", f"{branch}"):
+            return rule
+    # Fallback: any protection that mentions the base branch.
+    for rule in protections:
+        if not isinstance(rule, dict):
+            continue
+        name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
+        if name and (branch in name or name.endswith(branch)):
+            return rule
+    return None
+
+
+def _branch_protection_policy(
+    base_url: str,
+    auth: dict,
+    *,
+    base_branch: str,
+) -> dict:
+    """Read the live branch-protection policy for ``base_branch`` (#751).
+
+    Exposes both the current-base rule (``block_on_outdated_branch``) and the
+    status-check requirement (``enable_status_check`` / ``status_check_contexts``)
+    from the same payload, so the checks assessor can tell "no checks required"
+    apart from "required checks pending".
+
+    ``determinable`` is False only when the policy genuinely could not be read
+    (missing branch or API failure) — never merely because no rule exists. A
+    successful read that finds no protection for the branch is authoritative
+    evidence that status checks are not required.
+    """
+    policy: dict[str, Any] = {
+        "determinable": False,
+        "protection_found": False,
+        "requires_current_base": None,
+        "checks_enabled": None,
+        "required_contexts": [],
+        "base_branch": (base_branch or "").strip() or None,
+    }
+    branch = (base_branch or "").strip()
+    if not branch:
+        return policy
+
+    def _apply(rule: dict) -> None:
+        policy["protection_found"] = True
+        if "block_on_outdated_branch" in rule:
+            policy["requires_current_base"] = bool(
+                rule.get("block_on_outdated_branch")
+            )
+        if "enable_status_check" in rule:
+            policy["checks_enabled"] = bool(rule.get("enable_status_check"))
+        contexts = rule.get("status_check_contexts")
+        if isinstance(contexts, list):
+            policy["required_contexts"] = [
+                str(ctx).strip() for ctx in contexts if str(ctx or "").strip()
+            ]
+
+    try:
+        protections = api_request(
+            "GET", f"{base_url}/branch_protections", auth
+        )
+        rule = _matching_protection_rule(protections, branch)
+        if rule is not None:
+            _apply(rule)
+        if not policy["protection_found"]:
+            # Branch payload may embed effective protection.
+            br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
+            prot = br.get("protection") or br.get("effective_branch_protection") or {}
+            if isinstance(prot, dict) and prot:
+                _apply(prot)
+        policy["determinable"] = True
+    except Exception:
+        # Genuine read failure — leave determinable False so callers fail closed.
+        return policy
+
+    if not policy["protection_found"]:
+        # Authoritative absence: no protection governs the branch, so no status
+        # check is required by policy.
+        policy["checks_enabled"] = False
+    elif policy["checks_enabled"] is None:
+        # Protection exists but omits the status-check field entirely: Gitea
+        # only enforces contexts when the flag is set, so absence means off.
+        policy["checks_enabled"] = False
+
+    return policy
+
+
 def _branch_protection_requires_current_base(
     base_url: str,
     auth: dict,
     *,
     base_branch: str,
 ) -> bool | None:
-    """Read Gitea branch protection ``block_on_outdated_branch`` when present."""
-    branch = (base_branch or "").strip()
-    if not branch:
-        return None
+    """Read Gitea branch protection ``block_on_outdated_branch`` when present.
+
+    Thin accessor over :func:`_branch_protection_policy`; return semantics are
+    unchanged (``None`` when the rule is absent or unreadable).
+    """
+    policy = _branch_protection_policy(base_url, auth, base_branch=base_branch)
+    return policy.get("requires_current_base")
+
+
+def _commit_checks_snapshot(
+    base_url: str,
+    auth: dict,
+    *,
+    sha: str,
+) -> dict:
+    """Read the combined commit status *and its context collection* (#751).
+
+    The combined ``state`` alone is not decisive: Gitea reports ``pending`` for
+    a commit with an empty status-context collection, which is indistinguishable
+    from executing CI unless the collection itself is inspected.
+    """
+    snapshot: dict[str, Any] = {
+        "determinable": False,
+        "combined_state": None,
+        "statuses": [],
+    }
+    head = (sha or "").strip()
+    if not head:
+        return snapshot
     try:
-        # Prefer the named protection rule matching the base branch.
-        protections = api_request(
-            "GET", f"{base_url}/branch_protections", auth
-        ) or []
-        if isinstance(protections, list):
-            for rule in protections:
-                if not isinstance(rule, dict):
-                    continue
-                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
-                # Exact match or glob-ish contains for common patterns.
-                if name == branch or name in ("*", f"{branch}"):
-                    if "block_on_outdated_branch" in rule:
-                        return bool(rule.get("block_on_outdated_branch"))
-            # Fallback: any protection that mentions the base branch.
-            for rule in protections:
-                if not isinstance(rule, dict):
-                    continue
-                name = (rule.get("branch_name") or rule.get("rule_name") or "").strip()
-                if branch in name or name.endswith(branch):
-                    if "block_on_outdated_branch" in rule:
-                        return bool(rule.get("block_on_outdated_branch"))
-        # Branch payload may embed effective protection.
-        br = api_request("GET", f"{base_url}/branches/{branch}", auth) or {}
-        prot = br.get("protection") or br.get("effective_branch_protection") or {}
-        if isinstance(prot, dict) and "block_on_outdated_branch" in prot:
-            return bool(prot.get("block_on_outdated_branch"))
+        payload = api_request(
+            "GET", f"{base_url}/commits/{head}/status", auth
+        )
     except Exception:
-        return None
-    return None
+        return snapshot
+    if payload is None:
+        return snapshot
+    snapshot["determinable"] = True
+    if not isinstance(payload, dict):
+        return snapshot
+    snapshot["combined_state"] = (payload.get("state") or "").strip().lower() or None
+    statuses = payload.get("statuses")
+    snapshot["statuses"] = statuses if isinstance(statuses, list) else []
+    return snapshot
 
 
 def _prove_author_ownership_for_pr(
@@ -15349,11 +15852,14 @@ def gitea_assess_pr_sync_status(
             # Fail closed on unknown behind-count only when SHAs differ.
             commits_behind = 0 if pr_head_sha == base_head_sha else None
 
+    # Single live read of the base-branch protection policy; it carries both
+    # the current-base rule and the status-check requirement (#751).
+    protection_policy = _branch_protection_policy(
+        base, auth, base_branch=base_branch
+    )
     if branch_protection_requires_current_base is None:
-        branch_protection_requires_current_base = (
-            _branch_protection_requires_current_base(
-                base, auth, base_branch=base_branch
-            )
+        branch_protection_requires_current_base = protection_policy.get(
+            "requires_current_base"
         )
     # When protection cannot be read, fail closed by requiring current base
     # whenever the PR is behind (safer for protected repos). Callers may pass
@@ -15420,23 +15926,30 @@ def gitea_assess_pr_sync_status(
     except Exception:
         pass
 
+    # ── Live checks derivation (#751) ────────────────────────────────────
+    # Classify from the actual status-context collection plus the live
+    # protection policy. The combined ``state`` is never treated as proof that
+    # CI is executing, because Gitea reports ``pending`` for an empty
+    # collection. ``checks_required`` is always derived from live evidence and
+    # is deliberately not a caller-supplied input, so no session can declare
+    # checks optional without proof.
+    checks_snapshot = _commit_checks_snapshot(base, auth, sha=pr_head_sha or "")
+    checks_classification = pr_sync_status.classify_commit_checks(
+        combined_state=checks_snapshot.get("combined_state"),
+        statuses=checks_snapshot.get("statuses"),
+        checks_enabled=protection_policy.get("checks_enabled"),
+        required_contexts=protection_policy.get("required_contexts"),
+        policy_determinable=bool(protection_policy.get("determinable")),
+        status_determinable=bool(checks_snapshot.get("determinable")),
+    )
+    checks_required = bool(checks_classification.get("checks_required", True))
+    caller_supplied_checks_status = checks_status
     if checks_status is None:
-        checks_status = "unknown"
-        # Combined status on PR head when Actions/status API is available.
-        if pr_head_sha:
-            try:
-                st = api_request(
-                    "GET",
-                    f"{base}/commits/{pr_head_sha}/status",
-                    auth,
-                ) or {}
-                state = (st.get("state") or "").strip().lower()
-                if state:
-                    checks_status = state
-                elif not st:
-                    checks_status = "none"
-            except Exception:
-                checks_status = "unknown"
+        checks_status = checks_classification.get("checks_status")
+    elif checks_classification.get("checks_status") != pr_sync_status.CHECKS_UNKNOWN:
+        # Live evidence outranks a caller-supplied value; the override only
+        # applies when live status could not be classified at all.
+        checks_status = checks_classification.get("checks_status")
 
     assessment = pr_sync_status.assess_pr_sync_status(
         host=h,
@@ -15457,9 +15970,26 @@ def gitea_assess_pr_sync_status(
         active_reviewer_lease=active_reviewer_lease,
         active_merger_lease=active_merger_lease,
         prepared_verdict_head_sha=prepared_verdict_head_sha,
+        checks_required=checks_required,
     )
     assessment["remote"] = remote if remote in REMOTES else None
     assessment["base_branch"] = base_branch
+    # Evidence for the checks decision (#751) — no secrets, read-only.
+    assessment["checks_evidence"] = {
+        "combined_state": checks_classification.get("combined_state"),
+        "context_count": checks_classification.get("context_count"),
+        "observed_contexts": checks_classification.get("observed_contexts"),
+        "required_contexts": checks_classification.get("required_contexts"),
+        "missing_required_contexts": checks_classification.get(
+            "missing_required_contexts"
+        ),
+        "policy_determinable": checks_classification.get("policy_determinable"),
+        "status_determinable": checks_classification.get("status_determinable"),
+        "protection_found": protection_policy.get("protection_found"),
+        "checks_enabled": protection_policy.get("checks_enabled"),
+        "caller_supplied_checks_status": caller_supplied_checks_status,
+        "reasons": checks_classification.get("reasons"),
+    }
     assessment["success"] = True
     assessment["performed"] = False
     return assessment
@@ -15545,7 +16075,7 @@ def gitea_update_pr_branch_by_merge(
 
     control_clean = True
     try:
-        porcelain = _get_workspace_porcelain(PROJECT_ROOT)
+        porcelain = _get_workspace_porcelain(_canonical_local_git_root())
         # Untracked-only dirt is ignored; tracked edits contaminate control.
         tracked_dirty = [
             line for line in (porcelain or "").splitlines()
@@ -16550,6 +17080,8 @@ def gitea_resolve_task_capability(
         "gitea_acquire_merger_pr_lease",
         "adopt_merger_pr_lease",
         "gitea_adopt_merger_pr_lease",
+        "release_merger_pr_lease",
+        "gitea_release_merger_pr_lease",
         "create_branch",
         "push_branch",
         "create_pr",

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,400 @@
+"""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 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),
     }
 
 

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],
     *,

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 ────────────────────────────────────────
-    if checks_required and checks not in ("success", "passed", "ok", "none", "skipped", "not_required"):
-        if checks in ("pending", "running", "queued"):
+    # ── Checks gate for merge_now (#751) ─────────────────────────────────
+    # ``checks_required`` is derived from the live branch-protection policy by
+    # 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
-            return result
-        if checks in ("failure", "failed", "error", "cancelled"):
+        elif checks in _CTX_FAILURE:
             reasons.append(f"required checks failed (status={checks})")
-            result["recommended_next_action"] = ACTION_BLOCKED
-            return result
-        # unknown — fail closed when checks_required
-        if checks == "unknown":
+        elif checks == CHECKS_MISSING_REQUIRED:
+            reasons.append(
+                "branch protection configures required status context(s) but no "
+                "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
-            return result
+        else:
+            # 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
-STALE_WARNING_MINUTES = 30
-RECLAIMABLE_MINUTES = 60
+# Reviewer and merger PR leases use a short *sliding* window (#747): a lease
+# expires 10 minutes after its last heartbeat, and every heartbeat slides the
+# 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"

task_capability_map.py

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

tests/test_issue_741_canonical_root_consumers.py

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

tests/test_issue_747_sliding_lease_ttl.py

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

tests/test_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_merger_lease_finalization.py

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

tests/test_reviewer_pr_lease.py

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