Merge pull request 'fix(mcp): allow author-lock recovery after the owning session exits (Closes #753)' (#754) from fix/issue-753-dead-pid-author-lock-recovery into master

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/gitea-execution-profiles.md

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

gitea_mcp_server.py

@@ -1650,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
@@ -1829,7 +1830,6 @@ def _seed_session_context(
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
 import reviewer_pr_lease  # noqa: E402
-import post_merge_moot_lease_gate  # noqa: E402  # #745 reconciler cleanup gate
 import merger_lease_adoption  # noqa: E402
 import merged_cleanup_reconcile  # noqa: E402
 import worktree_cleanup_audit  # noqa: E402
@@ -3156,8 +3156,11 @@ def gitea_lock_issue(
         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,
@@ -3199,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"),
@@ -3206,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)
@@ -3272,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
@@ -3305,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"] = (
@@ -8860,14 +8955,13 @@ def gitea_review_pr(
         return out
 
 
-def _repository_binding_block(
+def _delete_branch_repository_binding_block(
     remote: str | None,
     *,
     org: str | None,
     repo: str | None,
-    required_permission: str = "gitea.branch.delete",
 ) -> dict | None:
-    """#733: validate an explicit mutation target against the workspace binding.
+    """#733: validate an explicit delete target against the workspace binding.
 
     The trusted repository identity comes only from the verified,
     workspace-aligned git remote (never ``REMOTES`` defaults, never
@@ -8900,7 +8994,7 @@ def _repository_binding_block(
         return {
             "success": False,
             "performed": False,
-            "required_permission": required_permission,
+            "required_permission": "gitea.branch.delete",
             "reasons": canonical_reasons,
             "blocker_kind": "repository_binding",
         }
@@ -8919,9 +9013,9 @@ def _repository_binding_block(
             return {
                 "success": False,
                 "performed": False,
-                "required_permission": required_permission,
+                "required_permission": "gitea.branch.delete",
                 "reasons": list(override.get("reasons") or [
-                    "target repository does not match the workspace "
+                    "delete target repository does not match the workspace "
                     "binding (fail closed)"
                 ]),
                 "blocker_kind": "repository_binding",
@@ -8931,51 +9025,17 @@ def _repository_binding_block(
         return {
             "success": False,
             "performed": False,
-            "required_permission": required_permission,
+            "required_permission": "gitea.branch.delete",
             "reasons": [
                 "repository binding unverified: no workspace repository "
                 "identity could be established to corroborate the explicit "
-                "target (fail closed)"
+                "delete target (fail closed)"
             ],
             "blocker_kind": "repository_binding",
         }
     return None
 
 
-def _delete_branch_repository_binding_block(
-    remote: str | None,
-    *,
-    org: str | None,
-    repo: str | None,
-) -> dict | None:
-    """Delete-branch view of :func:`_repository_binding_block` (#733).
-
-    Retained as the named entry point for the delete path so #733/#739
-    regression coverage keeps exercising the exact permission label that
-    ``gitea_delete_branch`` reports.
-    """
-    return _repository_binding_block(
-        remote, org=org, repo=repo,
-        required_permission="gitea.branch.delete",
-    )
-
-
-def _bound_repository_slug(remote: str | None) -> str | None:
-    """Canonical repository slug for the active session, or None (#745).
-
-    Same trust order as ``_repository_binding_block``: the configured canonical
-    root when the namespace declares one, otherwise the workspace-derived
-    identity. Request parameters are never consulted, so they cannot redirect a
-    mutation at a foreign repository.
-    """
-    canonical_slug, canonical_reasons = _canonical_repository_slug(
-        get_profile(), remote
-    )
-    if canonical_reasons:
-        return None
-    return canonical_slug or _workspace_repository_slug(remote)
-
-
 @mcp.tool()
 def gitea_delete_branch(
     branch: str,
@@ -12404,10 +12464,6 @@ def gitea_diagnose_reviewer_pr_lease_handoff(
 def gitea_cleanup_post_merge_moot_lease(
     pr_number: int,
     apply: bool = False,
-    expected_session_id: str | None = None,
-    expected_candidate_head: str | None = None,
-    expected_lease_comment_id: int | None = None,
-    worktree_path: str | None = None,
     remote: str = "dadeschools",
     host: str | None = None,
     org: str | None = None,
@@ -12423,37 +12479,14 @@ def gitea_cleanup_post_merge_moot_lease(
     an append-only comment that neutralises the moot lease without deleting any
     other session's comment.
 
-    Role binding (#745). The two modes are gated differently, on purpose:
-
-    * ``apply=false`` (assessment) stays reachable under ``gitea.read`` for
-      **any** role, matching ``gitea_cleanup_stale_review_decision_lock`` and
-      ``gitea_cleanup_obsolete_reviewer_comment_lease``, so an operator can
-      diagnose a stuck lease from whichever namespace is attached. It performs
-      no mutation and records append-only dry-run evidence in-session.
-    * ``apply=true`` (mutation) additionally requires, in order: the session to
-      have resolved exactly ``cleanup_post_merge_moot_lease``; the
-      ``reconciler`` role; ``gitea.pr.comment``; a validated repository /
-      canonical-root / workspace binding; and matching dry-run evidence proving
-      ``lease_moot`` and ``cleanup_allowed`` for the same PR, lease session,
-      candidate head and lease marker. Author, reviewer and merger fail closed
-      here even though their profiles carry ``gitea.pr.comment``.
-
     Args:
         pr_number: The PR whose lingering lease to assess/clean.
         apply: When false (default) report only (read-only). When true, post the
-            terminal released marker if — and only if — cleanup is authorized.
+            terminal released marker if — and only if — cleanup is allowed.
-        expected_session_id: Optional lease session the caller expects; a
-            mismatch against the live lease fails closed.
-        expected_candidate_head: Optional leased head the caller expects; a
-            mismatch against the live lease fails closed.
-        expected_lease_comment_id: Optional lease marker id the caller expects;
-            a mismatch against the live lease fails closed.
-        worktree_path: Reconciler worktree to bind the apply-path preflight to.
         remote: Known instance — 'dadeschools' or 'prgs'.
         host: Override the Gitea host.
-        org: Override the owner/organization. Validated against the canonical
+        org: Override the owner/organization.
-            repository identity; it can never redirect the mutation.
+        repo: Override the repository name.
-        repo: Override the repository name. Validated as above.
 
     Returns:
         dict reporting PR merged/closed state, merge_commit_sha, linked-issue
@@ -12513,62 +12546,14 @@ def gitea_cleanup_post_merge_moot_lease(
         "reasons": assessment.get("reasons") or [],
     }
 
-    # The canonical repository identity comes from the session binding only —
-    # never from the org/repo request parameters (#745 requirement 10).
-    repository_slug = _bound_repository_slug(remote)
-    report["repository_slug"] = repository_slug
-    report["required_task"] = post_merge_moot_lease_gate.CLEANUP_TASK
-    report["required_role_kind"] = post_merge_moot_lease_gate.REQUIRED_ROLE
-
     if not apply:
-        # Append-only dry-run evidence. Recorded for every assessment, allowed
-        # or not, so a later apply can prove the lease it saw is the lease that
-        # is still there. Prior entries are never rewritten.
-        evidence = post_merge_moot_lease_gate.record_dry_run(
-            pr_number=pr_number,
-            repository_slug=repository_slug,
-            lease_moot=bool(assessment.get("is_moot")),
-            cleanup_allowed=bool(assessment.get("cleanup_allowed")),
-            session_id=active.get("session_id"),
-            candidate_head=active.get("candidate_head"),
-            lease_comment_id=active.get("comment_id"),
-        )
-        report["dry_run_evidence"] = evidence
         return report
-
-    # ---------------------------------------------------------------- apply --
-    # Preserve the existing dry-run safety assessment: a lease that is not moot
-    # (open PR, already-terminal, absent) is refused here exactly as before, and
-    # no mutation is attempted.
     if not assessment.get("cleanup_allowed"):
         report["cleanup_skipped_reason"] = (
             assessment.get("reasons") or ["cleanup not allowed"]
         )
         return report
 
-    # 1 + 2 + 6: exact resolved cleanup task, reconciler role, and matching
-    # append-only dry-run evidence. Permission alone is deliberately not enough.
-    authorization = post_merge_moot_lease_gate.assess_apply_authorization(
-        pr_number=pr_number,
-        repository_slug=repository_slug,
-        resolved_task=_preflight_resolved_task,
-        active_role_kind=_profile_role_kind(get_profile()),
-        assessment=assessment,
-        evidence=post_merge_moot_lease_gate.latest_dry_run(
-            pr_number=pr_number, repository_slug=repository_slug
-        ),
-        expected_session_id=expected_session_id,
-        expected_candidate_head=expected_candidate_head,
-        expected_lease_comment_id=expected_lease_comment_id,
-    )
-    report["authorization"] = authorization
-    if not authorization["allowed"]:
-        report["success"] = False
-        report["reasons"] = authorization["reasons"]
-        report["blocker_kind"] = authorization["blocker_kind"]
-        return report
-
-    # 3: the dedicated mutation permission.
     comment_block = _profile_operation_gate("gitea.pr.comment")
     if comment_block:
         report["success"] = False
@@ -12576,28 +12561,7 @@ def gitea_cleanup_post_merge_moot_lease(
         report["permission_report"] = _permission_block_report("gitea.pr.comment")
         return report
 
-    # 4: explicit target must agree with the canonical repository identity
+    verify_preflight_purity(remote)
-    # before any preflight or mutation (#733/#739).
-    repo_binding_block = _repository_binding_block(
-        remote, org=org, repo=repo,
-        required_permission="gitea.pr.comment",
-    )
-    if repo_binding_block:
-        report["success"] = False
-        report["reasons"] = repo_binding_block["reasons"]
-        report["blocker_kind"] = repo_binding_block["blocker_kind"]
-        return report
-
-    # 4 (cont.): canonical root, workspace binding and the resolved-task match
-    # are enforced by the shared preflight, with the explicit target forwarded
-    # so the #604 anti-stomp resolution validates the targeted repository.
-    verify_preflight_purity(
-        remote,
-        worktree_path=worktree_path,
-        task=post_merge_moot_lease_gate.CLEANUP_TASK,
-        org=org,
-        repo=repo,
-    )
     body = assessment["release_body"]
     comment_url = f"{repo_api_url(h, o, r)}/issues/{pr_number}/comments"
     with _audited(
@@ -15499,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."""
+    """Read Gitea branch protection ``block_on_outdated_branch`` when present.
-    branch = (base_branch or "").strip()
+
-    if not branch:
+    Thin accessor over :func:`_branch_protection_policy`; return semantics are
-        return None
+    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.
+        payload = api_request(
-        protections = api_request(
+            "GET", f"{base_url}/commits/{head}/status", auth
-            "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"))
     except Exception:
-        return None
+        return snapshot
-    return None
+    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(
@@ -15783,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 = protection_policy.get(
-            _branch_protection_requires_current_base(
+            "requires_current_base"
-                base, auth, base_branch=base_branch
-            )
         )
     # When protection cannot be read, fail closed by requiring current base
     # whenever the PR is behind (safer for protected repos). Callers may pass
@@ -15854,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"
+        checks_status = checks_classification.get("checks_status")
-        # Combined status on PR head when Actions/status API is available.
+    elif checks_classification.get("checks_status") != pr_sync_status.CHECKS_UNKNOWN:
-        if pr_head_sha:
+        # Live evidence outranks a caller-supplied value; the override only
-            try:
+        # applies when live status could not be classified at all.
-                st = api_request(
+        checks_status = checks_classification.get("checks_status")
-                    "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"
 
     assessment = pr_sync_status.assess_pr_sync_status(
         host=h,
@@ -15891,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

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

post_merge_moot_lease_gate.py

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

pr_sync_status.py

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

role_session_router.py

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

task_capability_map.py

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

tests/test_issue_745_moot_lease_reconciler_gate.py

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

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

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