feat: common anti-stomp preflight before mutation tools (Closes #604)

Add a shared fail-closed anti-stomp preflight that mutation tools invoke
before create/comment/lease/review/approve/request-changes/merge/cleanup/
label mutations. Composes repo/role/root/worktree/lease/terminal-lock/
head-SHA/stale-runtime/workflow-hash/contamination checks into a typed
blocker with an exact next action. No agent bypass flags.

anti_stomp_preflight.py

@@ -0,0 +1,705 @@
+"""Common anti-stomp preflight for every MCP mutation tool (#604).
+
+Even with leases, mutation tools need a **shared** preflight that prevents
+stale sessions, wrong worktrees, wrong repos, old prompts, terminal locks,
+foreign leases, contaminated approvals, and root-checkout mutations from
+slipping through.
+
+This module is the pure assessment core. Callers (MCP mutation entrypoints)
+gather live facts and pass them in — nothing here performs git, network, or
+durable-state I/O. Existing happy-path guards remain authoritative; this
+module composes them into one typed, fail-closed result.
+
+Required checks (issue #604):
+
+* repo/org verification
+* profile/role verification
+* root checkout clean and not used for mutation (when role requires branches/)
+* worktree under branches when required
+* active lease ownership (when lease is required for the mutation)
+* terminal lock status (when applicable)
+* expected head SHA (when pinned)
+* stale runtime status
+* workflow hash (when a workflow load is required)
+* source contamination status
+* no manual state/mtime/source-file bypass
+
+Failure returns a typed blocker and an exact next action. There is **no**
+agent-facing bypass flag.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import author_mutation_worktree
+import master_parity_gate
+import remote_repo_guard
+import root_checkout_guard
+import stable_branch_push_guard
+
+# ── public constants ──────────────────────────────────────────────────────────
+
+# Typed blocker kinds returned in the structured result.
+BLOCKER_WRONG_REPO = "wrong_repo"
+BLOCKER_WRONG_ROLE = "wrong_role"
+BLOCKER_ROOT_CHECKOUT = "root_checkout_mutation"
+BLOCKER_WRONG_WORKTREE = "wrong_worktree"
+BLOCKER_FOREIGN_LEASE = "foreign_lease"
+BLOCKER_TERMINAL_LOCK = "terminal_lock"
+BLOCKER_HEAD_SHA = "head_sha_mismatch"
+BLOCKER_STALE_RUNTIME = "stale_runtime"
+BLOCKER_WORKFLOW_HASH = "workflow_hash"
+BLOCKER_SOURCE_CONTAMINATION = "source_contamination"
+BLOCKER_MANUAL_BYPASS = "manual_bypass"
+
+BLOCKER_KINDS = frozenset({
+    BLOCKER_WRONG_REPO,
+    BLOCKER_WRONG_ROLE,
+    BLOCKER_ROOT_CHECKOUT,
+    BLOCKER_WRONG_WORKTREE,
+    BLOCKER_FOREIGN_LEASE,
+    BLOCKER_TERMINAL_LOCK,
+    BLOCKER_HEAD_SHA,
+    BLOCKER_STALE_RUNTIME,
+    BLOCKER_WORKFLOW_HASH,
+    BLOCKER_SOURCE_CONTAMINATION,
+    BLOCKER_MANUAL_BYPASS,
+})
+
+# Mutation tasks that must invoke this preflight before acting.
+# Covers create issue/comment, lease acquire, submit review, approve,
+# request changes, merge, cleanup, and label mutation (issue #604 AC1).
+MUTATION_TASKS = frozenset({
+    "create_issue",
+    "comment_issue",
+    "close_issue",
+    "claim_issue",
+    "mark_issue",
+    "lock_issue",
+    "set_issue_labels",
+    "create_label",
+    "create_pr",
+    "comment_pr",
+    "close_pr",
+    "commit_files",
+    "gitea_commit_files",
+    "create_branch",
+    "push_branch",
+    "delete_branch",
+    "cleanup_merged_pr_branch",
+    "cleanup_stale_claims",
+    "cleanup_stale_review_decision_lock",
+    "gitea_cleanup_stale_review_decision_lock",
+    "reconcile_merged_cleanups",
+    "reconcile_already_landed_pr",
+    "reconcile_close_superseded_pr",
+    "reconciliation_cleanup",
+    "post_heartbeat",
+    "acquire_reviewer_pr_lease",
+    "gitea_acquire_reviewer_pr_lease",
+    "adopt_merger_pr_lease",
+    "heartbeat_reviewer_pr_lease",
+    "release_reviewer_pr_lease",
+    "review_pr",
+    "submit_pr_review",
+    "approve_pr",
+    "request_changes_pr",
+    "merge_pr",
+    "mark_final_review_decision",
+    "save_review_draft",
+    "resume_review_draft",
+})
+
+# Roles that must mutate from a branches/ worktree (not the control checkout).
+_BRANCHES_REQUIRED_ROLES = frozenset({"author"})
+
+# Reviewer/merger mutations that require an owned lease + head pinning when
+# the caller supplies those facts.
+_LEASE_AWARE_TASKS = frozenset({
+    "review_pr",
+    "submit_pr_review",
+    "approve_pr",
+    "request_changes_pr",
+    "merge_pr",
+    "mark_final_review_decision",
+    "acquire_reviewer_pr_lease",
+    "gitea_acquire_reviewer_pr_lease",
+    "adopt_merger_pr_lease",
+    "heartbeat_reviewer_pr_lease",
+    "release_reviewer_pr_lease",
+    "resume_review_draft",
+})
+
+_NEXT_ACTIONS: dict[str, str] = {
+    BLOCKER_WRONG_REPO: (
+        "Pass explicit org= and repo= matching the local git remote "
+        "(e.g. org=Scaled-Tech-Consulting repo=Gitea-Tools) and retry."
+    ),
+    BLOCKER_WRONG_ROLE: (
+        "Call gitea_resolve_task_capability, switch to the required "
+        "namespace/profile, re-verify with gitea_whoami, then retry."
+    ),
+    BLOCKER_ROOT_CHECKOUT: (
+        "Leave the root control checkout on clean master; create or switch "
+        "to a session-owned worktree under branches/ and retry the mutation "
+        "with worktree_path set."
+    ),
+    BLOCKER_WRONG_WORKTREE: (
+        "Create or bind a session-owned worktree under branches/, set "
+        "worktree_path (or GITEA_ACTIVE_WORKTREE), and retry."
+    ),
+    BLOCKER_FOREIGN_LEASE: (
+        "Stop. Do not stomp a foreign lease. Wait for expiry, request "
+        "takeover through the sanctioned path, or hand off to the owner."
+    ),
+    BLOCKER_TERMINAL_LOCK: (
+        "Stop. A terminal review-decision lock is active for this head. "
+        "Do not re-approve/merge; follow the #332/#620 recovery path."
+    ),
+    BLOCKER_HEAD_SHA: (
+        "Re-fetch the live PR head with gitea_view_pr, re-pin "
+        "expected_head_sha to the current head, re-validate, then retry."
+    ),
+    BLOCKER_STALE_RUNTIME: (
+        "Restart the Gitea MCP server so it reloads master's capability "
+        "gates, re-run gitea_whoami + gitea_resolve_task_capability, then retry."
+    ),
+    BLOCKER_WORKFLOW_HASH: (
+        "Reload the canonical workflow via gitea_load_review_workflow, then "
+        "rerun the full review-merge workflow from inventory (no approve/merge replay)."
+    ),
+    BLOCKER_SOURCE_CONTAMINATION: (
+        "Stop. Session is source-contaminated. Hand off to a reconciler for "
+        "audit/clear; do not clear markers by deleting session-state files."
+    ),
+    BLOCKER_MANUAL_BYPASS: (
+        "Stop. Manual state/mtime/source-file bypass is forbidden. Use "
+        "sanctioned MCP tools only; never delete or rewrite session-state "
+        "or lock files by hand."
+    ),
+}
+
+
+def is_mutation_task(task: str | None) -> bool:
+    """True when *task* is in the #604 mutation set (normalized)."""
+    name = (task or "").strip().lower().removeprefix("gitea_")
+    if not name:
+        return False
+    if name in MUTATION_TASKS:
+        return True
+    # Accept gitea_ prefixed membership for aliases already in the set.
+    return f"gitea_{name}" in MUTATION_TASKS
+
+
+def roles_compatible(active_role: str | None, required_role: str | None) -> bool:
+    """Whether *active_role* may perform a task that maps to *required_role*.
+
+    Exact match always passes. Reconcilers may run author-class mutations
+    (comment/close/cleanup) that the capability map stamps as ``author`` —
+    matching the long-standing reconciler exemption for control-checkout
+    work. No other cross-role substitutions are allowed.
+    """
+    active = (active_role or "").strip().lower()
+    required = (required_role or "").strip().lower()
+    if not active or not required:
+        return True
+    if active == required:
+        return True
+    if active == "reconciler" and required in {"author", "reconciler"}:
+        return True
+    return False
+
+
+def _blocker(
+    kind: str,
+    reasons: list[str],
+    *,
+    exact_next_action: str | None = None,
+    detail: dict | None = None,
+) -> dict[str, Any]:
+    if kind not in BLOCKER_KINDS:
+        raise ValueError(f"unknown anti-stomp blocker kind: {kind!r}")
+    return {
+        "kind": kind,
+        "reasons": list(reasons),
+        "exact_next_action": exact_next_action or _NEXT_ACTIONS[kind],
+        "detail": dict(detail or {}),
+    }
+
+
+def _allowed_result(checks: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "allowed": True,
+        "block": False,
+        "blockers": [],
+        "reasons": [],
+        "exact_next_action": "proceed",
+        "blocker_kind": None,
+        "checks": checks,
+    }
+
+
+def _blocked_result(
+    blockers: list[dict[str, Any]],
+    checks: dict[str, Any],
+) -> dict[str, Any]:
+    primary = blockers[0]
+    all_reasons: list[str] = []
+    for b in blockers:
+        for r in b.get("reasons") or []:
+            if r not in all_reasons:
+                all_reasons.append(r)
+    return {
+        "allowed": False,
+        "block": True,
+        "blockers": blockers,
+        "reasons": all_reasons,
+        "exact_next_action": primary["exact_next_action"],
+        "blocker_kind": primary["kind"],
+        "checks": checks,
+    }
+
+
+def assess_anti_stomp_preflight(
+    *,
+    task: str | None = None,
+    # repo/org
+    remote: str | None = None,
+    resolved_org: str | None = None,
+    resolved_repo: str | None = None,
+    local_remote_url: str | None = None,
+    org_explicit: bool = False,
+    repo_explicit: bool = False,
+    check_repo: bool = True,
+    # profile/role
+    profile_name: str | None = None,
+    profile_role: str | None = None,
+    required_role: str | None = None,
+    check_role: bool = True,
+    # root checkout + worktree
+    workspace_path: str | None = None,
+    project_root: str | None = None,
+    current_branch: str | None = None,
+    root_head_sha: str | None = None,
+    root_porcelain: str | None = None,
+    remote_master_sha: str | None = None,
+    check_root_checkout: bool = True,
+    check_worktree: bool = True,
+    # stale runtime (master parity)
+    startup_head: str | None = None,
+    current_code_head: str | None = None,
+    check_stale_runtime: bool = True,
+    # lease ownership
+    lease_required: bool = False,
+    foreign_lease: bool | None = None,
+    lease_owner_session: str | None = None,
+    active_session_id: str | None = None,
+    lease_owner_identity: str | None = None,
+    active_identity: str | None = None,
+    lease_reasons: list[str] | None = None,
+    # terminal lock
+    terminal_lock_blocks: bool | None = None,
+    terminal_lock_reasons: list[str] | None = None,
+    # expected head SHA
+    require_head_sha: bool = False,
+    expected_head_sha: str | None = None,
+    live_head_sha: str | None = None,
+    # workflow hash
+    workflow_hash_valid: bool | None = None,
+    workflow_hash_reasons: list[str] | None = None,
+    # source contamination (stable-branch push / approval contamination)
+    source_contaminated: bool | None = None,
+    contamination_reasons: list[str] | None = None,
+    # manual bypass
+    manual_bypass_attempted: bool = False,
+    manual_bypass_reasons: list[str] | None = None,
+) -> dict[str, Any]:
+    """Fail-closed common anti-stomp assessment (pure).
+
+    Only checks for which facts are supplied (or which are required by the
+    mutation category) are evaluated. Unsupplied optional facts are recorded
+    as ``skipped`` so callers can see coverage without inventing defaults.
+
+    Returns a structured dict with ``allowed``/``block``, typed ``blockers``,
+    aggregated ``reasons``, ``exact_next_action``, and per-check ``checks``.
+    """
+    task_name = (task or "").strip()
+    role = (profile_role or "").strip().lower() or None
+    req_role = (required_role or "").strip().lower() or None
+    checks: dict[str, Any] = {
+        "task": task_name or None,
+        "profile_name": profile_name,
+        "profile_role": role,
+        "required_role": req_role,
+    }
+    blockers: list[dict[str, Any]] = []
+
+    # ── manual bypass (always first; never skippable when attempted) ─────────
+    if manual_bypass_attempted:
+        reasons = list(manual_bypass_reasons or [
+            "manual state/mtime/source-file bypass attempt detected (fail closed)"
+        ])
+        blockers.append(_blocker(BLOCKER_MANUAL_BYPASS, reasons))
+        checks["manual_bypass"] = {"block": True, "reasons": reasons}
+    else:
+        checks["manual_bypass"] = {"block": False, "skipped": False}
+
+    # ── repo/org ─────────────────────────────────────────────────────────────
+    if check_repo and resolved_org is not None and resolved_repo is not None:
+        repo_assessment = remote_repo_guard.assess_remote_repo_match(
+            remote=remote or "",
+            resolved_org=resolved_org,
+            resolved_repo=resolved_repo,
+            local_remote_url=local_remote_url,
+            org_explicit=org_explicit,
+            repo_explicit=repo_explicit,
+        )
+        checks["repo"] = {
+            "block": bool(repo_assessment.get("block")),
+            "reasons": list(repo_assessment.get("reasons") or []),
+            "resolved_org": resolved_org,
+            "resolved_repo": resolved_repo,
+        }
+        if repo_assessment.get("block"):
+            blockers.append(
+                _blocker(
+                    BLOCKER_WRONG_REPO,
+                    list(repo_assessment.get("reasons") or ["repo/org mismatch"]),
+                    detail={
+                        "resolved_org": resolved_org,
+                        "resolved_repo": resolved_repo,
+                        "local_remote_url": local_remote_url,
+                    },
+                )
+            )
+    else:
+        checks["repo"] = {"block": False, "skipped": True}
+
+    # ── profile/role ─────────────────────────────────────────────────────────
+    if check_role and req_role and role:
+        if not roles_compatible(role, req_role):
+            reasons = [
+                f"active profile role '{role}' does not match required role "
+                f"'{req_role}' for task '{task_name or '(unknown)'}' "
+                f"(profile={profile_name or '(unknown)'})"
+            ]
+            checks["role"] = {"block": True, "reasons": reasons}
+            blockers.append(
+                _blocker(
+                    BLOCKER_WRONG_ROLE,
+                    reasons,
+                    detail={
+                        "profile_name": profile_name,
+                        "profile_role": role,
+                        "required_role": req_role,
+                    },
+                )
+            )
+        else:
+            checks["role"] = {"block": False, "reasons": []}
+    else:
+        checks["role"] = {"block": False, "skipped": True}
+
+    # ── stale runtime (master parity) ────────────────────────────────────────
+    if check_stale_runtime and (
+        startup_head is not None or current_code_head is not None
+    ):
+        parity = master_parity_gate.assess_master_parity(
+            {"startup_head": startup_head},
+            current_code_head,
+        )
+        stale_reasons = master_parity_gate.parity_block_reasons(parity)
+        checks["stale_runtime"] = {
+            "block": bool(stale_reasons),
+            "reasons": list(stale_reasons),
+            "startup_head": startup_head,
+            "current_code_head": current_code_head,
+            "stale": bool(parity.get("stale")),
+        }
+        if stale_reasons:
+            blockers.append(
+                _blocker(
+                    BLOCKER_STALE_RUNTIME,
+                    list(stale_reasons),
+                    detail={
+                        "startup_head": startup_head,
+                        "current_code_head": current_code_head,
+                    },
+                )
+            )
+    else:
+        checks["stale_runtime"] = {"block": False, "skipped": True}
+
+    # ── root checkout ────────────────────────────────────────────────────────
+    if (
+        check_root_checkout
+        and workspace_path is not None
+        and project_root is not None
+        and root_porcelain is not None
+    ):
+        root_assessment = root_checkout_guard.assess_root_checkout_guard(
+            workspace_path=workspace_path,
+            canonical_repo_root=project_root,
+            current_branch=current_branch,
+            head_sha=root_head_sha,
+            porcelain_status=root_porcelain,
+            remote_master_sha=remote_master_sha,
+            resolved_role=req_role or role,
+            actual_role=role,
+        )
+        checks["root_checkout"] = {
+            "block": bool(root_assessment.get("block")),
+            "reasons": list(root_assessment.get("reasons") or []),
+        }
+        if root_assessment.get("block"):
+            blockers.append(
+                _blocker(
+                    BLOCKER_ROOT_CHECKOUT,
+                    list(root_assessment.get("reasons") or [
+                        "control checkout is not clean master"
+                    ]),
+                    detail={
+                        "workspace_path": workspace_path,
+                        "project_root": project_root,
+                        "current_branch": current_branch,
+                    },
+                )
+            )
+    else:
+        checks["root_checkout"] = {"block": False, "skipped": True}
+
+    # ── worktree under branches (author) ─────────────────────────────────────
+    worktree_role = role or req_role
+    if (
+        check_worktree
+        and workspace_path is not None
+        and project_root is not None
+        and worktree_role in _BRANCHES_REQUIRED_ROLES
+    ):
+        wt = author_mutation_worktree.assess_author_mutation_worktree(
+            workspace_path=workspace_path,
+            project_root=project_root,
+            current_branch=current_branch,
+        )
+        checks["worktree"] = {
+            "block": bool(wt.get("block")),
+            "reasons": list(wt.get("reasons") or []),
+            "under_branches": wt.get("under_branches"),
+        }
+        if wt.get("block"):
+            blockers.append(
+                _blocker(
+                    BLOCKER_WRONG_WORKTREE,
+                    list(wt.get("reasons") or [
+                        "mutation worktree is not under branches/"
+                    ]),
+                    detail={
+                        "workspace_path": workspace_path,
+                        "project_root": project_root,
+                    },
+                )
+            )
+    else:
+        checks["worktree"] = {
+            "block": False,
+            "skipped": worktree_role not in _BRANCHES_REQUIRED_ROLES
+            or workspace_path is None,
+        }
+
+    # ── foreign lease ────────────────────────────────────────────────────────
+    lease_needed = lease_required or (
+        task_name.removeprefix("gitea_") in {
+            t.removeprefix("gitea_") for t in _LEASE_AWARE_TASKS
+        }
+        and foreign_lease is not None
+    )
+    if lease_needed or foreign_lease is True:
+        is_foreign = bool(foreign_lease)
+        if foreign_lease is None and lease_required:
+            # Ownership must be proven when lease_required; missing ownership
+            # facts fail closed.
+            owner_ok = (
+                (not lease_owner_session and not active_session_id)
+                or (
+                    lease_owner_session
+                    and active_session_id
+                    and lease_owner_session == active_session_id
+                )
+            )
+            identity_ok = (
+                (not lease_owner_identity and not active_identity)
+                or (
+                    lease_owner_identity
+                    and active_identity
+                    and lease_owner_identity == active_identity
+                )
+            )
+            if not owner_ok or not identity_ok:
+                is_foreign = True
+        reasons = list(lease_reasons or [])
+        if is_foreign and not reasons:
+            reasons = [
+                "active lease is owned by a foreign session or identity "
+                "(anti-stomp fail closed)"
+            ]
+        checks["lease"] = {
+            "block": is_foreign,
+            "reasons": reasons if is_foreign else [],
+            "lease_owner_session": lease_owner_session,
+            "active_session_id": active_session_id,
+        }
+        if is_foreign:
+            blockers.append(
+                _blocker(BLOCKER_FOREIGN_LEASE, reasons)
+            )
+    else:
+        checks["lease"] = {"block": False, "skipped": True}
+
+    # ── terminal lock ────────────────────────────────────────────────────────
+    if terminal_lock_blocks is not None:
+        reasons = list(terminal_lock_reasons or [])
+        if terminal_lock_blocks and not reasons:
+            reasons = [
+                "terminal review-decision lock is active for this head "
+                "(anti-stomp fail closed)"
+            ]
+        checks["terminal_lock"] = {
+            "block": bool(terminal_lock_blocks),
+            "reasons": reasons if terminal_lock_blocks else [],
+        }
+        if terminal_lock_blocks:
+            blockers.append(_blocker(BLOCKER_TERMINAL_LOCK, reasons))
+    else:
+        checks["terminal_lock"] = {"block": False, "skipped": True}
+
+    # ── expected head SHA ────────────────────────────────────────────────────
+    if require_head_sha or (
+        expected_head_sha is not None and live_head_sha is not None
+    ):
+        exp = (expected_head_sha or "").strip().lower()
+        live = (live_head_sha or "").strip().lower()
+        mismatch = False
+        reasons: list[str] = []
+        if require_head_sha and not exp:
+            mismatch = True
+            reasons.append(
+                "expected_head_sha is required before this mutation "
+                "(anti-stomp fail closed)"
+            )
+        elif exp and live and exp != live:
+            mismatch = True
+            reasons.append(
+                f"expected_head_sha '{exp}' does not match live PR head "
+                f"'{live}' (anti-stomp fail closed; stale prompt data)"
+            )
+        elif require_head_sha and exp and not live:
+            mismatch = True
+            reasons.append(
+                "live PR head SHA could not be determined to corroborate "
+                "expected_head_sha (anti-stomp fail closed)"
+            )
+        checks["head_sha"] = {
+            "block": mismatch,
+            "reasons": reasons,
+            "expected_head_sha": expected_head_sha,
+            "live_head_sha": live_head_sha,
+        }
+        if mismatch:
+            blockers.append(_blocker(BLOCKER_HEAD_SHA, reasons))
+    else:
+        checks["head_sha"] = {"block": False, "skipped": True}
+
+    # ── workflow hash ────────────────────────────────────────────────────────
+    if workflow_hash_valid is not None:
+        reasons = list(workflow_hash_reasons or [])
+        if not workflow_hash_valid and not reasons:
+            reasons = [
+                "workflow load proof missing or hash stale "
+                "(anti-stomp fail closed)"
+            ]
+        checks["workflow_hash"] = {
+            "block": not bool(workflow_hash_valid),
+            "reasons": reasons if not workflow_hash_valid else [],
+        }
+        if not workflow_hash_valid:
+            blockers.append(_blocker(BLOCKER_WORKFLOW_HASH, reasons))
+    else:
+        checks["workflow_hash"] = {"block": False, "skipped": True}
+
+    # ── source contamination ─────────────────────────────────────────────────
+    if source_contaminated is not None:
+        reasons = list(contamination_reasons or [])
+        if source_contaminated and not reasons:
+            reasons = [
+                "session is source-contaminated (stable-branch push or "
+                "contaminated approval path); anti-stomp fail closed"
+            ]
+        checks["source_contamination"] = {
+            "block": bool(source_contaminated),
+            "reasons": reasons if source_contaminated else [],
+        }
+        if source_contaminated:
+            blockers.append(
+                _blocker(BLOCKER_SOURCE_CONTAMINATION, reasons)
+            )
+    else:
+        checks["source_contamination"] = {"block": False, "skipped": True}
+
+    if blockers:
+        return _blocked_result(blockers, checks)
+    return _allowed_result(checks)
+
+
+def format_anti_stomp_error(assessment: dict[str, Any]) -> str:
+    """Single RuntimeError message for MCP mutation gates."""
+    kind = assessment.get("blocker_kind") or "unknown"
+    reasons = "; ".join(
+        assessment.get("reasons")
+        or ["anti-stomp preflight blocked the mutation"]
+    )
+    next_action = assessment.get("exact_next_action") or "stop and diagnose"
+    return (
+        f"Anti-stomp preflight (#604) blocked mutation "
+        f"[{kind}]: {reasons}. "
+        f"exact_next_action: {next_action}"
+    )
+
+
+def block_response(
+    assessment: dict[str, Any],
+    **extra_fields: Any,
+) -> dict[str, Any]:
+    """Structured tool-return payload for a blocked mutation."""
+    payload = {
+        "success": False,
+        "performed": False,
+        "blocked": True,
+        "anti_stomp": True,
+        "blocker_kind": assessment.get("blocker_kind"),
+        "blockers": list(assessment.get("blockers") or []),
+        "reasons": list(assessment.get("reasons") or []),
+        "exact_next_action": assessment.get("exact_next_action"),
+        "checks": assessment.get("checks") or {},
+    }
+    payload.update(extra_fields)
+    return payload
+
+
+def contamination_from_stable_marker(
+    marker: dict | None,
+    *,
+    task: str | None,
+    actual_role: str | None,
+) -> tuple[bool, list[str]]:
+    """Derive source-contamination facts from a #671 durable marker."""
+    if not marker:
+        return False, []
+    gate = stable_branch_push_guard.assess_contamination_gate(
+        marker,
+        task=task,
+        actual_role=actual_role,
+    )
+    if gate.get("block"):
+        return True, list(gate.get("reasons") or [])
+    return False, []

gitea_mcp_server.py

@@ -619,18 +619,212 @@ 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()
+    req_role = None
+    if task:
+        try:
+            req_role = task_capability_map.required_role(task)
+        except KeyError:
+            req_role = _preflight_resolved_role
+
+    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",
+        "mark_final_review_decision",
+    }:
+        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,
+        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,
     task: 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,
