"""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 the shared anti-stomp preflight before # acting (issue #604 AC1). Every member must appear as a live task= kwarg # to verify_preflight_purity / _run_anti_stomp_preflight (or the review # anti_task map) in gitea_mcp_server.py. Inventory↔wiring tests fail if # a declared task is not wired. MUTATION_TASKS = frozenset({ "create_issue", "comment_issue", "close_issue", "mark_issue", "lock_issue", "set_issue_labels", "create_label", "create_pr", "close_pr", "edit_pr", "commit_files", "gitea_commit_files", "delete_branch", "cleanup_merged_pr_branch", "cleanup_stale_claims", "reconcile_merged_cleanups", "reconcile_already_landed_pr", "reconcile_close_superseded_pr", "post_heartbeat", "acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease", "adopt_merger_pr_lease", "review_pr", "submit_pr_review", "approve_pr", "request_changes_pr", "merge_pr", }) # Intentionally excluded from MUTATION_TASKS. Each has a dedicated # fail-closed gate that preserves decision-lock ownership, workflow-hash, # head-SHA, and repository consistency. Do not re-add without either # wiring shared preflight or updating this rationale (#604 Blocker B). DEDICATED_GATE_MUTATIONS: dict[str, str] = { "mark_final_review_decision": ( "Local review-decision lock only. Enforced by session lock ownership, " "workflow-hash, expected head SHA, PR work lease, eligibility, and " "terminal-lock gates. No Gitea review POST; shared anti-stomp would " "duplicate without side-effect-ordering value." ), "save_review_draft": ( "Local session-state draft save with live PR head/base consistency " "checks. Not a Gitea review/merge mutation; dedicated head-SHA and " "worktree resolution gates apply." ), "resume_review_draft": ( "Live-state resume with terminal lock, lease ownership, head/base SHA, " "and parity checks. When submit=True, delegates to gitea_submit_pr_review " "which runs shared anti-stomp before the Gitea review POST." ), "cleanup_stale_review_decision_lock": ( "Moot-lock cleanup with identity match, live PR merged/closed proof, " "and reviewer capability gate (#594). Distinct from shared anti-stomp." ), "gitea_cleanup_stale_review_decision_lock": ( "Alias of cleanup_stale_review_decision_lock; dedicated #594 path." ), "heartbeat_reviewer_pr_lease": ( "Lease heartbeat posts via verify_preflight_purity(task='review_pr') " "plus in-session lease ownership checks; not a separate mutation class." ), "release_reviewer_pr_lease": ( "Lease release uses workspace binding + ownership checks; posts only " "when the session owns the active lease." ), } # 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", "acquire_reviewer_pr_lease", "gitea_acquire_reviewer_pr_lease", "adopt_merger_pr_lease", }) _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. Capability-authorized multi-role tasks (e.g. reviewer holding ``gitea.issue.comment`` for ``comment_issue``) are handled by :func:`authorization_compatible`, not by expanding this role matrix. """ 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 authorization_compatible( active_role: str | None, required_role: str | None, *, required_permission: str | None = None, allowed_operations: Any = None, ) -> bool: """Whether the active session may run a task given role *and* capability. Order: 1. :func:`roles_compatible` (exact role or narrow reconciler→author). 2. Capability possession: when *required_permission* is non-empty and present in *allowed_operations*, allow even if the nominal task role differs (e.g. reviewer/merger with ``gitea.issue.comment`` on ``comment_issue`` / ``mark_issue`` / ``set_issue_labels`` / ``lock_issue``). Fail closed otherwise. This is **not** a broad reviewer→author or merger→author role rewrite: without the specific permission the check still denies. Missing *allowed_operations* when a permission is required also fails closed (callers must supply the live profile op list). """ if roles_compatible(active_role, required_role): return True perm = (required_permission or "").strip() if not perm: return False if allowed_operations is None: return False try: ops = {str(o).strip() for o in allowed_operations if o is not None} except TypeError: return False return perm in ops 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 + capability profile_name: str | None = None, profile_role: str | None = None, required_role: str | None = None, required_permission: str | None = None, allowed_operations: Any = 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 req_perm = (required_permission or "").strip() or None checks: dict[str, Any] = { "task": task_name or None, "profile_name": profile_name, "profile_role": role, "required_role": req_role, "required_permission": req_perm, } 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 + capability ──────────────────────────────────────────── # Role match OR possession of the task's required permission (Blocker A). # Reviewer/merger may run issue-comment-class tasks when they hold # gitea.issue.comment; unauthorized escalation without the permission # still fails closed. if check_role and req_role and role: authorized = authorization_compatible( role, req_role, required_permission=req_perm, allowed_operations=allowed_operations, ) if not authorized: perm_clause = ( f", required_permission='{req_perm}'" if req_perm else "" ) reasons = [ f"active profile role '{role}' is not authorized for task " f"'{task_name or '(unknown)'}' (required_role='{req_role}'" f"{perm_clause}; profile={profile_name or '(unknown)'})" ] checks["role"] = { "block": True, "reasons": reasons, "required_permission": req_perm, "capability_authorized": False, } blockers.append( _blocker( BLOCKER_WRONG_ROLE, reasons, detail={ "profile_name": profile_name, "profile_role": role, "required_role": req_role, "required_permission": req_perm, }, ) ) else: checks["role"] = { "block": False, "reasons": [], "required_permission": req_perm, "capability_authorized": bool( req_perm and allowed_operations is not None and req_perm in { str(o).strip() for o in (allowed_operations or []) if o is not None } and not roles_compatible(role, req_role) ), } 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, []