feat: make master-parity live-remote aware so a stale daemon fails closed (Closes #610)

Master-parity previously compared only the daemon's startup commit against the
local on-disk HEAD. When the checkout was not pulled, parity reported green even
though the live remote master had advanced, so a stale daemon could claim a
mutation-safe result while running outdated capability gates (observed during
PR #592 recovery, where the resolver correctly required restart but parity said
in_parity=true).

Changes:
- master_parity_gate.assess_master_parity() gains an optional live_remote_head
  and reports the three commits distinctly (daemon_start_head, local_head,
  live_remote_head) plus live_known / live_stale / mutation_safe. A result is
  mutation_safe only when daemon, local checkout, and live remote all agree.
- parity_block_reasons() now blocks mutations on live-staleness too; read-only
  operations remain unblocked (non-goal: never block diagnostics offline).
- New parity_resolver_disagreement(): typed fail-closed blocker naming the
  capability resolver as authoritative when it requires restart but parity
  looks locally green.
- New read_remote_master_head(): best-effort `git ls-remote` for the live
  target, cached with a 60s TTL (bounded offline latency, no network probe per
  gate call); env override GITEA_TEST_LIVE_REMOTE_HEAD keeps tests hermetic.
- Server: _current_master_parity() reads the live remote head;
  gitea_assess_master_parity and runtime_context surface the distinguished
  SHAs, mutation_safe, and resolver-authoritative guidance.

Tests: 13 new cases (live-remote parity, live-stale blocking + typed blocker,
remote-head reader + TTL cache, server wiring). Full suite: 2423 passed; the 8
remaining failures (test_config TestAuthIntegration, test_credentials
TestGetCredentials) are baseline-proven keychain/env failures identical on the
unmodified base.

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

controller_closure_baseline_proof.py

@@ -0,0 +1,63 @@
+"""Controller-closure baseline-proof gate (#529, criterion 6).
+
+A controller (or any session) may close a tracking issue with a closure
+report that summarizes validation. When that report calls a non-zero
+test-suite exit an "expected pre-existing failure" (or baseline / known
+failure) it must carry pre-merge proof; otherwise the closure buries an
+unproven regression as an accepted baseline and weakens controller
+confidence in the durable state.
+
+This module fails closed on exactly that case by reusing the pre-merge
+baseline proof verifier (#533). A closure report is allowed when it is
+empty (no validation claim), a clean pass, or a baseline failure proven on
+the PR pre-merge base commit (or a documented known-failure record that
+predates the PR).
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from premerge_baseline_proof import CLEAN_PASS, assess_premerge_baseline_proof
+
+
+def assess_controller_closure_baseline_proof(
+    closure_report: str | None,
+) -> dict[str, Any]:
+    """Assess whether an issue-closure report may proceed.
+
+    Returns a dict with ``block``, ``proven``, ``label``, ``reasons``,
+    ``skipped`` and ``safe_next_action``. Only blocks when the closure
+    report claims a non-zero validation exit is an expected
+    pre-existing/baseline failure without valid pre-merge proof.
+    """
+    text = (closure_report or "").strip()
+    if not text:
+        return {
+            "block": False,
+            "proven": True,
+            "label": CLEAN_PASS,
+            "reasons": [],
+            "skipped": True,
+            "safe_next_action": "",
+        }
+
+    result = assess_premerge_baseline_proof(text)
+    block = bool(result.get("block"))
+    return {
+        "block": block,
+        "proven": not block,
+        "label": result.get("label"),
+        "reasons": list(result.get("reasons") or []),
+        "skipped": bool(result.get("skipped", False)),
+        "safe_next_action": (
+            result.get("safe_next_action")
+            or (
+                "provide pre-merge baseline proof (base commit, tested commit, "
+                "command, exit status, failure signature) before closing on an "
+                "expected pre-existing failure"
+            )
+        )
+        if block
+        else "",
+    }

docs/llm-workflow-runbooks.md

@@ -1049,6 +1049,72 @@ Special states:
 Final reports must not claim a comment was posted when
 `canonical_comment_validation.allowed` is false (#496 AC14).
 
+## Stale #332 review-decision lock cleanup (#594)
+
+#332 hard-stops a reviewer session after a terminal live review mutation
+(`approve` / `request_changes`). After #559 those locks are **durable** under
+`~/.cache/gitea-tools/session-state/review_decision_lock-<profile>.json`, so they
+can outlive the PR they protected.
+
+### When cleanup is **allowed**
+
+Use `gitea_cleanup_stale_review_decision_lock` from a **reviewer** profile when:
+
+1. A durable review-decision lock has a terminal live mutation, **and**
+2. Live Gitea state shows that terminal PR is already **merged** or **closed**
+   (so no same-PR merge sequence remains), **and**
+3. The active profile identity matches the lock, **and**
+4. Authenticated identity can be verified.
+
+Workflow:
+
+```text
+# 1) Assess only (default)
+gitea_cleanup_stale_review_decision_lock(apply=false, remote=prgs, ...)
+
+# 2) Apply after is_moot=true and cleanup_allowed=true
+gitea_cleanup_stale_review_decision_lock(
+  apply=true,
+  expected_terminal_pr=<merged-or-closed-pr>,
+  remote=prgs,
+  ...
+)
+```
+
+Successful apply:
+
+* clears in-memory + durable decision lock for that profile
+* returns a structured `audit` payload
+* posts an audit comment on the mooted PR when `gitea.pr.comment` is allowed
+  (`post_audit_comment=true` default)
+
+Same-profile auto-expire: if `gitea_merge_pr` succeeds on a PR that this same
+profile just approved, the decision lock is cleared after merge. Cross-profile
+locks (reviewer vs merger) still require the cleanup tool.
+
+### When cleanup is **forbidden**
+
+Do **not** clear when:
+
+* the last terminal PR is still **open** (active #332 hard-stop — continue merge
+  for that approve, or stop after request_changes)
+* live PR state cannot be fetched (fail closed)
+* profile identity does not match the lock
+* identity cannot be verified
+* `expected_terminal_pr` does not match the last terminal PR
+
+**Never** use manual deletion of session-state files as the normal workflow.
+**Never** use cleanup to skip review of an open PR or to bypass eligibility /
+mark-ready / submit gates on active work.
+
+### Related tools
+
+| Tool | Purpose |
+|------|---------|
+| `gitea_cleanup_stale_review_decision_lock` | Moot decision-lock cleanup (#594) |
+| `gitea_authorize_review_correction` | Operator-approved correction after a **mistaken** live review on still-active work (#211) |
+| `gitea_cleanup_post_merge_moot_lease` | Moot **lease** cleanup after merge (#515) — different object |
+
 ## Fail-closed behavior
 
 Before any mutating action the workflow verifies identity, active profile,

docs/wiki/Safety-and-Gates.md

@@ -15,5 +15,7 @@
 5. **Explicit merge confirmation** — `gitea_merge_pr` requires `confirmation="MERGE PR <n>"`.
 6. **No self-review / self-merge** — Authenticated user must differ from PR author for approve/merge.
 7. **Review decision lock** — Live review mutations require validation-phase dry-run and `gitea_mark_final_review_decision`.
-8. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
-9. **Wiki publication (#224)** — `docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).
+8. **Terminal review hard-stop (#332)** — After a terminal live review mutation, only same-PR merge after `approve` may continue. Durable locks (#559) must not be deleted by hand.
+9. **Stale decision-lock cleanup (#594)** — `gitea_cleanup_stale_review_decision_lock` may clear a durable #332 lock **only** when the last terminal mutation's PR is live-state **merged or closed**, identity/profile gates pass, and apply uses a reviewer-capable profile. Open/ambiguous locks stay fail-closed. Successful cleanup records a durable audit trail.
+10. **Redaction** — Tokens, passwords, and keychain material never appear in tool output.
+11. **Wiki publication (#224)** — `docs/wiki/` and the sync helper are prerequisites only. Closing a wiki issue requires live Gitea Wiki proof on the repo Wiki tab. See [Runbooks](Runbooks.md#wiki-publication-readiness-gate-224).

final_report_validator.py

@@ -40,6 +40,7 @@ FINAL_REPORT_TASK_KINDS = frozenset({
     "issue_filing",
     "issue_selection",
     "inventory",
+    "controller_close",
 })
 
 _TASK_KIND_ALIASES = {
@@ -51,6 +52,9 @@ _TASK_KIND_ALIASES = {
     "reconcile_already_landed": "reconcile_already_landed",
     "work-issue": "work_issue",
     "create_issue": "issue_filing",
+    "close_issue": "controller_close",
+    "close-issue": "controller_close",
+    "controller-close": "controller_close",
 }
 
 _HANDOFF_ROLE_BY_TASK = {
@@ -62,6 +66,7 @@ _HANDOFF_ROLE_BY_TASK = {
     "issue_filing": "issue_filing",
     "issue_selection": None,
     "inventory": "inventory",
+    "controller_close": None,
 }
 
 _LEGACY_WORKSPACE_MUTATIONS_RE = re.compile(
@@ -850,6 +855,27 @@ def _rule_reviewer_premerge_baseline_proof(report_text: str) -> list[dict[str, s
     )
 
 
+def _rule_reviewer_post_merge_validation(report_text: str) -> list[dict[str, str]]:
+    """Block active approval on an already-merged/closed PR, and post-merge moot
+    validation claimed without merged-state + merge-commit proof (#529)."""
+    from post_merge_validation import assess_post_merge_validation
+
+    result = assess_post_merge_validation(report_text)
+    if not result.get("block"):
+        return []
+    return _findings_from_reasons(
+        "reviewer.post_merge_validation",
+        result.get("reasons") or [],
+        field="Validation status",
+        severity="block",
+        safe_next_action=result.get("safe_next_action")
+        or (
+            "record post-merge moot validation instead of an active approval; "
+            "cite PR state merged/closed and the merge commit SHA"
+        ),
+    )
+
+
 def _rule_reviewer_validation_status_vocabulary(
     report_text: str,
     *,
@@ -1514,6 +1540,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_linked_issue,
         _rule_reviewer_baseline_on_failure,
         _rule_reviewer_premerge_baseline_proof,
+        _rule_reviewer_post_merge_validation,
         _rule_reviewer_validation_status_vocabulary,
         _rule_reviewer_main_checkout_baseline,
         _rule_reviewer_main_checkout_path,
@@ -1606,6 +1633,13 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         *_SHARED_CANONICAL_COMMENT_RULES,
         *_SHARED_ISSUE_LOCK_RULES,
     ],
+    # Controller issue closure (#529): a closure report must not bury an
+    # unproven non-zero suite exit as an "expected pre-existing failure".
+    # Kept intentionally narrow so a closure pre-check does not demand the
+    # full reviewer/author handoff schema.
+    "controller_close": [
+        _rule_reviewer_premerge_baseline_proof,
+    ],
 }
 
 

gitea_mcp_server.py

@@ -159,6 +159,17 @@ if PROJECT_ROOT not in sys.path:
     sys.path.insert(0, PROJECT_ROOT)
 
 import json
+import subprocess
+
+_process_boot_head_sha = None
+try:
+    _process_boot_head_sha = subprocess.run(
+        ["git", "rev-parse", "HEAD"],
+        capture_output=True, text=True, check=True, cwd=PROJECT_ROOT
+    ).stdout.strip()
+except Exception:
+    pass
+
 
 _preflight_whoami_called = False
 _preflight_capability_called = False
@@ -211,6 +222,21 @@ def _effective_workspace_role() -> str:
     )
 
 
+def _profile_role_kind(profile: dict) -> str:
+    """Resolve a profile's declared role before inferring from permissions."""
+    role = (profile.get("role") or profile.get("role_kind") or "").strip()
+    if role:
+        return role
+    profile_name = (profile.get("profile_name") or "").strip().lower()
+    for candidate in ("reconciler", "merger", "reviewer", "author"):
+        if candidate in profile_name:
+            return candidate
+    return _role_kind(
+        profile.get("allowed_operations") or [],
+        profile.get("forbidden_operations") or [],
+    )
+
+
 def _actual_profile_role() -> str:
     """Resolve the workspace role from the ACTIVE PROFILE alone (#540).
 
@@ -223,10 +249,7 @@ def _actual_profile_role() -> str:
     ``author`` here, so author blocking is preserved.
     """
     profile = get_profile()