+    org: str | None = None,
+    repo: str | None = None,
 ):
-    """Verify that identity and capability were verified prior to session edits."""
+    """Verify that identity and capability were verified prior to session edits.
+
+    Also runs the shared #604 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
 
     in_test = _preflight_in_test_mode()
     if in_test and not (
         os.environ.get("GITEA_TEST_FORCE_DIRTY")
         or os.environ.get("GITEA_TEST_PORCELAIN") is not None
+        or os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP")
     ):
         return
 
@@ -717,6 +911,30 @@ def verify_preflight_purity(
 
     _enforce_root_checkout_guard(worktree_path)
     _enforce_branches_only_author_mutation(worktree_path)
+
+    # #604: common anti-stomp preflight (typed blocker + exact next action).
+    # Runs after the legacy enforcers so existing happy paths stay green and
+    # the shared module remains the single fail-closed composition point for
+    # repo/role/worktree/lease/terminal-lock/head/stale/workflow/contamination.
+    _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()
 
 
@@ -930,6 +1148,7 @@ import author_mutation_worktree  # noqa: E402
 import root_checkout_guard  # noqa: E402
 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
@@ -3833,6 +4052,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"]:
@@ -5269,6 +5512,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)"

tests/test_anti_stomp_preflight.py

@@ -0,0 +1,566 @@
+"""Regression coverage for the common anti-stomp preflight (#604).
+
+Acceptance criteria:
+
+1. All mutation tools call the common preflight (wired via
+   ``verify_preflight_purity`` / ``_run_anti_stomp_preflight``).
+2. Failure returns a typed blocker and exact next action.
+3. Tests cover wrong repo defaulting to Timesheet, stale runtime, wrong
+   worktree, foreign lease, terminal lock, and contaminated approval.
+4. No mutation can proceed using stale prompt data if live state disagrees.
+5. Existing successful paths continue to work (happy-path assessment).
+"""
+
+from __future__ import annotations
+
+import os
+import unittest
+from unittest import mock
+
+import anti_stomp_preflight as asp
+
+
+LOCAL_GITEA_TOOLS_URL = "https://gitea.prgs.cc/Scaled-Tech-Consulting/Gitea-Tools.git"
+STARTUP = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+ADVANCED = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"
+HEAD_A = "1111111111111111111111111111111111111111"
+HEAD_B = "2222222222222222222222222222222222222222"
+
+
+def _happy_kwargs(**overrides):
+    base = dict(
+        task="create_issue",
+        remote="prgs",
+        resolved_org="Scaled-Tech-Consulting",
+        resolved_repo="Gitea-Tools",
+        local_remote_url=LOCAL_GITEA_TOOLS_URL,
+        org_explicit=True,
+        repo_explicit=True,
+        profile_name="prgs-author",
+        profile_role="author",
+        required_role="author",
+        workspace_path="/repo/branches/issue-604",
+        project_root="/repo",
+        current_branch="feat/issue-604-anti-stomp-preflight",
+        root_head_sha=STARTUP,
+        root_porcelain="",
+        remote_master_sha=STARTUP,
+        startup_head=STARTUP,
+        current_code_head=STARTUP,
+        source_contaminated=False,
+        manual_bypass_attempted=False,
+    )
+    base.update(overrides)
+    return base
+
+
+class TestIsMutationTask(unittest.TestCase):
+    def test_core_mutation_tasks(self):
+        for task in (
+            "create_issue",
+            "comment_issue",
+            "set_issue_labels",
+            "acquire_reviewer_pr_lease",
+            "submit_pr_review",
+            "approve_pr",
+            "request_changes_pr",
+            "merge_pr",
+            "cleanup_stale_claims",
+            "delete_branch",
+        ):
+            self.assertTrue(asp.is_mutation_task(task), task)
+
+    def test_gitea_prefix_accepted(self):
+        self.assertTrue(asp.is_mutation_task("gitea_create_issue"))
+        self.assertTrue(asp.is_mutation_task("gitea_merge_pr"))
+
+    def test_read_only_not_mutation(self):
+        self.assertFalse(asp.is_mutation_task("list_prs"))
+        self.assertFalse(asp.is_mutation_task("whoami"))
+        self.assertFalse(asp.is_mutation_task(""))
+        self.assertFalse(asp.is_mutation_task(None))
+
+
+class TestHappyPath(unittest.TestCase):
+    def test_allowed_when_all_checks_pass(self):
+        result = asp.assess_anti_stomp_preflight(**_happy_kwargs())
+        self.assertTrue(result["allowed"])
+        self.assertFalse(result["block"])
+        self.assertEqual(result["blockers"], [])
+        self.assertEqual(result["exact_next_action"], "proceed")
+        self.assertIsNone(result["blocker_kind"])
+
+    def test_reviewer_happy_path_skips_author_worktree(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="review_pr",
+                profile_name="prgs-reviewer",
+                profile_role="reviewer",
+                required_role="reviewer",
+                # Under branches/ the root guard short-circuits for non-merger.
+                workspace_path="/repo/branches/review-pr-42",
+                project_root="/repo",
+                current_branch="review/pr-42",
+                foreign_lease=False,
+                terminal_lock_blocks=False,
+                expected_head_sha=HEAD_A,
+                live_head_sha=HEAD_A,
+                workflow_hash_valid=True,
+            )
+        )
+        self.assertTrue(result["allowed"], result.get("reasons"))
+        self.assertTrue(result["checks"]["worktree"].get("skipped"))
+
+
+class TestWrongRepoTimesheet(unittest.TestCase):
+    """AC3: wrong repo defaulting to Timesheet."""
+
+    def test_prgs_default_timesheet_vs_local_gitea_tools(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                resolved_org="Scaled-Tech-Consulting",
+                resolved_repo="Timesheet",
+                local_remote_url=LOCAL_GITEA_TOOLS_URL,
+                org_explicit=False,
+                repo_explicit=False,
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_REPO)
+        self.assertTrue(result["exact_next_action"])
+        self.assertIn("org=", result["exact_next_action"])
+        self.assertTrue(
+            any("Timesheet" in r or "does not match" in r for r in result["reasons"])
+        )
+
+    def test_explicit_org_repo_skips_mismatch(self):
+        # Explicit intent is authoritative (remote_repo_guard contract).
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                resolved_repo="Timesheet",
+                org_explicit=True,
+                repo_explicit=True,
+            )
+        )
+        self.assertTrue(result["allowed"])
+
+
+class TestStaleRuntime(unittest.TestCase):
+    """AC3: stale runtime."""
+
+    def test_stale_runtime_blocks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                startup_head=STARTUP,
+                current_code_head=ADVANCED,
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_STALE_RUNTIME)
+        self.assertIn("Restart", result["exact_next_action"])
+        self.assertTrue(result["checks"]["stale_runtime"]["stale"])
+
+    def test_in_parity_allows(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                startup_head=STARTUP,
+                current_code_head=STARTUP,
+            )
+        )
+        self.assertTrue(result["allowed"])
+
+
+class TestWrongWorktree(unittest.TestCase):
+    """AC3: wrong worktree."""
+
+    def test_author_on_control_checkout_blocked(self):
+        # Clean master control checkout: root guard may pass for a clean master
+        # HEAD, but the author worktree guard must still refuse mutation from
+        # outside branches/.
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                workspace_path="/repo",
+                project_root="/repo",
+                current_branch="master",
+                root_porcelain="",
+                root_head_sha=STARTUP,
+                remote_master_sha=STARTUP,
+                profile_role="author",
+                required_role="author",
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_WORKTREE)
+        self.assertIn("branches/", result["exact_next_action"])
+
+    def test_author_under_branches_allowed(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                workspace_path="/repo/branches/issue-604",
+                project_root="/repo",
+            )
+        )
+        self.assertTrue(result["allowed"])
+
+
+class TestForeignLease(unittest.TestCase):
+    """AC3: foreign lease."""
+
+    def test_foreign_lease_blocks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="merge_pr",
+                profile_role="merger",
+                required_role="merger",
+                # Merger is not auto-exempted under branches/; pass clean master.
+                workspace_path="/repo",
+                project_root="/repo",
+                current_branch="master",
+                root_porcelain="",
+                foreign_lease=True,
+                lease_reasons=["lease owned by other-session"],
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_FOREIGN_LEASE)
+        self.assertIn("foreign lease", result["exact_next_action"].lower())
+
+    def test_lease_required_without_ownership_fails_closed(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="review_pr",
+                profile_role="reviewer",
+                required_role="reviewer",
+                workspace_path="/repo/branches/review-pr-1",
+                project_root="/repo",
+                lease_required=True,
+                lease_owner_session="sess-A",
+                active_session_id="sess-B",
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_FOREIGN_LEASE)
+
+    def test_owned_lease_allows(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="review_pr",
+                profile_role="reviewer",
+                required_role="reviewer",
+                workspace_path="/repo/branches/review-pr-1",
+                project_root="/repo",
+                foreign_lease=False,
+            )
+        )
+        self.assertTrue(result["allowed"])
+
+
+class TestTerminalLock(unittest.TestCase):
+    """AC3: terminal lock."""
+
+    def test_terminal_lock_blocks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="approve_pr",
+                profile_role="reviewer",
+                required_role="reviewer",
+                workspace_path="/repo/branches/review-pr-9",
+                project_root="/repo",
+                terminal_lock_blocks=True,
+                terminal_lock_reasons=["#332 terminal lock active for this head"],
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_TERMINAL_LOCK)
+        self.assertIn("#332", result["exact_next_action"])
+
+    def test_no_terminal_lock_allows(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="approve_pr",
+                profile_role="reviewer",
+                required_role="reviewer",
+                workspace_path="/repo/branches/review-pr-9",
+                project_root="/repo",
+                terminal_lock_blocks=False,
+            )
+        )
+        self.assertTrue(result["allowed"])
+
+
+class TestHeadShaStalePrompt(unittest.TestCase):
+    """AC4: no mutation with stale prompt head SHA."""
+
+    def _merger_kwargs(self, **overrides):
+        base = _happy_kwargs(
+            task="merge_pr",
+            profile_role="merger",
+            required_role="merger",
+            workspace_path="/repo",
+            project_root="/repo",
+            current_branch="master",
+            root_porcelain="",
+            root_head_sha=STARTUP,
+            remote_master_sha=STARTUP,
+        )
+        base.update(overrides)
+        return base
+
+    def test_head_mismatch_blocks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **self._merger_kwargs(
+                expected_head_sha=HEAD_A,
+                live_head_sha=HEAD_B,
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_HEAD_SHA)
+        self.assertIn("stale prompt", " ".join(result["reasons"]).lower())
+
+    def test_require_head_sha_missing_blocks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **self._merger_kwargs(
+                require_head_sha=True,
+                expected_head_sha=None,
+                live_head_sha=HEAD_A,
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_HEAD_SHA)
+
+    def test_matching_head_allows(self):
+        result = asp.assess_anti_stomp_preflight(
+            **self._merger_kwargs(
+                expected_head_sha=HEAD_A,
+                live_head_sha=HEAD_A,
+            )
+        )
+        self.assertTrue(result["allowed"])
+
+
+class TestContaminatedApproval(unittest.TestCase):
+    """AC3: contaminated approval / source contamination."""
+
+    def test_source_contamination_blocks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="merge_pr",
+                profile_role="merger",
+                required_role="merger",
+                workspace_path="/repo",
+                project_root="/repo",
+                current_branch="master",
+                root_porcelain="",
+                source_contaminated=True,
+                contamination_reasons=[
+                    "session contaminated by direct stable-branch push attempt"
+                ],
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_SOURCE_CONTAMINATION)
+        self.assertIn("reconciler", result["exact_next_action"].lower())
+
+    def test_manual_bypass_blocks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                manual_bypass_attempted=True,
+                manual_bypass_reasons=[
+                    "attempted manual deletion of session-state lock files"
+                ],
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_MANUAL_BYPASS)
+
+
+class TestWrongRole(unittest.TestCase):
+    def test_author_cannot_merge(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="merge_pr",
+                profile_role="author",
+                required_role="merger",
+                workspace_path="/repo/branches/issue-604",
+                project_root="/repo",
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE)
+
+    def test_reconciler_may_run_author_class_tasks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="comment_issue",
+                profile_name="prgs-reconciler",
+                profile_role="reconciler",
+                required_role="author",
+                workspace_path="/repo",
+                project_root="/repo",
+                current_branch="master",
+                root_porcelain="",
+            )
+        )
+        self.assertTrue(result["allowed"], result.get("reasons"))
+        self.assertTrue(asp.roles_compatible("reconciler", "author"))
+        self.assertFalse(asp.roles_compatible("author", "merger"))
+
+
+class TestWorkflowHash(unittest.TestCase):
+    def test_stale_workflow_hash_blocks(self):
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="review_pr",
+                profile_role="reviewer",
+                required_role="reviewer",
+                workspace_path="/repo/branches/review-pr-3",
+                project_root="/repo",
+                workflow_hash_valid=False,
+                workflow_hash_reasons=["stored workflow hash is stale"],
+            )
+        )
+        self.assertTrue(result["block"])
+        self.assertEqual(result["blocker_kind"], asp.BLOCKER_WORKFLOW_HASH)
+
+
+class TestTypedBlockerResponse(unittest.TestCase):
+    """AC2: typed blocker + exact next action in response payload."""
+
+    def test_block_response_shape(self):
+        assessment = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                task="review_pr",
+                profile_role="reviewer",
+                required_role="reviewer",
+                workspace_path="/repo/branches/review-pr-7",
+                project_root="/repo",
+                foreign_lease=True,
+            )
+        )
+        payload = asp.block_response(assessment, pr_number=42)
+        self.assertFalse(payload["success"])
+        self.assertFalse(payload["performed"])
+        self.assertTrue(payload["blocked"])
+        self.assertTrue(payload["anti_stomp"])
+        self.assertEqual(payload["blocker_kind"], asp.BLOCKER_FOREIGN_LEASE)
+        self.assertTrue(payload["exact_next_action"])
+        self.assertEqual(payload["pr_number"], 42)
+        self.assertTrue(payload["blockers"])
+        self.assertEqual(payload["blockers"][0]["kind"], asp.BLOCKER_FOREIGN_LEASE)
+
+    def test_format_error_includes_kind_and_next_action(self):
+        assessment = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                startup_head=STARTUP,
+                current_code_head=ADVANCED,
+            )
+        )
+        msg = asp.format_anti_stomp_error(assessment)
+        self.assertIn("#604", msg)
+        self.assertIn(asp.BLOCKER_STALE_RUNTIME, msg)
+        self.assertIn("exact_next_action", msg)
+
+
+class TestRootCheckoutContamination(unittest.TestCase):
+    def test_dirty_control_checkout_blocks_author(self):
+        # Workspace is control checkout with dirty porcelain.
+        result = asp.assess_anti_stomp_preflight(
+            **_happy_kwargs(
+                workspace_path="/repo",
+                project_root="/repo",
+                current_branch="master",
+                root_porcelain=" M gitea_mcp_server.py\n",
+                root_head_sha=STARTUP,
+                remote_master_sha=STARTUP,
+            )
+        )
+        self.assertTrue(result["block"])
+        # Author worktree check or root checkout may fire first.
+        self.assertIn(
+            result["blocker_kind"],
+            {asp.BLOCKER_ROOT_CHECKOUT, asp.BLOCKER_WRONG_WORKTREE},
+        )
+
+
+class TestMutationTaskInventory(unittest.TestCase):
+    """AC1: mutation task set covers issue-listed paths."""
+
+    def test_issue_required_paths_covered(self):
+        required = {
+            "create_issue",
+            "comment_issue",
+            "set_issue_labels",
+            "acquire_reviewer_pr_lease",
+            "submit_pr_review",
+            "approve_pr",
+            "request_changes_pr",
+            "merge_pr",
+            "cleanup_merged_pr_branch",
+            "cleanup_stale_claims",
+        }
+        missing = required - asp.MUTATION_TASKS
+        self.assertFalse(missing, f"missing mutation tasks: {missing}")
+
+
+class TestServerWiring(unittest.TestCase):
+    """Smoke: MCP server imports anti_stomp and exposes the runner."""
+
+    def test_server_imports_anti_stomp(self):
+        import gitea_mcp_server as server
+
+        self.assertTrue(hasattr(server, "_run_anti_stomp_preflight"))
+        self.assertTrue(hasattr(server, "anti_stomp_preflight"))
+        self.assertIs(server.anti_stomp_preflight, asp)
+
+    def test_runner_skipped_under_pytest_by_default(self):
+        import gitea_mcp_server as server
+
+        # Default pytest path must not raise (suite isolation).
+        self.assertIsNone(
+            server._run_anti_stomp_preflight(
+                "create_issue",
+                remote="prgs",
+                org="Scaled-Tech-Consulting",
+                repo="Timesheet",
+            )
+        )
+
+    def test_runner_blocks_when_forced(self):
+        import gitea_mcp_server as server
+
+        with mock.patch.dict(
+            os.environ,
+            {"GITEA_TEST_FORCE_ANTI_STOMP": "1"},
+            clear=False,
+        ):
+            # Force assessor to return a block without full git/workspace setup.
+            blocked = {
+                "allowed": False,
+                "block": True,
+                "blockers": [{
+                    "kind": asp.BLOCKER_STALE_RUNTIME,
+                    "reasons": ["stale"],
+                    "exact_next_action": "restart",
+                    "detail": {},
+                }],
+                "reasons": ["stale"],
+                "exact_next_action": "restart",
+                "blocker_kind": asp.BLOCKER_STALE_RUNTIME,
+                "checks": {},
+            }
+            with mock.patch.object(
+                asp, "assess_anti_stomp_preflight", return_value=blocked
+            ):
+                with self.assertRaises(RuntimeError) as ctx:
+                    server._run_anti_stomp_preflight(
+                        "create_issue",
+                        remote="prgs",
+                        org="Scaled-Tech-Consulting",
+                        repo="Gitea-Tools",
+                    )
+                self.assertIn("#604", str(ctx.exception))
+                self.assertIn(asp.BLOCKER_STALE_RUNTIME, str(ctx.exception))
+
+
+if __name__ == "__main__":
+    unittest.main()