Merge branch 'master' into fix/issue-702-stale-binding-lease-recovery

2026-07-17 10:33:46 -05:00
parent e33d8874a5 a19af4de22
commit 07083eae09
10 changed files with 3511 additions and 79 deletions
@@ -755,6 +755,188 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
        )


+def _anti_stomp_in_test_mode() -> bool:
+    """Whether the #604 anti-stomp gate is skipped under pytest.
+
+    Mirrors remote-repo / stable-contamination test bypasses so the existing
+    suite (bare remotes, mocked APIs) stays green. Set
+    ``GITEA_TEST_FORCE_ANTI_STOMP=1`` to exercise the live gate under tests.
+    """
+    if not _preflight_in_test_mode():
+        return False
+    return not bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
+
+
+def _run_anti_stomp_preflight(
+    task: str | None,
+    *,
+    remote: str | None = None,
+    worktree_path: str | None = None,
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+    expected_head_sha: str | None = None,
+    live_head_sha: str | None = None,
+    require_head_sha: bool = False,
+    lease_required: bool = False,
+    foreign_lease: bool | None = None,
+    lease_reasons: list[str] | None = None,
+    terminal_lock_blocks: bool | None = None,
+    terminal_lock_reasons: list[str] | None = None,
+    workflow_hash_valid: bool | None = None,
+    workflow_hash_reasons: list[str] | None = None,
+    raise_on_block: bool = True,
+) -> dict | None:
+    """Shared #604 anti-stomp preflight for MCP mutation entrypoints.
+
+    Gathers live workspace/parity/role/repo facts and invokes the pure
+    :func:`anti_stomp_preflight.assess_anti_stomp_preflight` assessor. On
+    block, raises ``RuntimeError`` (default) or returns a structured
+    :func:`anti_stomp_preflight.block_response` when *raise_on_block* is
+    False. Returns ``None`` when the mutation may proceed.
+    """
+    if _anti_stomp_in_test_mode():
+        return None
+
+    # Only enforce when the caller names a known mutation task. Read-only
+    # paths and legacy verify_preflight_purity calls without a task stay
+    # on the pre-#604 gate chain (whoami/capability/root/worktree).
+    if not task or not anti_stomp_preflight.is_mutation_task(task):
+        return None
+
+    profile = get_profile()
+    profile_name = profile.get("profile_name")
+    profile_role = _actual_profile_role()
+    allowed_operations = list(profile.get("allowed_operations") or [])
+    req_role = None
+    req_perm = None
+    if task:
+        try:
+            req_role = task_capability_map.required_role(task)
+        except KeyError:
+            req_role = _preflight_resolved_role
+        try:
+            req_perm = task_capability_map.required_permission(task)
+        except KeyError:
+            req_perm = None
+
+    ctx = _resolve_namespace_mutation_context(worktree_path)
+    workspace = ctx["workspace_path"]
+    canonical_root = ctx["canonical_repo_root"]
+    git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
+    remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
+
+    # Repo/org facts (best-effort; explicit org/repo when provided).
+    resolved_org = org
+    resolved_repo = repo
+    local_remote_url = None
+    org_explicit = org is not None
+    repo_explicit = repo is not None
+    if remote:
+        try:
+            local_remote_url = _local_git_remote_url(remote)
+        except Exception:
+            local_remote_url = None
+        if resolved_org is None or resolved_repo is None:
+            try:
+                rem = _effective_remote(remote)
+                if rem in REMOTES:
+                    if resolved_org is None:
+                        resolved_org = REMOTES[rem]["org"]
+                    if resolved_repo is None:
+                        resolved_repo = REMOTES[rem]["repo"]
+            except Exception:
+                pass
+
+    parity = _current_master_parity()
+    startup_head = parity.get("startup_head")
+    current_code_head = parity.get("current_head")
+
+    # Source contamination from durable #671 marker (role-aware).
+    source_contaminated = None
+    contamination_reasons = None
+    try:
+        marker = _load_stable_contamination_marker(remote)
+        contaminated, cont_reasons = anti_stomp_preflight.contamination_from_stable_marker(
+            marker,
+            task=task,
+            actual_role=profile_role,
+        )
+        if marker is not None:
+            source_contaminated = contaminated
+            contamination_reasons = cont_reasons
+    except Exception:
+        # Fail soft on marker I/O: existing #671 enforcer still runs separately.
+        source_contaminated = None
+
+    # Workflow-hash facts when the task is review/merge oriented.
+    if workflow_hash_valid is None and task in {
+        "review_pr",
+        "submit_pr_review",
+        "approve_pr",
+        "request_changes_pr",
+        "merge_pr",
+    }:
+        try:
+            wf_reasons = _review_workflow_load_gate_reasons()
+            workflow_hash_valid = not bool(wf_reasons)
+            workflow_hash_reasons = list(wf_reasons) if wf_reasons else None
+        except Exception:
+            pass
+
+    # Mirror remote_repo_guard's pytest bypass: under the unit suite bare
+    # remote=prgs still defaults to Timesheet, and re-checking here would
+    # break happy-path reconciler/author tests that already rely on the
+    # #530 gate being opt-in via GITEA_FORCE_REMOTE_REPO_CHECK.
+    check_repo = True
+    if "pytest" in sys.modules and not os.environ.get(
+        "GITEA_FORCE_REMOTE_REPO_CHECK"
+    ):
+        check_repo = False
+
+    assessment = anti_stomp_preflight.assess_anti_stomp_preflight(
+        task=task,
+        remote=remote,
+        resolved_org=resolved_org,
+        resolved_repo=resolved_repo,
+        local_remote_url=local_remote_url,
+        org_explicit=org_explicit,
+        repo_explicit=repo_explicit,
+        check_repo=check_repo,
+        profile_name=profile_name,
+        profile_role=profile_role,
+        required_role=req_role or _preflight_resolved_role,
+        required_permission=req_perm,
+        allowed_operations=allowed_operations,
+        workspace_path=workspace,
+        project_root=canonical_root,
+        current_branch=git_state.get("current_branch"),
+        root_head_sha=git_state.get("head_sha"),
+        root_porcelain=git_state.get("porcelain_status") or "",
+        remote_master_sha=remote_master_sha,
+        startup_head=startup_head,
+        current_code_head=current_code_head,
+        lease_required=lease_required,
+        foreign_lease=foreign_lease,
+        lease_reasons=lease_reasons,
+        terminal_lock_blocks=terminal_lock_blocks,
+        terminal_lock_reasons=terminal_lock_reasons,
+        require_head_sha=require_head_sha,
+        expected_head_sha=expected_head_sha,
+        live_head_sha=live_head_sha,
+        workflow_hash_valid=workflow_hash_valid,
+        workflow_hash_reasons=workflow_hash_reasons,
+        source_contaminated=source_contaminated,
+        contamination_reasons=contamination_reasons,
+        manual_bypass_attempted=False,
+    )
+    if not assessment.get("block"):
+        return None
+    if raise_on_block:
+        raise RuntimeError(anti_stomp_preflight.format_anti_stomp_error(assessment))
+    return anti_stomp_preflight.block_response(assessment)
+
+
 def verify_preflight_purity(
    remote: str | None = None,
    worktree_path: str | None = None,
@@ -762,13 +944,29 @@ def verify_preflight_purity(
    *,
    target_issue_number: int | None = None,
    require_author_lock: bool = False,
+    expected_head_sha: str | None = None,
+    live_head_sha: str | None = None,
+    require_head_sha: bool = False,
+    lease_required: bool = False,
+    foreign_lease: bool | None = None,
+    lease_reasons: list[str] | None = None,
+    terminal_lock_blocks: bool | None = None,
+    terminal_lock_reasons: list[str] | None = None,
+    workflow_hash_valid: bool | None = None,
+    workflow_hash_reasons: list[str] | None = None,
+    org: str | None = None,
+    repo: str | None = None,
 ):
-    """Verify identity/capability order, then production workspace guards.
+    """Verify identity/capability order, production workspace guards, then anti-stomp.

    #683: pytest/unittest must not skip production root/branches/scope
    enforcement when force-on signals request production behavior. The
    early return below only skips *preflight-order* purity checks under
    pure unit-test isolation — never when production guards are active.
+
+    #604: also runs the shared anti-stomp preflight for mutation tasks (typed
+    blocker + exact next action). Extended lease/head/terminal facts may be
+    supplied by review/merge callers.
    """
    global _preflight_reviewer_violation_files

@@ -779,7 +977,12 @@ def verify_preflight_purity(
    # Pure unit-test isolation: skip purity-order unless legacy dirty/porcelain
    # force flags request the dirtiness path. #683 FORCE_PRODUCTION_GUARDS alone
    # runs production root/branches/scope without requiring whoami/capability.
-    skip_purity_order = in_test and not workflow_scope_guard.purity_order_forced()
+    force_anti_stomp = bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
+    skip_purity_order = (
+        in_test
+        and not workflow_scope_guard.purity_order_forced()
+        and not force_anti_stomp
+    )

    if not skip_purity_order:
        if not _preflight_whoami_called:
@@ -874,6 +1077,25 @@ def verify_preflight_purity(
            target_issue_number=target_issue_number,
            require_author_lock=require_author_lock,
        )
+        # #604: common anti-stomp preflight after legacy + #683 enforcers.
+        _run_anti_stomp_preflight(
+            task,
+            remote=remote,
+            worktree_path=worktree_path,
+            org=org,
+            repo=repo,
+            expected_head_sha=expected_head_sha,
+            live_head_sha=live_head_sha,
+            require_head_sha=require_head_sha,
+            lease_required=lease_required,
+            foreign_lease=foreign_lease,
+            lease_reasons=lease_reasons,
+            terminal_lock_blocks=terminal_lock_blocks,
+            terminal_lock_reasons=terminal_lock_reasons,
+            workflow_hash_valid=workflow_hash_valid,
+            workflow_hash_reasons=workflow_hash_reasons,
+            raise_on_block=True,
+        )
        _clear_preflight_capability_state()
        return

@@ -888,6 +1110,25 @@ def verify_preflight_purity(
            target_issue_number=target_issue_number,
            require_author_lock=require_author_lock,
        )
+        if force_anti_stomp:
+            _run_anti_stomp_preflight(
+                task,
+                remote=remote,
+                worktree_path=worktree_path,
+                org=org,
+                repo=repo,
+                expected_head_sha=expected_head_sha,
+                live_head_sha=live_head_sha,
+                require_head_sha=require_head_sha,
+                lease_required=lease_required,
+                foreign_lease=foreign_lease,
+                lease_reasons=lease_reasons,
+                terminal_lock_blocks=terminal_lock_blocks,
+                terminal_lock_reasons=terminal_lock_reasons,
+                workflow_hash_valid=workflow_hash_valid,
+                workflow_hash_reasons=workflow_hash_reasons,
+                raise_on_block=True,
+            )


 def _session_issue_lock_snapshot(
@@ -1256,6 +1497,7 @@ import root_checkout_guard  # noqa: E402
 import workflow_scope_guard  # noqa: E402  # #683 production scope / force-on guards
 import stable_branch_push_guard  # noqa: E402
 import remote_repo_guard  # noqa: E402
+import anti_stomp_preflight  # noqa: E402
 import issue_claim_heartbeat  # noqa: E402
 import issue_work_duplicate_gate  # noqa: E402
 import issue_workflow_labels  # noqa: E402
@@ -3617,6 +3859,8 @@ def init_review_decision_lock(remote: str | None, task: str | None, force: bool
        "live_mutations": [],
        "correction_authorized": False,
        "correction_reason": None,
+        "correction_pr_number": None,
+        "correction_head_sha": None,
    })


@@ -3712,21 +3956,30 @@ def check_review_decision_gate(
    if stale_review_decision_lock.prior_live_mutations_block_boundary(
        lock, pr_number=pr_number, expected_head_sha=ready_head
    ):
-        if prior and not lock.get("correction_authorized"):
+        scoped = stale_review_decision_lock.correction_applies(
+            lock, pr_number=pr_number, expected_head_sha=ready_head
+        )
+        if prior and not scoped:
            reasons.append(
                "live review mutation already recorded for this PR head in this "
                "run; only one live review mutation is allowed per head unless "
-                "gitea_authorize_review_correction was invoked (fail closed, #620)"
+                "gitea_authorize_review_correction was invoked for this same "
+                "PR/head (fail closed, #620/#693)"
            )
        elif (
            action in _TERMINAL_REVIEW_ACTIONS
-            and any(m.get("action") in _TERMINAL_REVIEW_ACTIONS for m in prior)
-            and not lock.get("correction_authorized")
+            and any(
+                isinstance(m, dict)
+                and m.get("action") in _TERMINAL_REVIEW_ACTIONS
+                and m.get("pr_number") == pr_number
+                for m in prior
+            )
+            and not scoped
        ):
            reasons.append(
                "terminal review decision already submitted for this PR head in "
                "this run; blocked unless an operator-approved correction was "
-                "authorized (fail closed, #620)"
+                "authorized for this same PR/head (fail closed, #620/#693)"
            )

    return reasons
@@ -3755,6 +4008,8 @@ def record_live_review_mutation(pr_number: int, action: str, review_id: int | No
    if lock.get("correction_authorized"):
        lock["correction_authorized"] = False
        lock["correction_reason"] = None
+        lock["correction_pr_number"] = None
+        lock["correction_head_sha"] = None
    _save_review_decision_lock(lock)


@@ -3798,11 +4053,13 @@ def terminal_review_hard_stop_reasons(
        and last.get("pr_number") == pr_number
    ):
        return []
-    if operation in ("mark_ready", "review", "resume") and lock.get(
-        "correction_authorized"
+    if operation in ("mark_ready", "review", "resume") and (
+        stale_review_decision_lock.correction_applies(
+            lock, pr_number=pr_number, expected_head_sha=expected_head_sha
+        )
    ):
        return []
-    # #620: same PR, different reviewed head → fresh decision boundary.
+    # #620 same-PR new head + #693 cross-PR isolation.
    if stale_review_decision_lock.terminal_boundary_allows_fresh_decision(
        lock,
        pr_number=pr_number,
@@ -3824,16 +4081,20 @@ def terminal_review_hard_stop_reasons(
        else ""
    )
    recovery = (
-        "if that PR is already merged/closed, use "
-        "gitea_cleanup_stale_review_decision_lock (apply=true) after live-state "
-        "proof (#594); if the PR is still open but the head moved, mark/submit "
-        "with the new expected_head_sha (#620); never delete session-state "
-        "files by hand"
+        "exact_next_action: if that PR is already merged/closed, use "
+        "gitea_cleanup_stale_review_decision_lock(apply=true) after live-state "
+        "proof (#594); if the same PR head moved, mark/submit with the new "
+        "expected_head_sha (#620); if the target is a different PR, "
+        "gitea_diagnose_review_decision_lock then mark_final on the target "
+        "(cross-PR isolation, #693); if same-head verdict was mistaken, "
+        "gitea_authorize_review_correction with matching target_pr_number "
+        "(not a generic unlock); never delete session-state files by hand; "
+        "if still blocked create/update a durable Gitea issue and stop"
    )
    return [
        "terminal review mutation already consumed in this run "
        f"({last.get('action')} on PR #{last.get('pr_number')}{head_note}); "
-        f"{guidance} (fail closed, #332); {recovery}"
+        f"{guidance} (fail closed, #332/#693); {recovery}"
    ]


@@ -4393,6 +4654,30 @@ def _evaluate_pr_review_submission(
        result["pr_work_lease"] = lease_block
        return result

+    if live:
+        # #604: common anti-stomp with live head + lease proof (typed blocker).
+        anti_task = {
+            "approve": "approve_pr",
+            "request_changes": "request_changes_pr",
+            "comment": "submit_pr_review",
+        }.get(action, "submit_pr_review")
+        try:
+            _run_anti_stomp_preflight(
+                anti_task,
+                remote=remote,
+                worktree_path=worktree_path,
+                org=org,
+                repo=repo,
+                expected_head_sha=pinned_sha,
+                live_head_sha=actual_sha,
+                require_head_sha=True,
+                foreign_lease=False,
+                terminal_lock_blocks=False,
+            )
+        except RuntimeError as exc:
+            reasons.append(str(exc))
+            return result
+
    if (body or "").strip():
        gate = _canonical_comment_gate(body, context="pr_review")
        if gate["blocked"]:
@@ -4498,10 +4783,33 @@ def gitea_mark_final_review_decision(
            "reasons": [
                "review decision lock missing; resolve review_pr capability first"
            ],
+            "exact_next_action": (
+                "gitea_resolve_task_capability(task='review_pr') then retry "
+                "mark_final"
+            ),
+            "durable_issue_handoff": {
+                "required": True,
+                "instruction": (
+                    "If the lock cannot be seeded, create/update a durable "
+                    "Gitea issue and stop (#693 AC8)"
+                ),
+            },
        }
    session_reasons = _review_decision_session_reasons(lock)
    if session_reasons:
-        return {"marked_ready": False, "reasons": session_reasons}
+        classification = stale_review_decision_lock.classify_review_decision_lock(
+            lock,
+            target_pr_number=pr_number,
+            target_head_sha=expected_head_sha,
+            session_blockers=session_reasons,
+        )
+        fail = stale_review_decision_lock.build_precise_mark_final_failure(
+            classification
+        )
+        return {
+            "marked_ready": False,
+            **fail,
+        }
    if remote != lock.get("remote"):
        return {
            "marked_ready": False,
@@ -4509,6 +4817,7 @@ def gitea_mark_final_review_decision(
                f"requested remote '{remote}' does not match locked remote "
                f"'{lock.get('remote')}' (fail closed)"
            ],
+            "exact_next_action": "use the remote bound on the decision lock",
        }
    try:
        _, resolved_org, resolved_repo = _resolve(remote, None, org, repo)
@@ -4543,7 +4852,26 @@ def gitea_mark_final_review_decision(
        pr_number, "mark_ready", expected_head_sha=expected_head_sha
    )
    if hard_stop:
-        return {"marked_ready": False, "reasons": hard_stop}
+        classification = stale_review_decision_lock.classify_review_decision_lock(
+            lock,
+            target_pr_number=pr_number,
+            target_head_sha=expected_head_sha,
+            active_profile_identity=_decision_lock_binding().get(
+                "profile_identity"
+            ),
+        )
+        fail = stale_review_decision_lock.build_precise_mark_final_failure(
+            classification
+        )
+        # Prefer classification next_action when more precise; keep hard_stop
+        # text as primary reasons for operators.
+        return {
+            "marked_ready": False,
+            "reasons": hard_stop + list(fail.get("reasons") or []),
+            "classification": fail.get("classification"),
+            "exact_next_action": fail.get("exact_next_action"),
+            "durable_issue_handoff": fail.get("durable_issue_handoff"),
+        }
    if action not in _REVIEW_ACTIONS:
        return {
            "marked_ready": False,
@@ -4559,17 +4887,63 @@ def gitea_mark_final_review_decision(
                "expected_head_sha required before marking final review "
                "decision (fail closed, #399)"
            ],
+            "exact_next_action": "retry mark_final with expected_head_sha pin",
+        }
+    # Safe idempotency: already ready for same PR/action/head (#693 AC4).
+    ready_head = stale_review_decision_lock.normalize_head_sha(
+        lock.get("ready_expected_head_sha")
+    )
+    target_head = stale_review_decision_lock.normalize_head_sha(expected_head_sha)
+    if (
+        lock.get("final_review_decision_ready")
+        and lock.get("ready_pr_number") == pr_number
+        and lock.get("ready_action") == action
+        and ready_head
+        and target_head
+        and stale_review_decision_lock.heads_equal(ready_head, target_head)
+        and lock.get("ready_remote") == remote
+        and lock.get("ready_org") == org
+        and lock.get("ready_repo") == repo
+    ):
+        return {
+            "marked_ready": True,
+            "idempotent": True,
+            "pr_number": pr_number,
+            "action": action,
+            "expected_head_sha": expected_head_sha,
+            "remote": remote,
+            "org": org,
+            "repo": repo,
+            "final_review_decision_ready": True,
+            "head_scoped": True,
+            "reasons": [
+                "final review decision already marked ready for this PR/action/"
+                "head (idempotent, #693)"
+            ],
        }
    if stale_review_decision_lock.prior_live_mutations_block_boundary(
        lock, pr_number=pr_number, expected_head_sha=expected_head_sha
    ):
+        classification = stale_review_decision_lock.classify_review_decision_lock(
+            lock,
+            target_pr_number=pr_number,
+            target_head_sha=expected_head_sha,
+        )
+        fail = stale_review_decision_lock.build_precise_mark_final_failure(
+            classification
+        )
        return {
            "marked_ready": False,
            "reasons": [
                "cannot mark final decision after a live review mutation was "
                "already recorded for this PR head unless an operator-approved "
-                "correction was authorized (fail closed, #620)"
-            ],
+                "correction was authorized for this same PR/head "
+                "(fail closed, #620/#693)"
+            ]
+            + list(fail.get("reasons") or []),
+            "classification": fail.get("classification"),
+            "exact_next_action": fail.get("exact_next_action"),
+            "durable_issue_handoff": fail.get("durable_issue_handoff"),
        }
    elig = gitea_check_pr_eligibility(
        pr_number=pr_number,
@@ -4653,8 +5027,19 @@ def gitea_authorize_review_correction(
    prior_review_state: str,
    reason: str,
    operator_authorized: bool = False,
+    target_pr_number: int | None = None,
+    expected_head_sha: str | None = None,
+    remote: str = "dadeschools",
+    org: str | None = None,
+    repo: str | None = None,
+    post_audit_comment: bool = True,
 ) -> dict:
-    """Authorize one operator-approved correction after a mistaken live review."""
+    """Authorize one operator-approved correction after a mistaken live review.
+
+    #693: correction is scoped to the named prior review's **PR and head**.
+    It cannot unlock a different PR (no generic decision-lock bypass).
+    Successful authorization posts a thread-visible audit on the target PR.
+    """
    reason = (reason or "").strip()
    prior_review_state = (prior_review_state or "").strip().lower()
    lock = _load_review_decision_lock()
@@ -4680,7 +5065,20 @@ def gitea_authorize_review_correction(
    last_mutation = prior[-1]
    last_review_id = last_mutation.get("review_id")
    last_review_state = last_mutation.get("action")
+    last_pr = last_mutation.get("pr_number")
+    last_head = stale_review_decision_lock.mutation_head_sha(last_mutation, lock)
    reasons = []
+    if target_pr_number is None:
+        reasons.append(
+            "target_pr_number is required so correction cannot become a "
+            "generic unlock (fail closed, #693)"
+        )
+    elif last_pr is not None and int(target_pr_number) != int(last_pr):
+        reasons.append(
+            f"target_pr_number #{target_pr_number} does not match last "
+            f"recorded mutation PR #{last_pr}; correction cannot unlock a "
+            f"different PR (fail closed, #693)"
+        )
    if last_review_id is not None and last_review_id != prior_review_id:
        reasons.append(
            f"prior review ID '{prior_review_id}' does not match last "
@@ -4691,14 +5089,202 @@ def gitea_authorize_review_correction(
            f"prior review state '{prior_review_state}' does not match last "
            f"recorded review state '{last_review_state}' (fail closed)"
        )
+    want_head = stale_review_decision_lock.normalize_head_sha(expected_head_sha)
+    if want_head and last_head and not stale_review_decision_lock.heads_equal(
+        want_head, last_head
+    ):
+        reasons.append(
+            f"expected_head_sha does not match last mutation head "
+            f"{last_head[:12]}… (fail closed, #693)"
+        )
    if not reason:
        reasons.append("correction reason is required")
+    try:
+        actor = None
+        h, o, r = _resolve(remote, None, org, repo)
+        try:
+            actor = _authenticated_username(h)
+        except Exception:
+            actor = None
+    except Exception as exc:  # noqa: BLE001
+        return {
+            "authorized": False,
+            "reasons": [f"could not resolve remote for audit: {_redact(str(exc))}"],
+        }
+    profile = get_profile()
+    profile_name = (profile.get("profile_name") or "").strip() or None
+    audit = stale_review_decision_lock.build_correction_audit_record(
+        prior_review_id=prior_review_id,
+        prior_review_state=prior_review_state,
+        target_pr_number=target_pr_number if target_pr_number is not None else last_pr,
+        target_head_sha=want_head or last_head,
+        reason=reason,
+        actor_username=actor,
+        profile_name=profile_name,
+        authorized=False,
+    )
    if reasons:
-        return {"authorized": False, "reasons": reasons}
+        return {
+            "authorized": False,
+            "reasons": reasons,
+            "audit": audit,
+            "exact_next_action": (
+                "use same-PR correction only, or for a different PR use "
+                "gitea_diagnose_review_decision_lock / cross-PR isolation "
+                "(#693) or moot cleanup (#594)"
+            ),
+        }
+    scope_pr = int(target_pr_number) if target_pr_number is not None else int(last_pr)
+    scope_head = want_head or last_head
    lock["correction_authorized"] = True
    lock["correction_reason"] = reason
+    lock["correction_pr_number"] = scope_pr
+    lock["correction_head_sha"] = scope_head
    _save_review_decision_lock(lock)
-    return {"authorized": True, "correction_reason": reason, "reasons": []}
+    audit = stale_review_decision_lock.build_correction_audit_record(
+        prior_review_id=prior_review_id,
+        prior_review_state=prior_review_state,
+        target_pr_number=scope_pr,
+        target_head_sha=scope_head,
+        reason=reason,
+        actor_username=actor,
+        profile_name=profile_name,
+        authorized=True,
+    )
+    report = {
+        "authorized": True,
+        "correction_reason": reason,
+        "correction_pr_number": scope_pr,
+        "correction_head_sha": scope_head,
+        "audit": audit,
+        "audit_comment_id": None,
+        "reasons": [],
+        "scope": "same_pr_head_only",
+    }
+    if post_audit_comment and scope_pr is not None:
+        comment_block = _profile_operation_gate("gitea.pr.comment")
+        if comment_block:
+            report["audit_comment_skipped"] = comment_block
+        else:
+            try:
+                auth = _auth(h)
+                body = stale_review_decision_lock.format_correction_audit_comment(
+                    audit
+                )
+                with _audited(
+                    "comment_pr",
+                    host=h,
+                    remote=remote,
+                    org=o,
+                    repo=r,
+                    pr_number=scope_pr,
+                    request_metadata={"source": "authorize_review_correction"},
+                ):
+                    posted = api_request(
+                        "POST",
+                        f"{repo_api_url(h, o, r)}/issues/{scope_pr}/comments",
+                        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_diagnose_review_decision_lock(
+    target_pr_number: int,
+    expected_head_sha: str | None = None,
+    remote: str = "dadeschools",
+    host: str | None = None,
+    org: str | None = None,
+    repo: str | None = None,
+) -> dict:
+    """Read-only: classify durable review-decision lock state for a target PR (#693).
+
+    Returns a deterministic classification, owner/evidence, exact next_action,
+    and whether mark_final / cleanup / scoped correction are appropriate.
+    Never mutates the lock and never authorizes correction.
+    """
+    read_block = _profile_operation_gate("gitea.read")
+    if read_block:
+        return {
+            "success": False,
+            "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()
+    lock = _load_review_decision_lock()
+    session_blockers = _review_decision_session_reasons(lock) if lock else []
+    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
+            pr_lookup_error = _redact(str(exc))
+    classification = stale_review_decision_lock.classify_review_decision_lock(
+        lock,
+        target_pr_number=target_pr_number,
+        target_head_sha=expected_head_sha,
+        pr_live=pr_live,
+        pr_lookup_error=pr_lookup_error,
+        active_profile_identity=binding.get("profile_identity"),
+        session_blockers=session_blockers,
+    )
+    recovery = stale_review_decision_lock.assess_open_pr_decision_lock_recovery(
+        lock,
+        target_pr_number=target_pr_number,
+        target_head_sha=expected_head_sha,
+        last_terminal_pr_live=pr_live,
+        pr_lookup_error=pr_lookup_error,
+        active_profile_identity=binding.get("profile_identity"),
+        session_blockers=session_blockers,
+    )
+    try:
+        actor = _authenticated_username(h)
+    except Exception:
+        actor = None
+    return {
+        "success": True,
+        "remote": remote,
+        "org": o,
+        "repo": r,
+        "authenticated_username": actor,
+        "active_profile_identity": binding.get("profile_identity"),
+        "classification": classification.get("classification"),
+        "mark_final_allowed": classification.get("mark_final_allowed"),
+        "cleanup_allowed": classification.get("cleanup_allowed"),
+        "correction_eligible": classification.get("correction_eligible"),
+        "recovery_allowed": recovery.get("recovery_allowed"),
+        "recovery_mode": recovery.get("recovery_mode"),
+        "exact_next_action": classification.get("next_action"),
+        "owner_profile_identity": classification.get("owner_profile_identity"),
+        "last_terminal_pr": classification.get("last_terminal_pr"),
+        "last_terminal_action": classification.get("last_terminal_action"),
+        "locked_head_sha": classification.get("locked_head_sha"),
+        "target_pr_number": target_pr_number,
+        "target_head_sha": stale_review_decision_lock.normalize_head_sha(
+            expected_head_sha
+        ),
+        "lock_summary": classification.get("lock_summary"),
+        "evidence": classification.get("evidence"),
+        "reasons": list(classification.get("reasons") or []),
+        "manual_file_delete_forbidden": True,
+        "correction_as_generic_unlock_forbidden": True,
+        "durable_issue_handoff": stale_review_decision_lock.build_precise_mark_final_failure(
+            classification
+        ).get("durable_issue_handoff"),
+    }



@@ -6672,7 +7258,12 @@ def gitea_edit_pr(
        raise ValueError("At least one field to edit (title, body, state, base) must be provided.")

    closing = payload.get("state") == "closed"
-    verify_preflight_purity(remote, task="close_pr" if closing else None)
+    # Closing uses close_pr; non-closing title/body/base edits use edit_pr so
+    # the shared #604 anti-stomp inventory stays consistent with wiring.
+    if closing:
+        verify_preflight_purity(remote, task="close_pr")
+    else:
+        verify_preflight_purity(remote, task="edit_pr")

    # PR closure is a first-class capability, distinct from retitling or
    # rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
@@ -7164,6 +7755,25 @@ def gitea_merge_pr(
        reasons.extend(lease_block.get("reasons") or [])
        result["pr_work_lease"] = lease_block
        return result
+
+    # #604: common anti-stomp with live head + lease proof (typed blocker).
+    try:
+        _run_anti_stomp_preflight(
+            "merge_pr",
+            remote=remote,
+            worktree_path=worktree_path,
+            org=org,
+            repo=repo,
+            expected_head_sha=expected_head_sha,
+            live_head_sha=actual_sha,
+            require_head_sha=True,
+            foreign_lease=False,
+            terminal_lock_blocks=False,
+        )
+    except RuntimeError as exc:
+        reasons.append(str(exc))
+        return result
+
    if expected_head_sha and actual_sha and expected_head_sha != actual_sha:
        reasons.append(
            "expected head SHA does not match current PR head (fail closed)"
@@ -10040,7 +10650,11 @@ def gitea_acquire_reviewer_pr_lease(
            "permission_report": _permission_block_report("gitea.pr.comment"),
        }

-    _verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
+    # task=acquire_reviewer_pr_lease so verify_preflight_purity runs shared #604
+    # anti-stomp for the declared lease-acquire mutation inventory entry.
+    _verify_role_mutation_workspace(
+        remote, worktree=worktree, task="acquire_reviewer_pr_lease"
+    )
    h, o, r = _resolve(remote, host, org, repo)
    auth = _auth(h)
    profile = get_profile()
@@ -10181,6 +10795,7 @@ def gitea_adopt_merger_pr_lease(
            "permission_report": _permission_block_report("gitea.pr.merge"),
        }

+    # task=adopt_merger_pr_lease so verify_preflight_purity runs shared #604 anti-stomp.
    _verify_role_mutation_workspace(
        remote, worktree=worktree, task="adopt_merger_pr_lease"
    )