-    role = _role_kind(
-        profile.get("allowed_operations") or [],
-        profile.get("forbidden_operations") or [],
-    )
+    role = _profile_role_kind(profile)
     return nwb.normalize_role_kind(
         role,
         profile_name=profile.get("profile_name"),
@@ -807,6 +830,7 @@ import review_proofs  # noqa: E402
 import review_workflow_boundary  # noqa: E402
 import review_workflow_load  # noqa: E402
 import mcp_session_state  # noqa: E402
+import stale_review_decision_lock  # noqa: E402
 import agent_temp_artifacts
 import issue_lock_worktree  # noqa: E402
 import issue_lock_provenance  # noqa: E402
@@ -3096,10 +3120,15 @@ def terminal_review_hard_stop_reasons(pr_number: int, operation: str) -> list[st
         )
     else:
         guidance = "the session must stop and produce a final report"
+    recovery = (
+        "if that PR is already merged/closed, use "
+        "gitea_cleanup_stale_review_decision_lock (apply=true) after live-state "
+        "proof (#594); never delete session-state files by hand"
+    )
     return [
         "terminal review mutation already consumed in this run "
         f"({last.get('action')} on PR #{last.get('pr_number')}); {guidance} "
-        "(fail closed, #332)"
+        f"(fail closed, #332); {recovery}"
     ]
 
 
@@ -3872,6 +3901,205 @@ def gitea_authorize_review_correction(
     return {"authorized": True, "correction_reason": reason, "reasons": []}
 
 
+@mcp.tool()
+def gitea_cleanup_stale_review_decision_lock(
+    apply: bool = False,
+    expected_terminal_pr: int | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    post_audit_comment: bool = True,
+) -> dict:
+    """Clear a durable #332 review-decision lock only when it is moot (#594).
+
+    Read-first by default (``apply=False``). A lock is moot when its last
+    terminal live mutation references a PR that is already **merged** or
+    **closed** on live Gitea — so no same-PR merge sequence remains.
+
+    Fail-closed when:
+    - no lock / no terminal mutation
+    - live PR is still open (active #332 hard-stop preserved)
+    - live PR state cannot be fetched
+    - active profile identity does not match the lock
+    - apply requested without reviewer capability (``gitea.pr.review``)
+
+    Never weakens #332 for open PRs. Never instructs manual session-state
+    file deletion. On successful apply, clears memory + durable store and
+    returns a durable audit record (optionally posted as a PR comment).
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "cleanup_performed": False,
+            "is_moot": False,
+            "cleanup_allowed": False,
+            "mode": "apply" if apply else "read_only",
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+
+    h, o, r = _resolve(remote, host, org, repo)
+    auth = _auth(h)
+    binding = _decision_lock_binding()
+    active_identity = binding.get("profile_identity")
+    lock = _load_review_decision_lock()
+    last = stale_review_decision_lock.last_terminal_mutation(lock)
+    pr_live = None
+    pr_lookup_error = None
+    last_pr = last.get("pr_number") if last else None
+
+    if last_pr is not None:
+        try:
+            pr_live = api_request(
+                "GET", f"{repo_api_url(h, o, r)}/pulls/{last_pr}", auth
+            ) or {}
+            if not pr_live:
+                pr_lookup_error = "empty PR payload"
+        except Exception as exc:  # noqa: BLE001 — surface redacted
+            pr_lookup_error = _redact(str(exc))
+
+    assessment = stale_review_decision_lock.assess_stale_review_decision_lock(
+        lock,
+        pr_live=pr_live,
+        pr_lookup_error=pr_lookup_error,
+        active_profile_identity=active_identity,
+    )
+
+    # Optional pin: refuse apply against a different terminal PR than expected.
+    if (
+        expected_terminal_pr is not None
+        and assessment.get("last_terminal_pr") is not None
+        and int(expected_terminal_pr) != int(assessment["last_terminal_pr"])
+    ):
+        assessment = dict(assessment)
+        assessment["cleanup_allowed"] = False
+        assessment["reasons"] = list(assessment.get("reasons") or []) + [
+            f"expected_terminal_pr #{expected_terminal_pr} does not match "
+            f"last terminal PR #{assessment.get('last_terminal_pr')} "
+            "(fail closed, #594)"
+        ]
+
+    try:
+        actor = _authenticated_username(h)
+    except Exception:
+        actor = None
+    profile = get_profile()
+    profile_name = (profile.get("profile_name") or "").strip() or None
+
+    audit = stale_review_decision_lock.build_cleanup_audit_record(
+        assessment=assessment,
+        actor_username=actor,
+        profile_name=profile_name,
+        applied=False,
+    )
+
+    report = {
+        "success": True,
+        "cleanup_performed": False,
+        "mode": "apply" if apply else "read_only",
+        "remote": remote,
+        "org": o,
+        "repo": r,
+        "active_profile_identity": active_identity,
+        "authenticated_username": actor,
+        "has_lock": assessment.get("has_lock"),
+        "is_moot": assessment.get("is_moot"),
+        "cleanup_allowed": assessment.get("cleanup_allowed"),
+        "last_terminal_pr": assessment.get("last_terminal_pr"),
+        "last_terminal_action": assessment.get("last_terminal_action"),
+        "pr_state": assessment.get("pr_state"),
+        "pr_merged": assessment.get("pr_merged"),
+        "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
+        "merge_commit_sha": assessment.get("merge_commit_sha"),
+        "lock_summary": assessment.get("lock_summary"),
+        "audit": audit,
+        "audit_comment_id": None,
+        "reasons": list(assessment.get("reasons") or []),
+        "manual_file_delete_forbidden": True,
+    }
+
+    if not apply:
+        return report
+
+    if not assessment.get("cleanup_allowed"):
+        report["success"] = False
+        report["cleanup_skipped_reason"] = list(assessment.get("reasons") or [])
+        return report
+
+    if not actor:
+        report["success"] = False
+        report["reasons"] = [
+            "authenticated identity could not be verified; refuse lock "
+            "cleanup (fail closed, #594)"
+        ]
+        return report
+
+    review_block = _profile_operation_gate("gitea.pr.review")
+    if review_block:
+        report["success"] = False
+        report["reasons"] = review_block
+        report["permission_report"] = _permission_block_report("gitea.pr.review")
+        return report
+
+    session_reasons = _review_decision_session_reasons(lock)
+    if session_reasons:
+        report["success"] = False
+        report["reasons"] = session_reasons
+        return report
+
+    # Clear durable + in-memory lock only after all gates pass.
+    prior_summary = assessment.get("lock_summary")
+    _save_review_decision_lock(None)
+    audit = stale_review_decision_lock.build_cleanup_audit_record(
+        assessment=assessment,
+        actor_username=actor,
+        profile_name=profile_name,
+        applied=True,
+    )
+    audit["prior_lock_summary"] = prior_summary
+    report["cleanup_performed"] = True
+    report["audit"] = audit
+    report["reasons"] = list(assessment.get("reasons") or []) + [
+        f"cleared durable review decision lock for moot terminal mutation "
+        f"on PR #{assessment.get('last_terminal_pr')} (#594)"
+    ]
+
+    if post_audit_comment and assessment.get("last_terminal_pr") is not None:
+        comment_block = _profile_operation_gate("gitea.pr.comment")
+        if comment_block:
+            report["audit_comment_skipped"] = comment_block
+        else:
+            try:
+                body = stale_review_decision_lock.format_cleanup_audit_comment(
+                    audit
+                )
+                comment_url = (
+                    f"{repo_api_url(h, o, r)}/issues/"
+                    f"{assessment['last_terminal_pr']}/comments"
+                )
+                with _audited(
+                    "comment_pr",
+                    host=h,
+                    remote=remote,
+                    org=o,
+                    repo=r,
+                    pr_number=assessment["last_terminal_pr"],
+                    request_metadata={
+                        "source": "cleanup_stale_review_decision_lock"
+                    },
+                ):
+                    posted = api_request(
+                        "POST", comment_url, auth, {"body": body}
+                    )
+                report["audit_comment_id"] = (posted or {}).get("id")
+            except Exception as exc:  # noqa: BLE001
+                report["audit_comment_error"] = _redact(str(exc))
+
+    return report
+
+
 @mcp.tool()
 def gitea_dry_run_pr_review(
     pr_number: int,
@@ -4609,6 +4837,29 @@ def gitea_merge_pr(
             reviewer_pr_lease.get_session_lease()
         )
     )
+    # Same-profile auto-expire (#594): if *this* profile's decision lock ends
+    # with an approve of the PR just merged, clear it so durable state cannot
+    # block later unrelated reviews. Cross-profile locks (e.g. reviewer vs
+    # merger) still require gitea_cleanup_stale_review_decision_lock.
+    try:
+        lock_after = _load_review_decision_lock()
+        last_term = stale_review_decision_lock.last_terminal_mutation(lock_after)
+        if (
+            last_term
+            and last_term.get("action") == "approve"
+            and last_term.get("pr_number") == pr_number
+        ):
+            _save_review_decision_lock(None)
+            result["review_decision_lock_cleared_after_merge"] = True
+            reasons.append(
+                f"cleared same-profile review decision lock after merge of "
+                f"approved PR #{pr_number} (#594)"
+            )
+        else:
+            result["review_decision_lock_cleared_after_merge"] = False
+    except Exception as clear_exc:  # noqa: BLE001 — never fail the merge
+        result["review_decision_lock_cleared_after_merge"] = False
+        result["review_decision_lock_clear_error"] = _redact(str(clear_exc))
     reasons.append(f"all gates passed; merged PR #{pr_number} via '{do}'")
     return result
 
@@ -6070,6 +6321,7 @@ def gitea_close_issue(
     host: str | None = None,
     org: str | None = None,
     repo: str | None = None,
+    closure_report: str | None = None,
 ) -> dict:
     """Close an issue by setting its state to 'closed'.
 
@@ -6079,6 +6331,9 @@ def gitea_close_issue(
         host: Override the Gitea host.
         org: Override the owner/organization.
         repo: Override the repository name.
+        closure_report: Optional validation/closure summary. When it claims a
+            non-zero test-suite exit is an "expected pre-existing"/baseline
+            failure without pre-merge proof, the close fails closed (#529).
 
     Returns:
         dict with 'success' boolean and 'message'.
@@ -6088,6 +6343,25 @@ def gitea_close_issue(
     if blocked:
         return blocked
     verify_preflight_purity(remote, task="close_issue")
+
+    # #529: block closure that buries an unproven non-zero suite exit as an
+    # "expected pre-existing failure". Skipped when no closure report is given.
+    from controller_closure_baseline_proof import (
+        assess_controller_closure_baseline_proof,
+    )
+    closure_gate = assess_controller_closure_baseline_proof(closure_report)
+    if closure_gate["block"]:
+        return {
+            "success": False,
+            "blocked": True,
+            "message": (
+                f"Issue #{issue_number} close blocked (#529): closure report "
+                "claims an expected pre-existing/baseline failure without "
+                "pre-merge proof."
+            ),
+            "reasons": closure_gate["reasons"],
+            "safe_next_action": closure_gate["safe_next_action"],
+        }
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
@@ -6335,10 +6609,39 @@ def _try_auto_switch_for_operation(op: str, host: str | None = None) -> bool:
     return False
 
 
+def _git_default_remote_name(root: str) -> str:
+    """First configured git remote name for *root*, defaulting to 'origin'.
+
+    Used to resolve the live remote master target for parity (#610). Best
+    effort: any failure falls back to 'origin' so callers never raise.
+    """
+    try:
+        res = subprocess.run(
+            ["git", "-C", root, "remote"],
+            capture_output=True, text=True, check=False,
+        )
+    except Exception:
+        return "origin"
+    if res.returncode != 0:
+        return "origin"
+    names = [n.strip() for n in (res.stdout or "").splitlines() if n.strip()]
+    return names[0] if names else "origin"
+
+
 def _current_master_parity() -> dict:
-    """Assess this process's code against the on-disk master HEAD (#420)."""
+    """Assess this process's code against local and live remote master (#420/#610).
+
+    Compares the daemon's startup commit, the on-disk checkout HEAD, and the
+    live remote master target. A stale daemon relative to live master fails
+    closed for mutations even when the local checkout HEAD still matches the
+    startup commit. The live-remote read is best effort: an unresolved live
+    head leaves read-only diagnostics unblocked but is never mutation-safe.
+    """
     current_head = master_parity_gate.read_git_head(PROJECT_ROOT)
-    return master_parity_gate.assess_master_parity(_STARTUP_PARITY, current_head)
+    live_head = master_parity_gate.read_remote_master_head(
+        PROJECT_ROOT, remote=_git_default_remote_name(PROJECT_ROOT))
+    return master_parity_gate.assess_master_parity(
+        _STARTUP_PARITY, current_head, live_remote_head=live_head)
 
 
 def _master_parity_block(op: str) -> list[str]:
@@ -6915,6 +7218,62 @@ def gitea_assess_reviewer_pr_lease(
     }
 
 
+@mcp.tool()
+def gitea_diagnose_reviewer_pr_lease_handoff(
+    pr_number: int,
+    proposed_worktree: str | None = None,
+    instructed_session_id: str | None = None,
+    instructed_comment_id: int | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: diagnose open-PR reviewer lease handoff and emit next action (#599).
+
+    Classifies no-lease / own / foreign / instructed-lease-missing-with-replacement
+    and worktree binding mismatch. Returns a canonical ``next_action`` without
+    mutating Gitea state. Never steals foreign leases.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "reasons": read_block,
+            "permission_report": _permission_block_report("gitea.read"),
+        }
+    comments = _fetch_pr_comments(
+        pr_number, remote=remote, host=host, org=org, repo=repo)
+    h, _o, _r = _resolve(remote, host, org, repo)
+    try:
+        username = _authenticated_username(h)
+    except Exception:
+        username = None
+    session_lease = reviewer_pr_lease.get_session_lease() or {}
+    env_wt = (
+        (os.environ.get("GITEA_REVIEWER_WORKTREE") or "").strip()
+        or (os.environ.get("GITEA_ACTIVE_WORKTREE") or "").strip()
+        or None
+    )
+    # Prefer in-session lease id when present; otherwise diagnose as a fresh
+    # caller without inventing ownership of an on-thread lease.
+    session_id = (session_lease.get("session_id") or "").strip() or None
+    diagnosis = reviewer_pr_lease.diagnose_reviewer_pr_lease_handoff(
+        comments,
+        pr_number=pr_number,
+        current_session_id=session_id,
+        current_reviewer_identity=username,
+        proposed_worktree=proposed_worktree,
+        env_bound_worktree=env_wt,
+        instructed_session_id=instructed_session_id,
+        instructed_comment_id=instructed_comment_id,
+    )
+    diagnosis["success"] = True
+    diagnosis["remote"] = remote
+    diagnosis["authenticated_user"] = username
+    return diagnosis
+
+
 @mcp.tool()
 def gitea_cleanup_post_merge_moot_lease(
     pr_number: int,
@@ -8612,15 +8971,34 @@ def gitea_get_runtime_context(
         "restart_required": parity["restart_required"],
         "startup_head": parity["startup_head"],
         "current_head": parity["current_head"],
+        # #610 distinguished mutation-safety signals:
+        "daemon_start_head": parity["daemon_start_head"],
+        "local_head": parity["local_head"],
+        "live_remote_head": parity["live_remote_head"],
+        "live_known": parity["live_known"],
+        "live_stale": parity["live_stale"],
+        "mutation_safe": parity["mutation_safe"],
         "summary": master_parity_gate.format_parity(parity),
         "mutation_gate_enforced": not master_parity_gate.gate_disabled(),
+        # #610: the capability resolver is authoritative for mutation safety;
+        # local parity alone must never authorize a mutation.
+        "resolver_authoritative_for_mutation_safety": True,
     }
-    if parity["stale"] and not master_parity_gate.gate_disabled():
-        safe_next_action = (
-            "Server code is stale relative to master; restart the Gitea MCP "
-            "server to load current capability gates before mutating. "
-            f"({master_parity_gate.format_parity(parity)})"
-        )
+    if parity["restart_required"] and not master_parity_gate.gate_disabled():
+        if parity["live_stale"]:
+            safe_next_action = (
+                "Daemon is stale relative to LIVE remote master "
+                f"(started {parity['startup_head'][:12] if parity['startup_head'] else 'unknown'}, "
+                f"live master {parity['live_remote_head'][:12] if parity['live_remote_head'] else 'unknown'}); "
+                "restart/reconnect the Gitea MCP server before mutating. The "
+                "capability resolver is authoritative for mutation safety."
+            )
+        else:
+            safe_next_action = (
+                "Server code is stale relative to master; restart the Gitea MCP "
+                "server to load current capability gates before mutating. "
+                f"({master_parity_gate.format_parity(parity)})"
+            )
         result["safe_next_action"] = safe_next_action
 
     if reveal and h:
@@ -8659,12 +9037,19 @@ def gitea_assess_master_parity(
         "determinable": parity["determinable"],
         "startup_head": parity["startup_head"],
         "current_head": parity["current_head"],
+        # #610 distinguished mutation-safety signals:
+        "daemon_start_head": parity["daemon_start_head"],
+        "local_head": parity["local_head"],
+        "live_remote_head": parity["live_remote_head"],
+        "live_known": parity["live_known"],
+        "live_stale": parity["live_stale"],
+        "mutation_safe": parity["mutation_safe"],
         "mutation_gate_enforced": enforced,
         "summary": master_parity_gate.format_parity(parity),
         "reasons": parity["reasons"],
         "process_root": PROJECT_ROOT,
     }
-    if parity["stale"] and enforced:
+    if parity["restart_required"] and enforced:
         out["report"] = master_parity_gate.parity_report(parity)
     return out
 def gitea_record_pre_review_command(
@@ -9798,6 +10183,34 @@ def gitea_route_task_session(
     )
 
 
+_restart_triggered = False
+
+
+def _trigger_mcp_auto_restart():
+    global _restart_triggered
+    if _restart_triggered or _preflight_in_test_mode():
+        return
+    _restart_triggered = True
+
+    config_path = os.environ.get(
+        "MCP_CONFIG_PATH",
+        os.path.expanduser("~/.gemini/config/mcp_config.json")
+    )
+
+    try:
+        if os.path.exists(config_path):
+            os.utime(config_path, None)
+    except Exception:
+        pass
+
+    import threading
+    import time
+    def delayed_exit():
+        time.sleep(1.0)
+        os._exit(0)
+    threading.Thread(target=delayed_exit, daemon=True).start()
+
+
 def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) -> list[str]:
     """Check running runtimes and return errors if they are missing or stale."""
     import subprocess
@@ -9819,6 +10232,19 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
     except Exception as exc:
         return [f"stale-runtime: failed to list running processes: {exc}"]
 
+    current_head_sha = None
+    try:
+        current_head_sha = subprocess.run(
+            ["git", "rev-parse", "HEAD"],
+            capture_output=True, text=True, check=True, cwd=PROJECT_ROOT
+        ).stdout.strip()
+    except Exception:
+        pass
+
+    git_stale = False
+    if _process_boot_head_sha and current_head_sha and _process_boot_head_sha != current_head_sha:
+        git_stale = True
+
     self_pid = os.getpid()
     self_stale = False
 
@@ -9854,7 +10280,7 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
         if match:
             profile = match.group(1)
 
-        is_stale = start_time < code_mtime
+        is_stale = (start_time < code_mtime) or git_stale
         if pid == self_pid and is_stale:
             self_stale = True
 
@@ -9866,9 +10292,11 @@ def _check_mcp_runtimes_diagnostics(task: str, matching_profiles: list[str]) ->
             }
 
     if self_stale:
+        _trigger_mcp_auto_restart()
         reasons.append(
             "stale-runtime: The active Gitea MCP server process is stale (running code from before changes were merged). "
-            "Please fully restart the Gitea MCP server (e.g. touch /Users/jasonwalker/.gemini/config/mcp_config.json) and retry."
+            "Auto-restart has been triggered: touched mcp_config.json to reload the daemon. "
+            "The current process will cleanly exit shortly."
         )
 
     if matching_profiles:
@@ -9917,6 +10345,24 @@ def gitea_resolve_task_capability(
 
     required_permission = task_capability_map.required_permission(task)
     required_role = task_capability_map.required_role(task)
+    role_exclusive_tasks = {
+        "review_pr",
+        "approve_pr",
+        "request_changes_pr",
+        "blind_pr_queue_review",
+        "pr_queue_cleanup",
+        "pr-queue-cleanup",
+        "merge_pr",
+        "create_branch",
+        "push_branch",
+        "create_pr",
+        "commit_files",
+        "gitea_commit_files",
+        "address_pr_change_requests",
+        "delete_branch",
+        "work_issue",
+        "work-issue",
+    }
 
     infra_assessment = role_session_router.assess_infra_stop(PROJECT_ROOT)
     if required_role == "reviewer":
@@ -10011,11 +10457,26 @@ def gitea_resolve_task_capability(
     # Load active permissions
     active_allowed = profile.get("allowed_operations") or []
     active_forbidden = profile.get("forbidden_operations") or []
+    active_role_kind = _profile_role_kind(profile)
 
     # Check if allowed in current session
-    allowed_in_current_session, _ = gitea_config.check_operation(
+    permission_allowed_in_current_session, _ = gitea_config.check_operation(
         required_permission, active_allowed, active_forbidden
     )
+    role_matches_current_session = (
+        task not in role_exclusive_tasks
+        or active_role_kind == required_role
+    )
+    role_mismatch_reason = None
+    if not role_matches_current_session:
+        role_mismatch_reason = (
+            f"Active profile role '{active_role_kind}' cannot perform "
+            f"{required_role} task '{task}' even if nearby permissions are "
+            "present (fail closed)."
+        )
+    allowed_in_current_session = (
+        permission_allowed_in_current_session and role_matches_current_session
+    )
 
     switching = gitea_config.is_runtime_switching_enabled()
     available_in_session = allowed_in_current_session
@@ -10042,7 +10503,13 @@ def gitea_resolve_task_capability(
                 except Exception:
                     pass
             ok, _ = gitea_config.check_operation(required_permission, p_allowed_n, p_forbidden_n)
-            if ok:
+            p_role = (p_data.get("role") or "").strip()
+            p_role_ok = (
+                task not in role_exclusive_tasks
+                or not p_role
+                or p_role == required_role
+            )
+            if ok and p_role_ok:
                 matching_profiles.append(p_name)
 
     configured = len(matching_profiles) > 0
@@ -10070,6 +10537,8 @@ def gitea_resolve_task_capability(
             reason_msg = (
                 f"No profile configured with permission '{required_permission}'."
             )
+        elif role_mismatch_reason:
+            reason_msg = role_mismatch_reason
     different_namespace_required = False
     next_safe_action = "None; ready for operations."
 
@@ -10101,6 +10570,8 @@ def gitea_resolve_task_capability(
     # into author-side mutations.
     stop_required = not allowed_in_current_session
     task_role_guidance = []
+    if role_mismatch_reason:
+        task_role_guidance.append(f"STOP: {role_mismatch_reason}")
     if required_role == "reviewer":
         if allowed_in_current_session:
             task_role_guidance.append(
@@ -10146,7 +10617,9 @@ def gitea_resolve_task_capability(
         "required_role_kind": required_role,
         "active_profile": profile["profile_name"],
         "active_identity": username,
+        "active_role_kind": active_role_kind,
         "active_profile_allowed_operations": active_allowed,
+        "active_profile_permission_allowed": permission_allowed_in_current_session,
         "allowed_in_current_session": allowed_in_current_session,
         "available_in_session": available_in_session,
         "configured": configured,

issue_workflow_labels.py

@@ -42,12 +42,38 @@ STATUS_LABEL_SPECS: tuple[LabelSpec, ...] = (
     LabelSpec("status:wontfix", "000000", "Issue will not be fixed"),
 )
 
+# Durable validation-outcome labels (#529): the four canonical distinctions a
+# reviewer report can carry. These are orthogonal to the single active status:*
+# label, so they use their own prefix and are not subject to the one-status
+# invariant.
+VALIDATION_LABEL_SPECS: tuple[LabelSpec, ...] = (
+    LabelSpec("validation:clean-pass", "0e8a16", "Validation was a clean pass"),
+    LabelSpec(
+        "validation:baseline-accepted",
+        "fbca04",
+        "Validation passed with a baseline-proven unrelated failure",
+    ),
+    LabelSpec(
+        "validation:blocked",
+        "b60205",
+        "Validation blocked by an unresolved failure",
+    ),
+    LabelSpec(
+        "validation:post-merge-moot",
+        "5319e7",
+        "Validation is post-merge moot (PR already merged/closed before review)",
+    ),
+)
+
 CANONICAL_LABEL_SPECS: tuple[LabelSpec, ...] = (
-    TYPE_LABEL_SPECS + STATUS_LABEL_SPECS
+    TYPE_LABEL_SPECS + STATUS_LABEL_SPECS + VALIDATION_LABEL_SPECS
 )
 
 TYPE_LABELS: frozenset[str] = frozenset(spec.name for spec in TYPE_LABEL_SPECS)
 STATUS_LABELS: frozenset[str] = frozenset(spec.name for spec in STATUS_LABEL_SPECS)
+VALIDATION_LABELS: frozenset[str] = frozenset(
+    spec.name for spec in VALIDATION_LABEL_SPECS
+)
 CANONICAL_LABELS: frozenset[str] = frozenset(
     spec.name for spec in CANONICAL_LABEL_SPECS
 )

master_parity_gate.py

@@ -24,12 +24,26 @@ from __future__ import annotations
 
 import os
 import subprocess
+import time
+
+# Live-remote head cache: the parity gate runs on every mutation and every
+# runtime-context read, so the ``git ls-remote`` result is cached briefly to
+# avoid a network round-trip per call (#610). Keyed by (root, remote, branch).
+_REMOTE_HEAD_CACHE: dict[tuple[str, str, str], tuple[float, str | None]] = {}
+_REMOTE_HEAD_TTL = 60.0
+
+
+def _clear_remote_head_cache() -> None:
+    """Reset the live-remote head cache (test isolation / forced refresh)."""
+    _REMOTE_HEAD_CACHE.clear()
 
 # Environment escape hatches (ops + tests):
 #   GITEA_MCP_DISABLE_PARITY_GATE  -> disable enforcement entirely (fail open).
 #   GITEA_TEST_CURRENT_HEAD        -> force the "current" HEAD read, for tests.
 ENV_DISABLE = "GITEA_MCP_DISABLE_PARITY_GATE"
 ENV_TEST_CURRENT_HEAD = "GITEA_TEST_CURRENT_HEAD"
+# GITEA_TEST_LIVE_REMOTE_HEAD -> force the live remote master read, for tests.
+ENV_TEST_LIVE_REMOTE_HEAD = "GITEA_TEST_LIVE_REMOTE_HEAD"
 
 
 def read_git_head(root: str) -> str | None:
@@ -58,6 +72,57 @@ def read_git_head(root: str) -> str | None:
     return (res.stdout or "").strip() or None
 
 
+def read_remote_master_head(
+    root: str,
+    remote: str = "origin",
+    branch: str = "master",
+    ttl: float = _REMOTE_HEAD_TTL,
+) -> str | None:
+    """Return the live remote ``branch`` commit SHA, or ``None`` (#610).
+
+    Resolves the *live* target commit via ``git ls-remote`` so parity can tell
+    a daemon that is behind the live remote master apart from one whose local
+    checkout simply hasn't been pulled. ``None`` means the live head could not
+    be resolved (offline, no such remote, git unavailable, error) -- callers
+    must treat unknown live state as *not mutation-safe* while never blocking
+    read-only diagnostics. A ``GITEA_TEST_LIVE_REMOTE_HEAD`` override takes
+    precedence so the wiring can be exercised deterministically and offline.
+
+    The result is cached for *ttl* seconds per (root, remote, branch) so the
+    gate does not run a network probe on every mutation/read (``ttl=0`` forces
+    a live probe). Both hits and ``None`` misses are cached to bound offline
+    latency; the env override bypasses the cache and the subprocess entirely.
+    """
+    forced = os.environ.get(ENV_TEST_LIVE_REMOTE_HEAD)
+    if forced is not None:
+        return forced.strip() or None
+    if not root:
+        return None
+    key = (root, remote, branch)
+    now = time.monotonic()
+    if ttl > 0:
+        cached = _REMOTE_HEAD_CACHE.get(key)
+        if cached is not None and (now - cached[0]) < ttl:
+            return cached[1]
+    sha: str | None = None
+    try:
+        res = subprocess.run(
+            ["git", "-C", root, "ls-remote", remote, f"refs/heads/{branch}"],
+            capture_output=True,
+            text=True,
+            check=False,
+            timeout=5,
+        )
+        if res.returncode == 0:
+            lines = (res.stdout or "").strip().splitlines()
+            if lines:
+                sha = lines[0].split("\t", 1)[0].split()[0].strip() or None
+    except Exception:
+        sha = None
+    _REMOTE_HEAD_CACHE[key] = (now, sha)
+    return sha
+
+
 def capture_startup_parity(root: str, head: str | None = None) -> dict:
     """Capture the process source-tree baseline once at server startup.
 
@@ -72,18 +137,38 @@ def _short(sha: str | None) -> str:
     return sha[:12] if sha else "unknown"
 
 
-def assess_master_parity(startup: dict | None, current_head: str | None) -> dict:
+def assess_master_parity(
+    startup: dict | None,
+    current_head: str | None,
+    live_remote_head: str | None = None,
+) -> dict:
     """Compare the startup baseline against the current on-disk ``HEAD``.
 
-    Pure: both HEADs are supplied by the caller. Returns a structured result:
+    Pure: all HEADs are supplied by the caller. Returns a structured result:
 
     - ``in_parity``    -- server code matches the on-disk master (or parity
       could not be determined, which is not treated as stale).
     - ``stale``        -- the on-disk master has definitively advanced past the
       running process.
-    - ``restart_required`` -- alias of ``stale``; the recovery action.
-    - ``determinable`` -- whether both HEADs were known well enough to compare.
+    - ``restart_required`` -- ``stale`` or ``live_stale``; the recovery action.
+    - ``determinable`` -- whether both local HEADs were known well enough to
+      compare.
     - ``startup_head`` / ``current_head`` / ``reasons``.
+
+    #610 adds live-remote awareness so a daemon that is stale relative to the
+    *live* remote master cannot report a mutation-safe result even when the
+    local checkout HEAD still matches the daemon's startup commit:
+
+    - ``daemon_start_head`` -- the commit the running process started at
+      (alias of ``startup_head``, named for clarity in reports).
+    - ``local_head`` -- the on-disk checkout HEAD (alias of ``current_head``).
+    - ``live_remote_head`` -- the live remote target commit, or ``None`` when it
+      could not be fetched.
+    - ``live_known`` -- whether the live remote target was resolved.
+    - ``live_stale`` -- the live remote master has advanced past the running
+      process (daemon is behind live master) even if local parity is green.
+    - ``mutation_safe`` -- the daemon code, local checkout, and live remote
+      target all agree; the only state in which a mutation may rely on parity.
     """
     startup_head = (startup or {}).get("startup_head")
     reasons: list[str] = []
@@ -91,32 +176,56 @@ def assess_master_parity(startup: dict | None, current_head: str | None) -> dict
     if startup_head is None:
         reasons.append(
             "startup commit was not captured; code parity cannot be enforced")
-        return _result(True, False, False, startup_head, current_head, reasons)
+        return _result(True, False, False, startup_head, current_head,
+                       live_remote_head, False, reasons)
 
     if current_head is None:
         reasons.append(
             "current workspace HEAD could not be read; code parity cannot be "
             "enforced")
-        return _result(True, False, False, startup_head, current_head, reasons)
+        return _result(True, False, False, startup_head, current_head,
+                       live_remote_head, False, reasons)
 
-    if startup_head == current_head:
-        return _result(True, False, True, startup_head, current_head, reasons)
+    local_in_parity = startup_head == current_head
+    local_stale = not local_in_parity
+    if local_stale:
+        reasons.append(
+            f"MCP server started at commit {_short(startup_head)} but the "
+            f"workspace master is now {_short(current_head)}; restart the "
+            f"server to load the current capability gates")
 
-    reasons.append(
-        f"MCP server started at commit {_short(startup_head)} but the workspace "
-        f"master is now {_short(current_head)}; restart the server to load the "
-        f"current capability gates")
-    return _result(False, True, True, startup_head, current_head, reasons)
+    live_known = live_remote_head is not None
+    live_stale = live_known and live_remote_head != startup_head
+    if live_stale:
+        reasons.append(
+            f"live remote master is {_short(live_remote_head)} but the MCP "
+            f"server started at {_short(startup_head)}; the daemon is stale "
+            f"relative to live master -- restart/reconnect before mutating")
+
+    return _result(
+        local_in_parity, local_stale, True, startup_head, current_head,
+        live_remote_head, live_stale, reasons)
 
 
-def _result(in_parity, stale, determinable, startup_head, current_head, reasons):
+def _result(in_parity, stale, determinable, startup_head, current_head,
+            live_remote_head, live_stale, reasons):
+    live_known = live_remote_head is not None
+    mutation_safe = (
+        determinable and in_parity and live_known and not live_stale)
     return {
         "in_parity": in_parity,
         "stale": stale,
-        "restart_required": stale,
+        "restart_required": stale or live_stale,
         "determinable": determinable,
         "startup_head": startup_head,
         "current_head": current_head,
+        # #610 distinguished signals:
+        "daemon_start_head": startup_head,
+        "local_head": current_head,
+        "live_remote_head": live_remote_head,
+        "live_known": live_known,
+        "live_stale": live_stale,
+        "mutation_safe": mutation_safe,
         "reasons": list(reasons),
     }
 
@@ -130,11 +239,13 @@ def parity_block_reasons(assessment: dict) -> list[str]:
     """Block reasons for a mutation gate (empty when the mutation may proceed).
 
     A disabled gate or an in-parity / non-determinable assessment yields no
-    reasons; only a definitively stale server blocks.
+    reasons. A definitively stale server blocks, and (#610) a daemon that is
+    stale relative to the *live* remote master blocks even when the local
+    checkout HEAD still matches the daemon's startup commit.
     """
     if gate_disabled():
         return []
-    if assessment.get("stale"):
+    if assessment.get("stale") or assessment.get("live_stale"):
         return list(assessment.get("reasons") or
                     ["server code is stale relative to master (fail closed)"])
     return []
@@ -147,6 +258,10 @@ def parity_report(assessment: dict) -> dict:
         "restart_required": True,
         "startup_head": assessment.get("startup_head"),
         "current_head": assessment.get("current_head"),
+        # #610: name the live remote target so the report distinguishes a
+        # local-code stale from a daemon-behind-live-master stale.
+        "live_remote_head": assessment.get("live_remote_head"),
+        "live_stale": bool(assessment.get("live_stale")),
         "reasons": list(assessment.get("reasons") or []),
         "recovery": [
             "The running MCP server is executing code older than the current "
@@ -157,6 +272,45 @@ def parity_report(assessment: dict) -> dict:
     }
 
 
+def parity_resolver_disagreement(
+    assessment: dict,
+    resolver_restart_required: bool,
+) -> dict | None:
+    """Typed blocker when the resolver requires restart but parity looks green.
+
+    The capability resolver (``gitea_resolve_task_capability``) detects stale
+    runtime authoritatively for mutation safety (#610). When it requires a
+    restart, local-only parity must never override it: this returns a typed,
+    fail-closed blocker that names the resolver as authoritative. Returns
+    ``None`` when the resolver does not require a restart.
+    """
+    if not resolver_restart_required:
+        return None
+    parity_optimistic = bool(assessment.get("in_parity")) and not (
+        assessment.get("stale") or assessment.get("live_stale"))
+    return {
+        "kind": "parity_resolver_disagreement",
+        "restart_required": True,
+        "resolver_authoritative": True,
+        "parity_optimistic": parity_optimistic,
+        "daemon_start_head": assessment.get("daemon_start_head"),
+        "local_head": assessment.get("local_head"),
+        "live_remote_head": assessment.get("live_remote_head"),
+        "reasons": [
+            "The capability resolver requires a restart/reconnect (stale "
+            "runtime) but master-parity reported local code as in-parity. "
+            "The resolver is authoritative for mutation safety; do not mutate "
+            "on local parity alone. Restart/reconnect the Gitea MCP server "
+            "and re-verify before mutating.",
+        ],
+        "recovery": [
+            "Trust the resolver: treat this session as stale.",
+            "Restart or /mcp reconnect the Gitea MCP namespace so it reloads "
+            "current master and live target state, then re-run preflight.",
+        ],
+    }
+
+
 def format_parity(assessment: dict) -> str:
     """One-line human summary for logs / runtime context."""
     if assessment.get("stale"):

post_merge_validation.py

@@ -0,0 +1,222 @@
+"""Canonical post-merge (moot) validation outcome and wording gate (#529).
+
+When a PR is already merged or closed *before* a reviewer submits their
+verdict, an ordinary "approve" is misleading: the review never gated the
+merge, so a normal active approval overstates controller confidence. The
+sanctioned outcome for that case is a **post-merge moot validation** — the
+reviewer records what validation found, but classifies it as moot rather than
+an active approval.
+
+This module defines the four canonical validation distinctions #529 requires
+and enforces the post-merge case:
+
+- ``CLEAN_PASS_OUTCOME`` — clean validation pass on an open, reviewable PR.
+- ``BASELINE_ACCEPTED_OUTCOME`` — validation pass with a baseline-proven
+  unrelated failure (proof handled by :mod:`premerge_baseline_proof`).
+- ``BLOCKED_OUTCOME`` — validation blocked by an unresolved failure.
+- ``POST_MERGE_MOOT_OUTCOME`` — the PR was already merged/closed before review
+  submission; validation is recorded as post-merge moot, not an active
+  approval.
+
+The gate blocks two mistakes:
+
+1. Claiming an *active approval* on a PR that was already merged/closed before
+   review submission (must be recorded as post-merge moot validation instead).
+2. Claiming post-merge moot validation without proof the PR is actually
+   merged/closed (a merged-state field plus a merge commit SHA).
+
+Each outcome maps to a durable ``validation:*`` process-state label so PRs and
+issues carry the distinction (#529 acceptance criterion).
+"""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+CLEAN_PASS_OUTCOME = "clean validation pass"
+BASELINE_ACCEPTED_OUTCOME = "validation pass with baseline-proven unrelated failure"
+BLOCKED_OUTCOME = "validation blocked by unresolved failure"
+POST_MERGE_MOOT_OUTCOME = "post-merge moot validation"
+
+# Canonical validation status label a reviewer writes in the report body for the
+# post-merge case; kept in sync with validation_status_vocabulary.
+POST_MERGE_MOOT_STATUS = "post-merge moot validation"
+
+# Durable process-state labels (registered in issue_workflow_labels).
+LABEL_CLEAN_PASS = "validation:clean-pass"
+LABEL_BASELINE_ACCEPTED = "validation:baseline-accepted"
+LABEL_BLOCKED = "validation:blocked"
+LABEL_POST_MERGE_MOOT = "validation:post-merge-moot"
+
+_OUTCOME_TO_LABEL: dict[str, str] = {
+    CLEAN_PASS_OUTCOME: LABEL_CLEAN_PASS,
+    BASELINE_ACCEPTED_OUTCOME: LABEL_BASELINE_ACCEPTED,
+    BLOCKED_OUTCOME: LABEL_BLOCKED,
+    POST_MERGE_MOOT_OUTCOME: LABEL_POST_MERGE_MOOT,
+}
+
+# The PR was already merged/closed before the review was submitted.
+_MERGED_BEFORE_REVIEW_RE = re.compile(
+    r"(?:already[- ]merged"
+    r"|merged\s+before\s+review"
+    r"|closed\s+before\s+review"
+    r"|pr\s+(?:is|was)\s+(?:already\s+)?(?:merged|closed)"
+    r"|pr\s+state\s*[:=]\s*(?:merged|closed)"
+    r"|merged\s*/\s*closed)",
+    re.IGNORECASE,
+)
+# An active approval verdict is being claimed.
+_ACTIVE_APPROVAL_RE = re.compile(
+    r"(?:review\s+decision\s*[:=]\s*approve"
+    r"|submitted\s+['\"]?approve['\"]?"
+    r"|active\s+approval"
+    r"|approving\s+the\s+pr"
+    r"|posting\s+an?\s+approval)",
+    re.IGNORECASE,
+)
+# The sanctioned post-merge moot validation wording.
+_POST_MERGE_MOOT_RE = re.compile(
+    r"post[- ]merge\s+(?:moot\s+)?validation"
+    r"|post[- ]merge\s+moot"
+    r"|moot\s+validation",
+    re.IGNORECASE,
+)
+# Proof the PR is genuinely merged/closed.
+_MERGED_STATE_PROOF_RE = re.compile(
+    r"(?:pr\s+state\s*[:=]\s*(?:merged|closed)"
+    r"|merged_at\s*[:=]\s*\S"
+    r"|closed_at\s*[:=]\s*\S"
+    r"|state\s*[:=]\s*(?:merged|closed))",
+    re.IGNORECASE,
+)
+_MERGE_COMMIT_SHA_RE = re.compile(
+    r"merge\s+commit(?:\s+sha)?\s*[:=]\s*[0-9a-f]{7,40}",
+    re.IGNORECASE,
+)
+
+
+POST_MERGE_VALIDATION_TEMPLATE = (
+    "Validation status: post-merge moot validation\n"
+    "PR state: merged\n"
+    "Merge commit sha: <40-hex merge commit>\n"
+    "Validation finding: <what the validation run showed, for the record>\n"
+    "Note: PR merged/closed before review submission; recorded as post-merge "
+    "moot validation, not an active approval."
+)
+
+
+def process_state_label(outcome: str) -> str | None:
+    """Return the durable ``validation:*`` label for a canonical outcome."""
+    return _OUTCOME_TO_LABEL.get(outcome)
+
+
+def assess_post_merge_validation(
+    report_text: str,
+    *,
+    pr_merged_or_closed: bool | None = None,
+) -> dict[str, Any]:
+    """Assess post-merge moot validation wording and proof (#529).
+
+    Args:
+        report_text: The reviewer final report text.
+        pr_merged_or_closed: Optional structured signal that the PR is already
+            merged/closed. When ``True`` it forces the merged-before-review
+            path even if the report text omits the phrasing; proof fields are
+            still required in the report body for durability.
+
+    Returns a dict with ``proven``, ``block``, ``outcome``,
+    ``process_state_label``, ``reasons``, ``skipped`` and ``safe_next_action``.
+    Only blocks when the report either claims an active approval on a
+    merged/closed PR, or claims post-merge moot validation without proof.
+    """
+    text = report_text or ""
+
+    merged_signal = bool(_MERGED_BEFORE_REVIEW_RE.search(text)) or bool(
+        pr_merged_or_closed
+    )
+    active_approval = bool(_ACTIVE_APPROVAL_RE.search(text))
+    moot_wording = bool(_POST_MERGE_MOOT_RE.search(text))
+
+    # Nothing about merged-state or moot validation — not applicable here.
+    if not merged_signal and not moot_wording:
+        return {
+            "proven": True,
+            "block": False,
+            "outcome": None,
+            "process_state_label": None,
+            "reasons": [],
+            "skipped": True,
+            "safe_next_action": "",
+        }
+
+    # An active approval on a PR already merged/closed before review, without
+    # the sanctioned moot wording, overstates confidence.
+    if merged_signal and active_approval and not moot_wording:
+        return {
+            "proven": False,
+            "block": True,
+            "outcome": POST_MERGE_MOOT_OUTCOME,
+            "process_state_label": LABEL_POST_MERGE_MOOT,
+            "reasons": [
+                "PR was already merged/closed before review submission; an "
+                "active approval overstates confidence — record 'post-merge "
+                "moot validation' instead of a normal approval"
+            ],
+            "skipped": False,
+            "safe_next_action": (
+                "classify the outcome as post-merge moot validation (not an "
+                "active approval); cite PR state merged/closed and the merge "
+                "commit SHA. Template:\n" + POST_MERGE_VALIDATION_TEMPLATE
+            ),
+        }
+
+    # Post-merge moot validation claimed — require merged-state + commit proof.
+    if moot_wording:
+        reasons: list[str] = []
+        if not _MERGED_STATE_PROOF_RE.search(text):
+            reasons.append(
+                "post-merge moot validation claimed without merged/closed state "
+                "proof (state: merged/closed, or merged_at/closed_at)"
+            )
+        if not _MERGE_COMMIT_SHA_RE.search(text):
+            reasons.append(
+                "post-merge moot validation claimed without a merge commit SHA"
+            )
+        if reasons:
+            return {
+                "proven": False,
+                "block": True,
+                "outcome": POST_MERGE_MOOT_OUTCOME,
+                "process_state_label": LABEL_POST_MERGE_MOOT,
+                "reasons": reasons,
+                "skipped": False,
+                "safe_next_action": (
+                    "prove the PR is merged/closed: cite PR state and the merge "
+                    "commit SHA. Template:\n" + POST_MERGE_VALIDATION_TEMPLATE
+                ),
+            }
+        return {
+            "proven": True,
+            "block": False,
+            "outcome": POST_MERGE_MOOT_OUTCOME,
+            "process_state_label": LABEL_POST_MERGE_MOOT,
+            "reasons": [],
+            "skipped": False,
+            "safe_next_action": "",
+        }
+
+    # Merged signal present but no approval claim and no moot wording yet —
+    # guide toward the moot outcome without blocking.
+    return {
+        "proven": True,
+        "block": False,
+        "outcome": POST_MERGE_MOOT_OUTCOME,
+        "process_state_label": LABEL_POST_MERGE_MOOT,
+        "reasons": [],
+        "skipped": False,
+        "safe_next_action": (
+            "PR appears merged/closed; if submitting a verdict, record "
+            "post-merge moot validation rather than an active approval"
+        ),
+    }

reviewer_pr_lease.py

@@ -522,4 +522,241 @@ def assess_lease_inventory(
         "active_review_leases": active,
         "stale_review_leases": stale,
         "reclaimable_review_leases": reclaimable,
-    }
+    }
+
+# Canonical next-action vocabulary for reviewer lease handoff (#599).
+NEXT_ACTION_ACQUIRE = "acquire"
+NEXT_ACTION_WAIT = "wait"
+NEXT_ACTION_RESUME_EXACT_OWNER_SESSION = "resume_exact_owner_session"
+NEXT_ACTION_RELEASE_EXPIRED_LEASE = "release_expired_lease"
+NEXT_ACTION_OPERATOR_AUTHORIZED_CLEANUP = "operator_authorized_cleanup"
+NEXT_ACTION_REPAIR_WORKTREE_BINDING = "repair_worktree_binding"
+
+_HANDOFF_CLASSIFICATIONS = frozenset({
+    "no_lease",
+    "own_active",
+    "own_expired",
+    "foreign_active",
+    "foreign_reclaimable",
+    "foreign_expired",
+    "instructed_lease_missing_with_replacement",
+    "worktree_binding_mismatch",
+})
+
+
+def _norm_path(value: str | None) -> str:
+    text = (value or "").strip()
+    if not text:
+        return ""
+    return os.path.normpath(text.rstrip("/"))
+
+
+def diagnose_reviewer_pr_lease_handoff(
+    comments: list[dict],
+    *,
+    pr_number: int,
+    current_session_id: str | None,
+    current_reviewer_identity: str | None,
+    proposed_worktree: str | None = None,
+    env_bound_worktree: str | None = None,
+    instructed_session_id: str | None = None,
+    instructed_comment_id: int | None = None,
+    now: datetime | None = None,
+) -> dict[str, Any]:
+    """Classify open-PR reviewer lease handoff and emit a canonical next action (#599).
+
+    Read-only diagnosis. Never steals, releases, or adopts a foreign lease.
+    Fail-closed acquisition rules for active foreign leases remain intact.
+
+    Returns a structured diagnosis with:
+    - classification
+    - next_action (one of wait / resume_exact_owner_session /
+      release_expired_lease / operator_authorized_cleanup /
+      repair_worktree_binding / acquire)
+    - active_lease identity fields when present
+    - worktree_binding match result
+    - instructed-lease mismatch flags
+    """
+    now = now or datetime.now(timezone.utc)
+    session_id = (current_session_id or "").strip()
+    identity = (current_reviewer_identity or "").strip()
+    instructed_sid = (instructed_session_id or "").strip()
+    reasons: list[str] = []
+
+    active = find_active_reviewer_lease(comments, pr_number=pr_number, now=now)
+    session_lease = get_session_lease()
+
+    # Worktree binding: env-bound vs proposed vs active lease worktree.
+    env_wt = _norm_path(env_bound_worktree)
+    prop_wt = _norm_path(proposed_worktree)
+    lease_wt = _norm_path((active or {}).get("worktree") if active else None)
+    binding_mismatch = False
+    binding_details: dict[str, Any] = {
+        "env_bound_worktree": env_bound_worktree or None,
+        "proposed_worktree": proposed_worktree or None,
+        "lease_worktree": (active or {}).get("worktree") if active else None,
+        "match": True,
+    }
+    paths = [p for p in (env_wt, prop_wt, lease_wt) if p]
+    if len(paths) >= 2 and len(set(paths)) > 1:
+        binding_mismatch = True
+        binding_details["match"] = False
+        reasons.append(
+            "worktree binding mismatch: env/proposed/lease worktree paths disagree"
+        )
+
+    # Instructed lease gone while a different lease is active (PR #592-style).
+    instructed_missing_with_replacement = False
+    if instructed_sid or instructed_comment_id is not None:
+        if not active:
+            reasons.append(
+                "instructed lease is gone and no active replacement lease remains"
+            )
+        else:
+            owner = (active.get("session_id") or "").strip()
+            cid = active.get("comment_id")
+            sid_mismatch = bool(instructed_sid and owner and owner != instructed_sid)
+            cid_mismatch = (
+                instructed_comment_id is not None
+                and cid is not None
+                and int(cid) != int(instructed_comment_id)
+            )
+            if sid_mismatch or cid_mismatch:
+                instructed_missing_with_replacement = True
+                reasons.append(
+                    "instructed lease is gone; a different active lease replaced it "
+                    f"(active session_id={owner}, comment_id={cid})"
+                )
+
+    # Classification + next_action.
+    classification = "no_lease"
+    next_action = NEXT_ACTION_ACQUIRE
+
+    if active:
+        owner = (active.get("session_id") or "").strip()
+        freshness = active.get("freshness") or classify_lease_freshness(
+            active, now=now
+        )
+        owner_identity = (active.get("reviewer_identity") or "").strip()
+        is_own = bool(session_id and owner and owner == session_id)
+        # Same identity alone is NOT ownership for resume; session_id must match.
+        same_identity = bool(
+            identity and owner_identity and identity == owner_identity
+        )
+
+        if is_own and freshness in {"active", "stale_warning"}:
+            classification = "own_active"
+            next_action = NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
+        elif is_own and freshness in {"reclaimable", "expired"}:
+            classification = "own_expired"
+            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
+            reasons.append(
+                f"own lease freshness is '{freshness}'; release via "
+                "gitea_release_reviewer_pr_lease then re-acquire"
+            )
+        elif not is_own and freshness in {"active", "stale_warning"}:
+            classification = "foreign_active"
+            next_action = NEXT_ACTION_WAIT
+            reasons.append(
+                f"foreign active reviewer lease (session_id={owner}, "
+                f"phase={active.get('phase')}, freshness={freshness}); "
+                "do not submit; do not steal"
+            )
+            if same_identity:
+                reasons.append(
+                    "lease identity matches current reviewer but session_id differs; "
+                    "resume only from the exact owner session_id or wait"
+                )
+        elif not is_own and freshness == "reclaimable":
+            classification = "foreign_reclaimable"
+            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
+            reasons.append(
+                f"foreign reclaimable lease (session_id={owner}); clear only via "
+                "sanctioned gitea_release_reviewer_pr_lease when reclaimable"
+            )
+        elif not is_own and freshness == "expired":
+            classification = "foreign_expired"
+            next_action = NEXT_ACTION_RELEASE_EXPIRED_LEASE
+            reasons.append(
+                f"foreign expired lease (session_id={owner}); use sanctioned release"
+            )
+        else:
+            classification = "foreign_active"
+            next_action = NEXT_ACTION_WAIT
+            reasons.append(
+                f"unclassified active lease state (session_id={owner}, "
+                f"freshness={freshness}); wait fail-closed"
+            )
+
+        if instructed_missing_with_replacement and classification.startswith(
+            "foreign"
+        ):
+            classification = "instructed_lease_missing_with_replacement"
+            # Foreign active still means wait; reclaimable still release.
+            if next_action == NEXT_ACTION_WAIT:
+                reasons.append(
+                    "replacement foreign lease is active — wait; "
+                    "operator_authorized_cleanup only with explicit operator authority"
+                )
+    else:
+        classification = "no_lease"
+        next_action = NEXT_ACTION_ACQUIRE
+        reasons.append("no active reviewer lease; acquire via gitea_acquire_reviewer_pr_lease")
+
+    # Binding mismatch is a first-class blocker before submit, but does not
+    # erase foreign-lease wait/release guidance. Override next_action only when
+    # the session would otherwise be free to acquire or resume (submit path).
+    if binding_mismatch:
+        if next_action in {
+            NEXT_ACTION_ACQUIRE,
+            NEXT_ACTION_RESUME_EXACT_OWNER_SESSION,
+        }:
+            classification = "worktree_binding_mismatch"
+            next_action = NEXT_ACTION_REPAIR_WORKTREE_BINDING
+        else:
+            reasons.append(
+                "also repair worktree binding before submit "
+                f"(next_action remains {next_action})"
+            )
+
+    lease_summary = None
+    if active:
+        lease_summary = {
+            "comment_id": active.get("comment_id"),
+            "session_id": active.get("session_id"),
+            "phase": active.get("phase"),
+            "candidate_head": active.get("candidate_head"),
+            "expires_at": active.get("expires_at"),
+            "last_activity": active.get("last_activity"),
+            "freshness": active.get("freshness")
+            or classify_lease_freshness(active, now=now),
+            "reviewer_identity": active.get("reviewer_identity"),
+            "profile": active.get("profile"),
+            "worktree": active.get("worktree"),
+            "blocker": active.get("blocker"),
+        }
+
+    return {
+        "pr_number": pr_number,
+        "classification": classification,
+        "next_action": next_action,
+        "active_lease": lease_summary,
+        "session_lease": session_lease,
+        "worktree_binding": binding_details,
+        "instructed_session_id": instructed_session_id,
+        "instructed_comment_id": instructed_comment_id,
+        "instructed_lease_missing_with_replacement": instructed_missing_with_replacement,
+        "mutation_allowed": (
+            next_action == NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
+            and not binding_mismatch
+            and bool(session_lease)
+        ),
+        "reasons": reasons,
+        "forbidden": [
+            "manual lock deletion",
+            "raw API bypass",
+            "mtime manipulation",
+            "direct _SESSION_LEASE seeding",
+            "silent foreign lease steal",
+        ],
+    }

role_session_router.py

@@ -55,7 +55,6 @@ def skip_python_scan_walk_root(project_root: str, walk_root: str) -> bool:
 
 REVIEWER_TASKS = frozenset({
     "review_pr",
-    "merge_pr",
     "blind_pr_queue_review",
     "pr_queue_cleanup",
     "pr-queue-cleanup",
@@ -63,6 +62,10 @@ REVIEWER_TASKS = frozenset({
     "approve_pr",
 })
 
+MERGER_TASKS = frozenset({
+    "merge_pr",
+})
+
 AUTHOR_TASKS = frozenset({
     "create_issue",
     "comment_issue",
@@ -98,7 +101,7 @@ TASK_REQUIRED_ROLE = {
     "address_pr_change_requests": "author",
     "delete_branch": "author",
     "review_pr": "reviewer",
-    "merge_pr": "reviewer",
+    "merge_pr": "merger",
     "blind_pr_queue_review": "reviewer",
     "pr_queue_cleanup": "reviewer",
     "pr-queue-cleanup": "reviewer",
@@ -128,6 +131,10 @@ WRONG_ROLE_RECONCILER_MSG = (
     "MCP namespace/profile with exact close capability."
 )
 
+WRONG_ROLE_MERGER_MSG = (
+    "Wrong role/session for merger task. Launch merger MCP namespace."
+)
+
 _session_last_route: dict | None = None
 
 
@@ -243,6 +250,25 @@ def route_task_session(
         _record_route(result)
         return result
 
+    if required_role == "merger":
+        result = {
+            "task_type": task_type,
+            "required_role": required_role,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": ROUTE_WRONG_ROLE,
+            "downstream_allowed": False,
+            "reasons": [
+                WRONG_ROLE_MERGER_MSG,
+                "Merger tasks cannot run in author or reviewer sessions.",
+            ],
+            "message": WRONG_ROLE_MERGER_MSG,
+            "runtime_switching_supported": runtime_switching_supported,
+            "profile_switch_blocked": not runtime_switching_supported,
+        }
+        _record_route(result)
+        return result
+
     if required_role == "author":
         route = ROUTE_TO_AUTHOR
         message = (

skills/llm-project-workflow/workflows/review-merge-pr.md

@@ -826,6 +826,75 @@ The final report must identify:
 * whether same-PR merge continuation was allowed
 * whether the run stopped as required
 
+## 26A-1. Stale durable review-decision lock cleanup (#594)
+
+After #559, the review-decision lock is durable under
+`~/.cache/gitea-tools/session-state/` (for example
+`review_decision_lock-prgs-reviewer.json`). That durability can outlive the work
+it protects: a terminal `approve` / `request_changes` on a PR that is already
+**merged or closed** still hard-stops later unrelated reviews under #332.
+
+**Do not** delete session-state files by hand. Use the canonical MCP path.
+
+### When cleanup is allowed
+
+Use `gitea_cleanup_stale_review_decision_lock` when:
+
+1. `gitea_mark_final_review_decision` / live review is blocked by
+   `terminal review mutation already consumed ... (#332)`.
+2. Live Gitea state for the **last terminal mutation's PR** is **merged** or
+   **closed** (no same-PR merge sequence remains).
+3. Active profile identity matches the durable lock (reviewer profile with
+   `gitea.pr.review` for `apply=true`).
+4. Optional pin: pass `expected_terminal_pr` to the prior terminal PR number
+   (for example `586` when resuming after a merged #586 lock).
+
+Recommended sequence:
+
+```text
+gitea_whoami
+gitea_resolve_task_capability(task="cleanup_stale_review_decision_lock")
+gitea_cleanup_stale_review_decision_lock(apply=false, remote=..., org=..., repo=...)
+# inspect is_moot / cleanup_allowed / last_terminal_pr
+gitea_cleanup_stale_review_decision_lock(
+    apply=true,
+    expected_terminal_pr=<last_terminal_pr>,
+    remote=..., org=..., repo=...,
+)
+gitea_resolve_task_capability(task="review_pr")
+gitea_load_review_workflow()
+# continue formal mark + submit for the *new* PR
+```
+
+Successful `apply=true` clears memory + durable store and returns an audit
+record (and posts a durable PR comment on the terminal PR when
+`post_audit_comment` is true and comment permission allows).
+
+Same-profile auto-expire: after `gitea_merge_pr` succeeds for the PR that this
+profile just approved, the decision lock for that profile is cleared
+automatically. Cross-profile locks (for example reviewer approve, merger merge)
+still require `gitea_cleanup_stale_review_decision_lock`.
+
+### When cleanup is forbidden
+
+Refuse / fail-closed — keep #332 hard-stop — when:
+
+* the last terminal PR is still **open** (active review or merge sequence may
+  still apply)
+* live PR state cannot be fetched (ambiguous)
+* no lock or no terminal live mutation exists
+* profile identity does not match the lock
+* apply requested without authenticated identity or `gitea.pr.review`
+* `expected_terminal_pr` does not match the last terminal PR
+
+Do **not** use cleanup to:
+
+* bypass #332 on an open PR
+* replace `gitea_authorize_review_correction` for a mistaken review on a still
+  active PR (#211)
+* auto-approve or auto-merge any PR
+* justify `rm` of session-state files
+
 ## 26B. Per-PR reviewer lease (#407)
 
 Parallel reviewer sessions are allowed only when each session holds a distinct,
@@ -1043,6 +1112,8 @@ Blocked handoffs must say to rerun the full workflow after the blocker clears.
 
 If the blocker is a terminal review mutation already consumed in the current run, the handoff must say that the next run must start from the beginning and must not reuse stale ready/approved state.
 
+If the referenced terminal PR is already **merged or closed** on live Gitea, the durable #332 decision lock is **moot**. Do **not** delete session-state files by hand. Use `gitea_cleanup_stale_review_decision_lock` (assess with `apply=false`, then `apply=true` with `expected_terminal_pr` set) from the reviewer profile after identity/profile checks (#594). Cleanup is **forbidden** while that PR is still open.
+
 ## 30. Final report must be precise
 
 Include:

stale_review_decision_lock.py

@@ -0,0 +1,252 @@
+"""Stale / moot #332 review-decision lock detection and cleanup policy (#594).
+
+#332 correctly hard-stops a reviewer session after a terminal live review
+mutation. After #559 those locks are durable on disk, so they can outlive the
+work they protect: when the referenced PR is already merged or closed, no
+same-PR merge sequence remains, yet the durable ledger still blocks unrelated
+new terminal reviews.
+
+This module is pure policy (no Gitea I/O). The MCP tool
+``gitea_cleanup_stale_review_decision_lock`` fetches live PR state, calls these
+helpers, and only then clears durable state when cleanup is allowed.
+
+Manual deletion of ``~/.cache/gitea-tools/session-state/*`` is never the
+canonical path.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from typing import Any
+
+# Keep in sync with gitea_mcp_server._TERMINAL_REVIEW_ACTIONS.
+TERMINAL_REVIEW_ACTIONS = frozenset({"approve", "request_changes"})
+
+
+def last_terminal_mutation(lock: dict | None) -> dict | None:
+    """Return the last terminal live mutation on *lock*, or None."""
+    if not lock:
+        return None
+    terminals = [
+        m
+        for m in (lock.get("live_mutations") or [])
+        if isinstance(m, dict) and m.get("action") in TERMINAL_REVIEW_ACTIONS
+    ]
+    return terminals[-1] if terminals else None
+
+
+def classify_pr_live_state(pr_live: dict | None) -> dict[str, Any]:
+    """Normalize a Gitea PR payload into merge/closed flags."""
+    pr = pr_live or {}
+    merged = bool(pr.get("merged") or pr.get("merged_at"))
+    state = (pr.get("state") or "").strip().lower() or None
+    closed = state == "closed"
+    return {
+        "pr_state": state,
+        "pr_merged": merged,
+        "pr_merged_or_closed": merged or closed,
+        "merge_commit_sha": pr.get("merge_commit_sha")
+        or pr.get("merged_commit_sha"),
+    }
+
+
+def lock_summary(lock: dict | None) -> dict[str, Any] | None:
+    """LLM-safe summary of a decision lock (no secrets)."""
+    if lock is None:
+        return None
+    last = last_terminal_mutation(lock)
+    mutations = list(lock.get("live_mutations") or [])
+    return {
+        "task": lock.get("task"),
+        "remote": lock.get("remote"),
+        "org": lock.get("org") or lock.get("ready_org"),
+        "repo": lock.get("repo") or lock.get("ready_repo"),
+        "profile_identity": lock.get("profile_identity")
+        or lock.get("session_profile_lock")
+        or lock.get("session_profile"),
+        "session_profile": lock.get("session_profile"),
+        "ready_pr_number": lock.get("ready_pr_number"),
+        "ready_action": lock.get("ready_action"),
+        "live_mutations_count": len(mutations),
+        "last_terminal": (
+            {
+                "pr_number": last.get("pr_number"),
+                "action": last.get("action"),
+                "review_id": last.get("review_id"),
+            }
+            if last
+            else None
+        ),
+        "correction_authorized": bool(lock.get("correction_authorized")),
+        "updated_at": lock.get("updated_at") or lock.get("recorded_at"),
+    }
+
+
+def assess_stale_review_decision_lock(
+    lock: dict | None,
+    *,
+    pr_live: dict | None = None,
+    pr_lookup_error: str | None = None,
+    active_profile_identity: str | None = None,
+) -> dict[str, Any]:
+    """Assess whether a durable review-decision lock is stale/moot (#594).
+
+    *pr_live* must be the live Gitea payload for the **last terminal
+    mutation's PR**. When no lock / no terminal mutation exists, cleanup is
+    not applicable. When the PR is still open (or lookup fails), cleanup is
+    forbidden and #332 hard-stop remains in force.
+    """
+    summary = lock_summary(lock)
+    result: dict[str, Any] = {
+        "has_lock": lock is not None,
+        "lock_summary": summary,
+        "last_terminal_pr": None,
+        "last_terminal_action": None,
+        "is_moot": False,
+        "cleanup_allowed": False,
+        "profile_match": True,
+        "pr_state": None,
+        "pr_merged": False,
+        "pr_merged_or_closed": False,
+        "merge_commit_sha": None,
+        "reasons": [],
+        "recovery_tool": "gitea_cleanup_stale_review_decision_lock",
+    }
+
+    if lock is None:
+        result["reasons"].append("no review decision lock present")
+        return result
+
+    # Profile identity gate (caller may re-check with env; pure assess still
+    # reports mismatch when active identity is supplied).
+    stored = (
+        (lock.get("profile_identity") or lock.get("session_profile_lock") or "")
+        .strip()
+    )
+    active = (active_profile_identity or "").strip()
+    if active and stored and active != stored:
+        result["profile_match"] = False
+        result["reasons"].append(
+            "review decision lock profile identity does not match active "
+            f"profile '{active}' (fail closed, #594)"
+        )
+        return result
+
+    last = last_terminal_mutation(lock)
+    if last is None:
+        result["reasons"].append(
+            "review decision lock has no terminal live mutations; nothing to clear"
+        )
+        return result
+
+    pr_number = last.get("pr_number")
+    action = last.get("action")
+    result["last_terminal_pr"] = pr_number
+    result["last_terminal_action"] = action
+
+    if pr_number is None:
+        result["reasons"].append(
+            "last terminal mutation missing pr_number; refuse cleanup (fail closed)"
+        )
+        return result
+
+    if pr_lookup_error:
+        result["reasons"].append(
+            f"could not load live state for PR #{pr_number}: {pr_lookup_error} "
+            "(fail closed, #594)"
+        )
+        return result
+
+    if pr_live is None:
+        result["reasons"].append(
+            f"live PR state required for terminal PR #{pr_number} before cleanup "
+            "(fail closed, #594)"
+        )
+        return result
+
+    live = classify_pr_live_state(pr_live)
+    result.update(
+        {
+            "pr_state": live["pr_state"],
+            "pr_merged": live["pr_merged"],
+            "pr_merged_or_closed": live["pr_merged_or_closed"],
+            "merge_commit_sha": live["merge_commit_sha"],
+        }
+    )
+
+    if not live["pr_merged_or_closed"]:
+        result["reasons"].append(
+            f"terminal review mutation on open PR #{pr_number} is still active "
+            f"({action}); #332 hard-stop remains — cleanup forbidden (#594)"
+        )
+        return result
+
+    # Merged or closed: lock is moot. Same-PR merge continuation is impossible.
+    result["is_moot"] = True
+    result["cleanup_allowed"] = True
+    result["reasons"].append(
+        f"terminal mutation ({action} on PR #{pr_number}) is moot: PR is "
+        f"{'merged' if live['pr_merged'] else 'closed'}; canonical cleanup allowed (#594)"
+    )
+    return result
+
+
+def build_cleanup_audit_record(
+    *,
+    assessment: dict[str, Any],
+    actor_username: str | None,
+    profile_name: str | None,
+    applied: bool,
+) -> dict[str, Any]:
+    """Structured durable audit payload for a cleanup attempt."""
+    last = (assessment.get("lock_summary") or {}).get("last_terminal") or {}
+    return {
+        "event": "stale_review_decision_lock_cleanup",
+        "issue_ref": "#594",
+        "applied": bool(applied),
+        "timestamp": datetime.now(timezone.utc).isoformat(),
+        "actor_username": actor_username,
+        "profile_name": profile_name,
+        "last_terminal_pr": assessment.get("last_terminal_pr") or last.get("pr_number"),
+        "last_terminal_action": assessment.get("last_terminal_action")
+        or last.get("action"),
+        "pr_state": assessment.get("pr_state"),
+        "pr_merged": assessment.get("pr_merged"),
+        "pr_merged_or_closed": assessment.get("pr_merged_or_closed"),
+        "merge_commit_sha": assessment.get("merge_commit_sha"),
+        "prior_live_mutations_count": (
+            (assessment.get("lock_summary") or {}).get("live_mutations_count")
+        ),
+        "prior_profile_identity": (
+            (assessment.get("lock_summary") or {}).get("profile_identity")
+        ),
+        "cleanup_allowed": assessment.get("cleanup_allowed"),
+        "is_moot": assessment.get("is_moot"),
+        "reasons": list(assessment.get("reasons") or []),
+    }
+
+
+def format_cleanup_audit_comment(audit: dict[str, Any]) -> str:
+    """Markdown body for a durable Gitea audit comment after cleanup."""
+    status = "APPLIED" if audit.get("applied") else "ASSESSED (not applied)"
+    lines = [
+        "## Stale #332 review-decision lock cleanup (#594)",
+        "",
+        f"Status: **{status}**",
+        "",
+        f"- actor: `{audit.get('actor_username')}`",
+        f"- profile: `{audit.get('profile_name')}`",
+        f"- timestamp: `{audit.get('timestamp')}`",
+        f"- last terminal: `{audit.get('last_terminal_action')}` on "
+        f"PR #{audit.get('last_terminal_pr')}",
+        f"- PR state: `{audit.get('pr_state')}` "
+        f"(merged={audit.get('pr_merged')})",
+        f"- merge_commit_sha: `{audit.get('merge_commit_sha')}`",
+        f"- prior live_mutations_count: "
+        f"`{audit.get('prior_live_mutations_count')}`",
+        f"- prior profile_identity: `{audit.get('prior_profile_identity')}`",
+        "",
+        "Manual deletion of session-state files is **not** the workflow.",
+        "This path only clears a lock when the referenced PR is merged/closed.",
+    ]
+    return "\n".join(lines)

task_capability_map.py

@@ -96,6 +96,17 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.pr.approve",
         "role": "reviewer",
     },
+    # #594: clear durable #332 decision lock only when last terminal PR is
+    # already merged/closed (moot). Apply path requires reviewer review
+    # permission; assessment itself uses gitea.read inside the tool.
+    "cleanup_stale_review_decision_lock": {
+        "permission": "gitea.pr.review",
+        "role": "reviewer",
+    },
+    "gitea_cleanup_stale_review_decision_lock": {
+        "permission": "gitea.pr.review",
+        "role": "reviewer",
+    },
     "delete_branch": {
         "permission": "gitea.branch.delete",
         "role": "author",

tests/conftest.py

@@ -18,13 +18,13 @@ def _reset_mutation_authority(monkeypatch):
     no-op: individual tests that need a specific authority state set it up
     explicitly.
 
-    Session-state isolation (#559 / #590): many tests use
+    Session-state isolation (#559 / #590 / #594): many tests use
     ``patch.dict(os.environ, ..., clear=True)``, which drops
     ``GITEA_MCP_SESSION_STATE_DIR``. Relying only on the env var therefore
     re-exposes the operator host cache
     (``~/.cache/gitea-tools/session-state``) and its review-mutation ledger.
     Pin ``default_state_dir`` / ``DEFAULT_STATE_DIR`` to a per-test temp dir
-    so durable load/save never touches host state, even after env clears.
+    so durable load/save never touches host state even after env clears.
     """
     monkeypatch.delenv("GITEA_SESSION_PROFILE_LOCK", raising=False)
     # Isolate durable session-state files so tests never share host cache (#559).

tests/test_controller_closure_baseline_proof.py

@@ -0,0 +1,113 @@
+"""Controller-closure baseline-proof gate tests (#529, criterion 6)."""
+
+import unittest
+
+from controller_closure_baseline_proof import (
+    assess_controller_closure_baseline_proof,
+)
+from premerge_baseline_proof import (
+    CLEAN_PASS,
+    PREMERGE_BASELINE_PROVEN_FAILURE,
+    UNRESOLVED_REGRESSION_RISK,
+)
+from final_report_validator import assess_final_report_validator
+
+# Complete pre-merge proof fields for a baseline claim (see #533).
+_PROOF_FIELDS = (
+    "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+    "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+    "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
+    "Exit status: 1\n"
+    "Failure signature: AssertionError: None is not true\n"
+)
+
+
+class TestControllerClosureBaselineProof(unittest.TestCase):
+    def test_empty_report_skipped_and_allowed(self):
+        result = assess_controller_closure_baseline_proof("")
+        self.assertFalse(result["block"])
+        self.assertTrue(result["skipped"])
+        self.assertEqual(result["label"], CLEAN_PASS)
+
+    def test_none_report_allowed(self):
+        result = assess_controller_closure_baseline_proof(None)
+        self.assertFalse(result["block"])
+        self.assertTrue(result["skipped"])
+
+    def test_clean_pass_closure_allowed(self):
+        result = assess_controller_closure_baseline_proof(
+            "Closing #529. Validation: full suite passed, exit status 0. Clean pass."
+        )
+        self.assertFalse(result["block"])
+        self.assertEqual(result["label"], CLEAN_PASS)
+
+    def test_expected_preexisting_failure_without_proof_blocked(self):
+        report = (
+            "Closing #529. Full suite: 1 failed. This is an expected "
+            "pre-existing failure, safe to close."
+        )
+        result = assess_controller_closure_baseline_proof(report)
+        self.assertTrue(result["block"])
+        self.assertFalse(result["proven"])
+        self.assertEqual(result["label"], UNRESOLVED_REGRESSION_RISK)
+        self.assertTrue(result["reasons"])
+        self.assertTrue(result["safe_next_action"])
+
+    def test_current_master_reproduction_only_blocked(self):
+        report = (
+            "Closing #529. Full suite: 1 failed. Pre-existing baseline failure — "
+            "reproduced on current master after merge.\n"
+            "Command: venv/bin/python -m pytest tests/test_foo.py -q\n"
+            "Exit status: 1\n"
+            "Failure signature: AssertionError: None is not true\n"
+        )
+        result = assess_controller_closure_baseline_proof(report)
+        self.assertTrue(result["block"])
+
+    def test_proven_baseline_closure_allowed(self):
+        report = (
+            "Closing #529. Full suite: 1 failed, an expected pre-existing "
+            "baseline failure proven on the PR pre-merge base commit.\n"
+            + _PROOF_FIELDS
+        )
+        result = assess_controller_closure_baseline_proof(report)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["label"], PREMERGE_BASELINE_PROVEN_FAILURE)
+
+
+class TestControllerCloseTaskKind(unittest.TestCase):
+    """The composable validator must cover the controller_close task kind."""
+
+    def test_controller_close_blocks_unproven_baseline(self):
+        report = (
+            "Full suite: 1 failed. Expected pre-existing failure, closing the "
+            "tracking issue."
+        )
+        result = assess_final_report_validator(report, "controller_close")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                "premerge_baseline_proof" in f["rule_id"]
+                for f in result["findings"]
+            )
+        )
+
+    def test_controller_close_alias_close_issue(self):
+        report = (
+            "Full suite: 1 failed. Expected pre-existing failure, closing the "
+            "tracking issue."
+        )
+        result = assess_final_report_validator(report, "close_issue")
+        self.assertTrue(result["blocked"])
+
+    def test_controller_close_allows_proven_baseline(self):
+        report = (
+            "Full suite: 1 failed, expected pre-existing baseline failure proven "
+            "on the PR pre-merge base commit.\n" + _PROOF_FIELDS
+        )
+        result = assess_final_report_validator(report, "controller_close")
+        self.assertFalse(result["blocked"])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_master_parity_gate.py

@@ -78,6 +78,95 @@ class TestBlockReasonsAndReport(unittest.TestCase):
         self.assertTrue(report["recovery"])
 
 
+class TestLiveRemoteParity(unittest.TestCase):
+    """#610: parity must account for the live remote master, not just local.
+
+    The daemon can be stale relative to the live remote target while the local
+    checkout HEAD still matches the daemon's startup commit, so local parity
+    reports green even though a mutation would run against outdated code.
+    """
+
+    SHA_C = "c" * 40
+
+    def test_distinguishes_three_shas(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        self.assertEqual(res["daemon_start_head"], SHA_A)
+        self.assertEqual(res["local_head"], SHA_A)
+        self.assertEqual(res["live_remote_head"], SHA_B)
+
+    def test_mutation_safe_only_when_all_three_match(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_A)
+        self.assertTrue(res["mutation_safe"])
+        self.assertTrue(res["live_known"])
+        self.assertFalse(res["live_stale"])
+
+    def test_live_stale_when_remote_advanced_past_daemon(self):
+        # Local checkout still matches the daemon start (local parity green),
+        # but the live remote master has advanced -> daemon is live-stale.
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        self.assertTrue(res["in_parity"])  # local parity still green
+        self.assertTrue(res["live_stale"])
+        self.assertFalse(res["mutation_safe"])
+        self.assertTrue(any("live" in r.lower() for r in res["reasons"]))
+
+    def test_live_unknown_is_not_mutation_safe_but_not_stale(self):
+        # Non-goal: unfetchable live remote must not be treated as stale for
+        # read-only, but a mutation-safe claim fails closed.
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=None)
+        self.assertFalse(res["live_known"])
+        self.assertFalse(res["mutation_safe"])
+        self.assertFalse(res["live_stale"])
+        self.assertTrue(res["in_parity"])
+
+    def test_default_live_remote_preserves_legacy_shape(self):
+        # Callers that do not supply a live head keep the pre-#610 behavior:
+        # in-parity, not live-stale, no live-derived block.
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
+        self.assertFalse(res["live_stale"])
+        self.assertEqual(mp.parity_block_reasons(res), [])
+
+
+class TestLiveStaleBlockAndReport(unittest.TestCase):
+    """#610: live-staleness must block mutations and surface a typed blocker."""
+
+    def test_live_stale_produces_block_reasons(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        self.assertTrue(mp.parity_block_reasons(res))
+
+    def test_disable_env_suppresses_live_stale_block(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        with patch.dict(os.environ, {mp.ENV_DISABLE: "1"}):
+            self.assertEqual(mp.parity_block_reasons(res), [])
+
+    def test_resolver_disagreement_returns_typed_blocker(self):
+        # Parity says local-green, resolver says restart required -> disagreement
+        # is a typed, fail-closed blocker naming the resolver as authoritative.
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
+        blocker = mp.parity_resolver_disagreement(res, resolver_restart_required=True)
+        self.assertIsNotNone(blocker)
+        self.assertEqual(blocker["kind"], "parity_resolver_disagreement")
+        self.assertTrue(blocker["restart_required"])
+        self.assertTrue(blocker["resolver_authoritative"])
+
+    def test_no_disagreement_when_resolver_agrees(self):
+        res = mp.assess_master_parity({"startup_head": SHA_A}, SHA_A)
+        self.assertIsNone(
+            mp.parity_resolver_disagreement(res, resolver_restart_required=False))
+
+    def test_live_stale_report_names_live_remote(self):
+        res = mp.assess_master_parity(
+            {"startup_head": SHA_A}, SHA_A, live_remote_head=SHA_B)
+        report = mp.parity_report(res)
+        self.assertEqual(report["live_remote_head"], SHA_B)
+        self.assertTrue(report["restart_required"])
+
+
 class TestReadGitHead(unittest.TestCase):
     def test_test_override_takes_precedence(self):
         with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_B}):
@@ -95,6 +184,78 @@ class TestReadGitHead(unittest.TestCase):
             self.assertIsNone(mp.read_git_head(""))
 
 
+class TestReadRemoteMasterHead(unittest.TestCase):
+    """#610: live remote master head reader (env-overridable, fails to None)."""
+
+    def test_test_override_takes_precedence(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
+            self.assertEqual(mp.read_remote_master_head("/nonexistent"), SHA_B)
+
+    def test_blank_override_is_none(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: "   "}):
+            self.assertIsNone(mp.read_remote_master_head("/nonexistent"))
+
+    def test_unfetchable_remote_is_none(self):
+        # No override; a bogus root/remote must fail closed to None, never raise.
+        env = {k: v for k, v in os.environ.items()
+               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
+        with patch.dict(os.environ, env, clear=True):
+            self.assertIsNone(
+                mp.read_remote_master_head("/nonexistent", remote="nope"))
+
+
+class TestRemoteHeadCache(unittest.TestCase):
+    """#610: live remote reads are cached with a TTL to stay off the network.
+
+    The parity gate runs on every mutation and every runtime-context read, so an
+    unbounded ``git ls-remote`` per call would be a latency/flakiness regression.
+    """
+
+    def setUp(self):
+        mp._clear_remote_head_cache()
+        env = {k: v for k, v in os.environ.items()
+               if k != mp.ENV_TEST_LIVE_REMOTE_HEAD}
+        self._env = patch.dict(os.environ, env, clear=True)
+        self._env.start()
+        self.addCleanup(self._env.stop)
+        self.addCleanup(mp._clear_remote_head_cache)
+
+    def _fake_run(self, sha):
+        class _R:
+            returncode = 0
+            stdout = f"{sha}\trefs/heads/master\n"
+        calls = {"n": 0}
+
+        def run(*args, **kwargs):
+            calls["n"] += 1
+            return _R()
+        return run, calls
+
+    def test_second_call_within_ttl_uses_cache(self):
+        run, calls = self._fake_run(SHA_B)
+        with patch.object(mp.subprocess, "run", run):
+            a = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
+            b = mp.read_remote_master_head("/repo", remote="prgs", ttl=100)
+        self.assertEqual(a, SHA_B)
+        self.assertEqual(b, SHA_B)
+        self.assertEqual(calls["n"], 1)
+
+    def test_zero_ttl_bypasses_cache(self):
+        run, calls = self._fake_run(SHA_B)
+        with patch.object(mp.subprocess, "run", run):
+            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
+            mp.read_remote_master_head("/repo", remote="prgs", ttl=0)
+        self.assertEqual(calls["n"], 2)
+
+    def test_env_override_never_touches_subprocess(self):
+        run, calls = self._fake_run(SHA_B)
+        with patch.dict(os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
+            with patch.object(mp.subprocess, "run", run):
+                self.assertEqual(
+                    mp.read_remote_master_head("/repo", remote="prgs"), SHA_A)
+        self.assertEqual(calls["n"], 0)
+
+
 class TestServerWiring(unittest.TestCase):
     """Integration with the gate choke point in the server namespace."""
 
@@ -105,6 +266,13 @@ class TestServerWiring(unittest.TestCase):
         self._saved = self.srv._STARTUP_PARITY
         self.srv._STARTUP_PARITY = {"root": self.srv.PROJECT_ROOT,
                                     "startup_head": SHA_A}
+        # Keep the live-remote read hermetic (no real ls-remote network call):
+        # default the live master to the daemon start so parity is fully green
+        # unless a test overrides the live head explicitly (#610).
+        self._live_patch = patch.dict(
+            os.environ, {mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A})
+        self._live_patch.start()
+        self.addCleanup(self._live_patch.stop)
 
     def tearDown(self):
         self.srv._STARTUP_PARITY = self._saved
@@ -147,6 +315,36 @@ class TestServerWiring(unittest.TestCase):
             self.assertTrue(out["in_parity"])
             self.assertNotIn("report", out)
 
+    # --- #610: live-remote wiring -------------------------------------------
+
+    def test_live_stale_blocks_mutation_though_local_green(self):
+        # Local checkout matches the daemon start (local parity green) but the
+        # live remote master has advanced -> mutations must fail closed.
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
+                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
+            self.assertEqual(self.srv._master_parity_block("gitea.read"), [])
+            self.assertTrue(
+                self.srv._master_parity_block("gitea.pr.create"))
+
+    def test_assess_tool_exposes_three_distinct_shas(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
+                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_B}):
+            out = self.srv.gitea_assess_master_parity(remote="prgs")
+            self.assertEqual(out["daemon_start_head"], SHA_A)
+            self.assertEqual(out["local_head"], SHA_A)
+            self.assertEqual(out["live_remote_head"], SHA_B)
+            self.assertTrue(out["live_stale"])
+            self.assertFalse(out["mutation_safe"])
+            self.assertIn("report", out)
+
+    def test_assess_tool_mutation_safe_when_all_three_match(self):
+        with patch.dict(os.environ, {mp.ENV_TEST_CURRENT_HEAD: SHA_A,
+                                     mp.ENV_TEST_LIVE_REMOTE_HEAD: SHA_A}):
+            out = self.srv.gitea_assess_master_parity(remote="prgs")
+            self.assertTrue(out["mutation_safe"])
+            self.assertFalse(out["live_stale"])
+            self.assertNotIn("report", out)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_mcp_server.py

@@ -774,8 +774,8 @@ class TestMergePR(unittest.TestCase):
     def setUp(self):
         import reviewer_pr_lease
 
-        # Start each merge test with a clean review-mutation ledger (#590).
-        # Host durable state can otherwise leak in when tests clear os.environ.
+        # Clean ledger each merge test so residual/host durable #332 state
+        # cannot short-circuit eligibility gates (#590 / #594).
         mcp_server._save_review_decision_lock(None)
         mcp_server.gitea_load_review_workflow()
         self._lease_patch = _install_owned_reviewer_lease(8)
@@ -2733,6 +2733,56 @@ class TestTrackerHygieneCleanup(unittest.TestCase):
         self.assertTrue(res["success"])
         self.assertEqual(res["cleanup_status"].get(1), "not present")
 
+    def test_close_issue_blocks_unproven_expected_preexisting_failure(self):
+        # #529: a closure report calling a non-zero suite exit an "expected
+        # pre-existing failure" without pre-merge proof must fail closed and
+        # never PATCH the issue state.
+        patched = []
+
+        def api_side_effect(method, url, auth, payload=None):
+            if method == "PATCH":
+                patched.append(url)
+            return {}
+        self.mock_api.side_effect = api_side_effect
+
+        res = gitea_close_issue(
+            issue_number=1,
+            closure_report=(
+                "Full suite: 1 failed. Expected pre-existing failure, safe to close."
+            ),
+        )
+        self.assertFalse(res["success"])
+        self.assertTrue(res.get("blocked"))
+        self.assertTrue(res.get("reasons"))
+        self.assertFalse(
+            patched, "must not PATCH issue state when the closure gate blocks"
+        )
+
+    def test_close_issue_allows_proven_baseline_closure(self):
+        def api_side_effect(method, url, auth, payload=None):
+            if method == "PATCH" and "issues/1" in url:
+                return {"state": "closed"}
+            if method == "GET" and "labels" in url and "issues" not in url:
+                return [{"name": "bug", "id": 2}]
+            if method == "GET" and "issues/1" in url:
+                return {"labels": [{"name": "bug"}]}
+            return {}
+        self.mock_api.side_effect = api_side_effect
+
+        res = gitea_close_issue(
+            issue_number=1,
+            closure_report=(
+                "Full suite: 1 failed, expected pre-existing baseline failure "
+                "proven on the PR pre-merge base commit.\n"
+                "Pre-merge base commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+                "Tested commit: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+                "Command: venv/bin/python -m pytest -q\n"
+                "Exit status: 1\n"
+                "Failure signature: AssertionError: None is not true\n"
+            ),
+        )
+        self.assertTrue(res["success"])
+
     def test_merge_pr_with_closes_removes_label(self):
         import reviewer_pr_lease
 

tests/test_mcp_stale_runtime.py

@@ -67,5 +67,66 @@ class TestMcpStaleRuntime(unittest.TestCase):
         # But does NOT contain missing author profile error
         self.assertFalse(any("None of the matching profiles for task" in r for r in reasons_author))
 
+    @patch("subprocess.run")
+    @patch("os.path.getmtime")
+    @patch("os.path.exists")
+    @patch("os.getpid")
+    @patch.dict("os.environ", {"GITEA_FORCE_MCP_RUNTIME_CHECK": "1"})
+    def test_git_head_change_triggers_staleness(self, mock_getpid, mock_exists, mock_getmtime, mock_run):
+        # Set boot SHA to FAKE1 and current to FAKE2
+        gitea_mcp_server._process_boot_head_sha = "FAKE1"
+        mock_getpid.return_value = 12345
+        mock_exists.return_value = True
+
+        # Code time is in the past (fresh process chronologically)
+        code_time = datetime(2026, 7, 8, 11, 0, 0)
+        mock_getmtime.return_value = code_time.timestamp()
+
+        ps_output = (
+            "  PID LSTART COMMAND\n"
+            "12345 Wed Jul  8 12:00:00 2026 /path/to/python mcp_server.py\n"
+        )
+
+        mock_run_ps = MagicMock()
+        mock_run_ps.stdout = ps_output
+
+        mock_run_env = MagicMock()
+        mock_run_env.stdout = "GITEA_MCP_PROFILE=prgs-author"
+
+        mock_run_git = MagicMock()
+        mock_run_git.stdout = "FAKE2" # different SHA
+
+        def side_effect(args, **kwargs):
+            if args[0] == "ps" and "eww" in args:
+                return mock_run_env
+            elif args[0] == "ps":
+                return mock_run_ps
+            elif args[0] == "git" and "rev-parse" in args:
+                return mock_run_git
+            raise ValueError(f"Unexpected: {args}")
+
+        mock_run.side_effect = side_effect
+
+        # Stale even though process started AFTER code mtime, because git HEAD changed
+        reasons = gitea_mcp_server._check_mcp_runtimes_diagnostics("create_issue", ["prgs-author"])
+        self.assertTrue(any("stale-runtime: The active Gitea MCP server process is stale" in r for r in reasons))
+
+    @patch("threading.Thread")
+    @patch("os.utime")
+    @patch("os.path.exists")
+    @patch.dict("os.environ", {"MCP_CONFIG_PATH": "/tmp/mcp_config.json"})
+    def test_auto_restart_trigger_touches_and_spawns(self, mock_exists, mock_utime, mock_thread):
+        mock_exists.return_value = True
+        gitea_mcp_server._restart_triggered = False
+
+        # Ensure we are not skipped in test mode for testing purposes
+        with patch("gitea_mcp_server._preflight_in_test_mode", return_value=False):
+            gitea_mcp_server._trigger_mcp_auto_restart()
+
+        mock_utime.assert_called_once_with("/tmp/mcp_config.json", None)
+        mock_thread.assert_called_once()
+        self.assertTrue(gitea_mcp_server._restart_triggered)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_post_merge_validation.py

@@ -0,0 +1,119 @@
+"""Post-merge moot validation wording + label tests (#529, criteria 1/4/5)."""
+
+import unittest
+
+from post_merge_validation import (
+    BASELINE_ACCEPTED_OUTCOME,
+    BLOCKED_OUTCOME,
+    CLEAN_PASS_OUTCOME,
+    LABEL_BASELINE_ACCEPTED,
+    LABEL_BLOCKED,
+    LABEL_CLEAN_PASS,
+    LABEL_POST_MERGE_MOOT,
+    POST_MERGE_MOOT_OUTCOME,
+    assess_post_merge_validation,
+    process_state_label,
+)
+from final_report_validator import assess_final_report_validator
+from issue_workflow_labels import CANONICAL_LABELS, VALIDATION_LABELS
+
+_MOOT_PROOF = (
+    "Validation status: post-merge moot validation\n"
+    "PR state: merged\n"
+    "Merge commit sha: 1a2b3c4d5e6f7a8b9c0d1e2f3a4b5c6d7e8f9a0b\n"
+    "Validation finding: full suite passed, for the record.\n"
+)
+
+
+class TestPostMergeValidation(unittest.TestCase):
+    def test_not_applicable_when_no_merged_or_moot_signal(self):
+        result = assess_post_merge_validation(
+            "Review decision: approve. Validation: full suite passed."
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["skipped"])
+
+    def test_active_approval_on_merged_pr_blocked(self):
+        report = (
+            "PR is already merged. Review decision: approve. "
+            "Validation: full suite passed."
+        )
+        result = assess_post_merge_validation(report)
+        self.assertTrue(result["block"])
+        self.assertEqual(result["outcome"], POST_MERGE_MOOT_OUTCOME)
+        self.assertEqual(result["process_state_label"], LABEL_POST_MERGE_MOOT)
+
+    def test_active_approval_on_merged_pr_via_structured_signal_blocked(self):
+        report = "Review decision: approve. Validation: full suite passed."
+        result = assess_post_merge_validation(report, pr_merged_or_closed=True)
+        self.assertTrue(result["block"])
+
+    def test_moot_wording_without_proof_blocked(self):
+        report = "Recording post-merge moot validation for this PR."
+        result = assess_post_merge_validation(report)
+        self.assertTrue(result["block"])
+        self.assertTrue(result["reasons"])
+
+    def test_moot_wording_with_full_proof_allowed(self):
+        result = assess_post_merge_validation(_MOOT_PROOF)
+        self.assertFalse(result["block"])
+        self.assertEqual(result["outcome"], POST_MERGE_MOOT_OUTCOME)
+
+    def test_merged_signal_only_guides_without_blocking(self):
+        result = assess_post_merge_validation(
+            "PR was already merged before review; recording findings."
+        )
+        self.assertFalse(result["block"])
+        self.assertTrue(result["safe_next_action"])
+
+    def test_process_state_label_mapping(self):
+        self.assertEqual(process_state_label(CLEAN_PASS_OUTCOME), LABEL_CLEAN_PASS)
+        self.assertEqual(
+            process_state_label(BASELINE_ACCEPTED_OUTCOME), LABEL_BASELINE_ACCEPTED
+        )
+        self.assertEqual(process_state_label(BLOCKED_OUTCOME), LABEL_BLOCKED)
+        self.assertEqual(
+            process_state_label(POST_MERGE_MOOT_OUTCOME), LABEL_POST_MERGE_MOOT
+        )
+        self.assertIsNone(process_state_label("nonsense"))
+
+
+class TestValidationLabelsRegistered(unittest.TestCase):
+    def test_validation_labels_are_canonical(self):
+        for label in (
+            LABEL_CLEAN_PASS,
+            LABEL_BASELINE_ACCEPTED,
+            LABEL_BLOCKED,
+            LABEL_POST_MERGE_MOOT,
+        ):
+            self.assertIn(label, VALIDATION_LABELS)
+            self.assertIn(label, CANONICAL_LABELS)
+
+
+class TestPostMergeValidationWiredIntoReview(unittest.TestCase):
+    def test_review_pr_blocks_active_approval_on_merged_pr(self):
+        report = (
+            "PR is already merged. Review decision: approve. "
+            "Validation: full suite passed."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.post_merge_validation"
+                for f in result["findings"]
+            )
+        )
+
+    def test_review_pr_allows_proven_post_merge_moot(self):
+        result = assess_final_report_validator(_MOOT_PROOF, "review_pr")
+        self.assertFalse(
+            any(
+                f["rule_id"] == "reviewer.post_merge_validation"
+                for f in result["findings"]
+            )
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_resolve_task_capability.py

@@ -287,6 +287,49 @@ class TestResolveTaskCapability(unittest.TestCase):
             self.assertEqual(res["required_operation_permission"], "gitea.pr.merge")
             self.assertEqual(res["required_role_kind"], "merger")
 
+    @patch("mcp_server.api_request", return_value={"login": "reviewer-user"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_reviewer_role_with_merge_permission_still_cannot_merge(self, _auth, _api):
+        # #539: exact permission alone is insufficient. The active profile
+        # role must also be merger for merge_pr.
+        config = json.loads(json.dumps(CONFIG_RESOLVER))
+        config["profiles"]["reviewer-with-merge"] = {
+            "enabled": True,
+            "context": "ctx",
+            "role": "reviewer",
+            "username": "reviewer-user",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_REVIEWER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.review", "gitea.pr.approve",
+                "gitea.pr.merge",
+            ],
+            "forbidden_operations": ["gitea.pr.create", "gitea.branch.push"],
+            "execution_profile": "reviewer-with-merge",
+        }
+        self._write_config(config)
+        with patch.dict(os.environ, self._env("reviewer-with-merge")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="merge_pr", remote="prgs"
+            )
+        self.assertEqual(res["required_role_kind"], "merger")
+        self.assertEqual(res["active_role_kind"], "reviewer")
+        self.assertTrue(res["active_profile_permission_allowed"])
+        self.assertFalse(res["allowed_in_current_session"])
+        self.assertTrue(res["stop_required"])
+        self.assertIn("cannot perform merger task", " ".join(res["task_role_guidance"]))
+
+    @patch("mcp_server.api_request", return_value={"login": "merger-user"})
+    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
+    def test_merger_profile_can_resolve_merge_task(self, _auth, _api):
+        with patch.dict(os.environ, self._env("merger-profile")):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="merge_pr", remote="prgs"
+            )
+        self.assertEqual(res["required_role_kind"], "merger")
+        self.assertEqual(res["active_role_kind"], "merger")
+        self.assertTrue(res["allowed_in_current_session"])
+        self.assertFalse(res["stop_required"])
+
     @patch("mcp_server.api_request")
     def test_self_review_blocked_structured(self, mock_api):
         # Test that self-approve (self-review gate) is blocked and returns structured reasons

tests/test_reviewer_pr_lease.py

@@ -237,5 +237,113 @@ class TestReviewerLeaseMcpGate(unittest.TestCase):
         self.assertTrue(any("lease" in r.lower() for r in reasons))
 
 
+
+class TestReviewerLeaseHandoffDiagnose(unittest.TestCase):
+    """#599: canonical next-action diagnosis for open-PR lease handoff."""
+
+    def setUp(self):
+        leases.clear_session_lease()
+
+    def test_no_lease_next_action_acquire(self):
+        result = leases.diagnose_reviewer_pr_lease_handoff(
+            [],
+            pr_number=592,
+            current_session_id="me-1",
+            current_reviewer_identity="sysadmin",
+            proposed_worktree="branches/review-pr-592",
+        )
+        self.assertEqual(result["classification"], "no_lease")
+        self.assertEqual(result["next_action"], leases.NEXT_ACTION_ACQUIRE)
+        self.assertIsNone(result["active_lease"])
+
+    def test_foreign_active_wait_pr592_style(self):
+        # Instructed lease gone; newer foreign claim is active.
+        old = _lease_comment(592, "23139-da018dab43ba", phase="released")
+        old["id"] = 8640
+        active = _lease_comment(592, "51515-2bfb70f0685f", phase="claimed")
+        active["id"] = 8647
+        result = leases.diagnose_reviewer_pr_lease_handoff(
+            [old, active],
+            pr_number=592,
+            current_session_id="fresh-session",
+            current_reviewer_identity="sysadmin",
+            proposed_worktree="branches/review-pr-592-issue-590",
+            instructed_session_id="23139-da018dab43ba",
+            instructed_comment_id=8640,
+        )
+        self.assertEqual(
+            result["classification"],
+            "instructed_lease_missing_with_replacement",
+        )
+        self.assertEqual(result["next_action"], leases.NEXT_ACTION_WAIT)
+        self.assertTrue(result["instructed_lease_missing_with_replacement"])
+        self.assertEqual(result["active_lease"]["session_id"], "51515-2bfb70f0685f")
+        self.assertEqual(result["active_lease"]["comment_id"], 8647)
+        self.assertFalse(result["mutation_allowed"])
+
+    def test_foreign_reclaimable_release_expired(self):
+        reclaim = _lease_comment(
+            592, "foreign-old", phase="claimed", minutes_ago=65
+        )
+        reclaim["id"] = 99
+        result = leases.diagnose_reviewer_pr_lease_handoff(
+            [reclaim],
+            pr_number=592,
+            current_session_id="me-1",
+            current_reviewer_identity="sysadmin",
+            proposed_worktree="branches/review-pr-592",
+        )
+        self.assertEqual(result["classification"], "foreign_reclaimable")
+        self.assertEqual(
+            result["next_action"], leases.NEXT_ACTION_RELEASE_EXPIRED_LEASE
+        )
+
+    def test_own_active_resume(self):
+        head = "a" * 40
+        claim = _lease_comment(592, "me-1", phase="claimed", candidate_head=head)
+        claim["id"] = 10
+        leases.record_session_lease({
+            "pr_number": 592,
+            "session_id": "me-1",
+            "candidate_head": head,
+            "comment_id": 10,
+        }, lease_provenance=mla.build_lease_provenance(
+            source=mla.SOURCE_ACQUIRE,
+            comment_id=10,
+        ))
+        result = leases.diagnose_reviewer_pr_lease_handoff(
+            [claim],
+            pr_number=592,
+            current_session_id="me-1",
+            current_reviewer_identity="rev1",
+            proposed_worktree="branches/review-pr382",
+        )
+        self.assertEqual(result["classification"], "own_active")
+        self.assertEqual(
+            result["next_action"], leases.NEXT_ACTION_RESUME_EXACT_OWNER_SESSION
+        )
+        self.assertTrue(result["mutation_allowed"])
+
+    def test_worktree_binding_mismatch(self):
+        claim = _lease_comment(592, "me-1", phase="claimed")
+        claim["id"] = 11
+        # _lease_comment uses branches/review-pr382
+        result = leases.diagnose_reviewer_pr_lease_handoff(
+            [claim],
+            pr_number=592,
+            current_session_id="me-1",
+            current_reviewer_identity="rev1",
+            proposed_worktree="branches/review-pr-592-issue-590",
+            env_bound_worktree="branches/review-pr-538",
+        )
+        self.assertEqual(result["classification"], "worktree_binding_mismatch")
+        self.assertEqual(
+            result["next_action"], leases.NEXT_ACTION_REPAIR_WORKTREE_BINDING
+        )
+        self.assertFalse(result["worktree_binding"]["match"])
+        self.assertFalse(result["mutation_allowed"])
+
+
+
 if __name__ == "__main__":
     unittest.main()

tests/test_role_session_router.py

@@ -52,6 +52,20 @@ CONFIG = {
             ],
             "execution_profile": "prgs-reviewer",
         },
+        "prgs-merger": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "merger",
+            "username": "sysadmin",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_MERGER"},
+            "allowed_operations": [
+                "gitea.read", "gitea.pr.merge", "gitea.issue.comment",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.create", "gitea.branch.push", "gitea.pr.approve",
+            ],
+            "execution_profile": "prgs-merger",
+        },
         "prgs-reconciler": {
             "enabled": True,
             "context": "ctx",
@@ -104,6 +118,7 @@ class TestRoleSessionRouter(unittest.TestCase):
             "GITEA_MCP_PROFILE": profile,
             "GITEA_TOKEN_AUTHOR": "author-pass",
             "GITEA_TOKEN_REVIEWER": "reviewer-pass",
+            "GITEA_TOKEN_MERGER": "merger-pass",
             "GITEA_TOKEN_RECONCILER": "reconciler-pass",
         }
 
@@ -227,6 +242,31 @@ class TestRoleSessionRouter(unittest.TestCase):
         self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
         self.assertTrue(route["downstream_allowed"])
 
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
+    def test_merge_task_under_reviewer_profile_wrong_role_stop(self, _auth, _api):
+        # #539: a reviewer session must not pass the pre-task merge route
+        # even if its config accidentally includes gitea.pr.merge.
+        with patch.dict(os.environ, self._env("prgs-reviewer")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="merge_pr", remote="prgs"
+            )
+        self.assertEqual(route["required_role"], "merger")
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_WRONG_ROLE)
+        self.assertFalse(route["downstream_allowed"])
+        self.assertIn("merger", route["message"].lower())
+
+    @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
+    @patch("mcp_server.get_auth_header", return_value="token merger-pass")
+    def test_merge_task_under_merger_profile_allowed(self, _auth, _api):
+        with patch.dict(os.environ, self._env("prgs-merger")):
+            route = mcp_server.gitea_route_task_session(
+                task_type="merge_pr", remote="prgs"
+            )
+        self.assertEqual(route["required_role"], "merger")
+        self.assertEqual(route["route_result"], role_session_router.ROUTE_ALLOWED)
+        self.assertTrue(route["downstream_allowed"])
+
     @patch("mcp_server.api_request", return_value={"login": "sysadmin"})
     @patch("mcp_server.get_auth_header", return_value="token reviewer-pass")
     def test_author_task_under_reviewer_profile_routes_to_author(self, _auth, _api):
@@ -334,4 +374,4 @@ class TestCheckMidMerge(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()

tests/test_stale_review_decision_lock_cleanup.py

@@ -0,0 +1,378 @@
+"""Stale #332 review-decision lock cleanup (#594).
+
+Proves:
+  a. active terminal locks still block unrelated terminal reviews
+  b. merged/closed PR locks can be cleared canonically
+  c. cleanup cannot bypass review gates on open PRs
+  d. after moot cleanup, mark_ready for a different PR can proceed
+     (PR #592-style after PR #586-style stale approve)
+"""
+
+from __future__ import annotations
+
+import os
+import unittest
+from unittest.mock import patch
+
+import mcp_server
+import stale_review_decision_lock as srdl
+from mcp_server import (
+    gitea_cleanup_stale_review_decision_lock,
+    gitea_mark_final_review_decision,
+)
+
+HEAD = "a" * 40
+FAKE_AUTH = "Basic dGVzdDp0ZXN0"
+
+REVIEWER_ENV = {
+    "GITEA_PROFILE_NAME": "prgs-reviewer",
+    "GITEA_ALLOWED_OPERATIONS": (
+        "gitea.read,gitea.pr.review,gitea.pr.approve,"
+        "gitea.pr.request_changes,gitea.pr.comment"
+    ),
+    "GITEA_SESSION_PROFILE_LOCK": "prgs-reviewer",
+}
+
+REVIEWER_PROFILE = {
+    "profile_name": "prgs-reviewer",
+    "allowed_operations": [
+        "gitea.read",
+        "gitea.pr.review",
+        "gitea.pr.approve",
+        "gitea.pr.request_changes",
+        "gitea.pr.comment",
+    ],
+    "forbidden_operations": [
+        "gitea.pr.merge",
+        "gitea.pr.create",
+        "gitea.branch.push",
+    ],
+}
+
+
+def _reviewer_profile_patch():
+    return patch.object(mcp_server, "get_profile", return_value=REVIEWER_PROFILE)
+
+
+def _lock(mutations=None, correction=False):
+    return {
+        "task": "review_pr",
+        "remote": "prgs",
+        "org": "Scaled-Tech-Consulting",
+        "repo": "Gitea-Tools",
+        "session_pid": os.getpid(),
+        "session_profile": "prgs-reviewer",
+        "session_profile_lock": "prgs-reviewer",
+        "profile_identity": "prgs-reviewer",
+        "final_review_decision_ready": False,
+        "ready_pr_number": None,
+        "ready_action": None,
+        "ready_expected_head_sha": None,
+        "ready_remote": None,
+        "ready_org": None,
+        "ready_repo": None,
+        "live_mutations": list(mutations or []),
+        "correction_authorized": correction,
+        "correction_reason": None,
+    }
+
+
+APPROVED_586 = {
+    "pr_number": 586,
+    "action": "approve",
+    "review_id": 395,
+    "review_state": "approve",
+}
+APPROVED_OPEN = {
+    "pr_number": 700,
+    "action": "approve",
+    "review_id": 1,
+    "review_state": "approve",
+}
+RC_OPEN = {
+    "pr_number": 701,
+    "action": "request_changes",
+    "review_id": 2,
+    "review_state": "request_changes",
+}
+
+
+def _seed(mutations=None, correction=False):
+    import review_workflow_load
+
+    review_workflow_load.record_review_workflow_load(mcp_server.PROJECT_ROOT)
+    mcp_server._save_review_decision_lock(_lock(mutations, correction))
+    mcp_server.gitea_load_review_workflow()
+
+
+def _merged_pr(pr_number=586, sha="6913ac9" + "0" * 33):
+    return {
+        "number": pr_number,
+        "state": "closed",
+        "merged": True,
+        "merged_at": "2026-07-09T18:20:03Z",
+        "merge_commit_sha": sha,
+    }
+
+
+def _open_pr(pr_number=700):
+    return {
+        "number": pr_number,
+        "state": "open",
+        "merged": False,
+        "merged_at": None,
+        "merge_commit_sha": None,
+    }
+
+
+# --------------------------------------------------------------------------- #
+# Pure assessment
+# --------------------------------------------------------------------------- #
+class TestAssessStaleLock(unittest.TestCase):
+    def test_no_lock(self):
+        a = srdl.assess_stale_review_decision_lock(None)
+        self.assertFalse(a["cleanup_allowed"])
+        self.assertFalse(a["is_moot"])
+
+    def test_no_terminal_mutations(self):
+        a = srdl.assess_stale_review_decision_lock(_lock([]))
+        self.assertFalse(a["cleanup_allowed"])
+
+    def test_open_pr_not_cleanable(self):
+        a = srdl.assess_stale_review_decision_lock(
+            _lock([APPROVED_OPEN]), pr_live=_open_pr(700)
+        )
+        self.assertFalse(a["is_moot"])
+        self.assertFalse(a["cleanup_allowed"])
+        self.assertTrue(any("open PR" in r for r in a["reasons"]))
+
+    def test_merged_pr_is_moot_and_cleanable(self):
+        a = srdl.assess_stale_review_decision_lock(
+            _lock([APPROVED_586]), pr_live=_merged_pr(586)
+        )
+        self.assertTrue(a["is_moot"])
+        self.assertTrue(a["cleanup_allowed"])
+        self.assertEqual(a["last_terminal_pr"], 586)
+
+    def test_closed_unmerged_is_moot(self):
+        a = srdl.assess_stale_review_decision_lock(
+            _lock([RC_OPEN]),
+            pr_live={"number": 701, "state": "closed", "merged": False},
+        )
+        self.assertTrue(a["is_moot"])
+        self.assertTrue(a["cleanup_allowed"])
+
+    def test_profile_mismatch_blocks(self):
+        a = srdl.assess_stale_review_decision_lock(
+            _lock([APPROVED_586]),
+            pr_live=_merged_pr(586),
+            active_profile_identity="other-profile",
+        )
+        self.assertFalse(a["cleanup_allowed"])
+        self.assertFalse(a["profile_match"])
+
+    def test_lookup_error_fail_closed(self):
+        a = srdl.assess_stale_review_decision_lock(
+            _lock([APPROVED_586]),
+            pr_lookup_error="timeout",
+        )
+        self.assertFalse(a["cleanup_allowed"])
+        self.assertTrue(any("timeout" in r for r in a["reasons"]))
+
+
+# --------------------------------------------------------------------------- #
+# Hard-stop still blocks active locks
+# --------------------------------------------------------------------------- #
+class TestActiveHardStopPreserved(unittest.TestCase):
+    def tearDown(self):
+        mcp_server._save_review_decision_lock(None)
+        mcp_server.review_workflow_load.clear_review_workflow_load()
+
+    def test_open_approve_still_blocks_other_pr_mark_ready(self):
+        _seed([APPROVED_OPEN])
+        reasons = mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready")
+        self.assertTrue(reasons)
+        self.assertIn("#332", reasons[0])
+        self.assertIn("gitea_cleanup_stale_review_decision_lock", reasons[0])
+
+    def test_request_changes_still_blocks_everything(self):
+        _seed([RC_OPEN])
+        for op in ("merge", "mark_ready", "review"):
+            self.assertTrue(
+                mcp_server.terminal_review_hard_stop_reasons(592, op),
+                msg=op,
+            )
+
+
+# --------------------------------------------------------------------------- #
+# MCP cleanup tool
+# --------------------------------------------------------------------------- #
+class TestCleanupTool(unittest.TestCase):
+    def setUp(self):
+        self._env = patch.dict(os.environ, REVIEWER_ENV, clear=False)
+        self._env.start()
+
+    def tearDown(self):
+        self._env.stop()
+        mcp_server._save_review_decision_lock(None)
+        mcp_server.review_workflow_load.clear_review_workflow_load()
+
+    def test_read_only_reports_moot_without_clearing(self):
+        _seed([APPROVED_586])
+        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
+                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
+                patch.object(
+                    mcp_server, "_authenticated_username", return_value="sysadmin"
+                ), \
+                _reviewer_profile_patch(), \
+                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
+            r = gitea_cleanup_stale_review_decision_lock(
+                apply=False, remote="prgs",
+                org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            )
+        self.assertTrue(r["success"])
+        self.assertTrue(r["is_moot"])
+        self.assertTrue(r["cleanup_allowed"])
+        self.assertFalse(r["cleanup_performed"])
+        self.assertIsNotNone(mcp_server._load_review_decision_lock())
+
+    def test_apply_clears_moot_lock(self):
+        _seed([APPROVED_586])
+        posts = []
+
+        def _api(method, url, auth, payload=None):
+            if method == "GET" and "/pulls/586" in url:
+                return _merged_pr()
+            if method == "POST" and "/comments" in url:
+                posts.append(payload)
+                return {"id": 9999}
+            return {}
+
+        with patch.object(mcp_server, "api_request", side_effect=_api), \
+                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
+                patch.object(
+                    mcp_server, "_authenticated_username", return_value="sysadmin"
+                ), \
+                _reviewer_profile_patch(), \
+                patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \
+                patch.object(
+                    mcp_server, "_audited",
+                    side_effect=lambda *a, **k: __import__(
+                        "contextlib"
+                    ).nullcontext(),
+                ):
+            r = gitea_cleanup_stale_review_decision_lock(
+                apply=True,
+                expected_terminal_pr=586,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(r["cleanup_performed"], r)
+        self.assertTrue(r["audit"]["applied"])
+        self.assertIsNone(mcp_server._load_review_decision_lock())
+        self.assertEqual(r.get("audit_comment_id"), 9999)
+        self.assertTrue(posts)
+
+    def test_apply_refuses_open_pr(self):
+        _seed([APPROVED_OPEN])
+        with patch.object(mcp_server, "api_request", return_value=_open_pr(700)), \
+                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
+                patch.object(
+                    mcp_server, "_authenticated_username", return_value="sysadmin"
+                ), \
+                _reviewer_profile_patch(), \
+                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
+            r = gitea_cleanup_stale_review_decision_lock(
+                apply=True, remote="prgs",
+                org="Scaled-Tech-Consulting", repo="Gitea-Tools",
+            )
+        self.assertFalse(r["cleanup_performed"])
+        self.assertFalse(r["cleanup_allowed"])
+        self.assertIsNotNone(mcp_server._load_review_decision_lock())
+
+    def test_expected_terminal_pr_pin(self):
+        _seed([APPROVED_586])
+        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
+                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
+                patch.object(
+                    mcp_server, "_authenticated_username", return_value="sysadmin"
+                ), \
+                _reviewer_profile_patch(), \
+                patch.object(mcp_server, "_profile_operation_gate", return_value=[]):
+            r = gitea_cleanup_stale_review_decision_lock(
+                apply=True,
+                expected_terminal_pr=999,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertFalse(r["cleanup_performed"])
+        self.assertTrue(any("expected_terminal_pr" in x for x in r["reasons"]))
+
+    def test_after_moot_cleanup_other_pr_mark_ready_unblocked(self):
+        """PR #592-style: after clearing merged #586 lock, new mark_ready works."""
+        _seed([APPROVED_586])
+        with patch.object(mcp_server, "api_request", return_value=_merged_pr()), \
+                patch.object(mcp_server, "_auth", return_value=FAKE_AUTH), \
+                patch.object(
+                    mcp_server, "_authenticated_username", return_value="sysadmin"
+                ), \
+                _reviewer_profile_patch(), \
+                patch.object(mcp_server, "_profile_operation_gate", return_value=[]), \
+                patch.object(
+                    mcp_server, "_audited",
+                    side_effect=lambda *a, **k: __import__(
+                        "contextlib"
+                    ).nullcontext(),
+                ):
+            cleaned = gitea_cleanup_stale_review_decision_lock(
+                apply=True,
+                expected_terminal_pr=586,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+                post_audit_comment=False,
+            )
+        self.assertTrue(cleaned["cleanup_performed"], cleaned)
+
+        # Hard-stop gone; re-init clean lock like a fresh review_pr resolve.
+        mcp_server.init_review_decision_lock(remote="prgs", task="review_pr")
+        mcp_server.gitea_load_review_workflow()
+        self.assertEqual(
+            mcp_server.terminal_review_hard_stop_reasons(592, "mark_ready"),
+            [],
+        )
+
+        no_lease = {"block": False, "reasons": [], "mutation_allowed": True}
+        with patch.object(mcp_server, "_list_pr_lease_comments", return_value=[]), \
+                patch.object(
+                    mcp_server, "_pr_work_lease_reviewer_block", return_value=no_lease
+                ), \
+                patch.object(
+                    mcp_server,
+                    "gitea_check_pr_eligibility",
+                    return_value={
+                        "eligible": True,
+                        "reasons": [],
+                        "authenticated_user": "sysadmin",
+                        "pr_author": "jcwalker3",
+                        "self_author": False,
+                    },
+                ), \
+                patch.object(
+                    mcp_server, "_authenticated_username", return_value="sysadmin"
+                ):
+            marked = gitea_mark_final_review_decision(
+                pr_number=592,
+                action="approve",
+                expected_head_sha=HEAD,
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Gitea-Tools",
+            )
+        self.assertTrue(marked.get("marked_ready"), marked)
+
+
+if __name__ == "__main__":
+    unittest.main()