From ea8e186549d4a21e7e3e8f41261ff527b6a60e97 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 12 Jul 2026 05:10:30 -0400 Subject: [PATCH 1/2] 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 | 705 +++++++++++++++++++++++++++++ gitea_mcp_server.py | 264 ++++++++++- tests/test_anti_stomp_preflight.py | 566 +++++++++++++++++++++++ 3 files changed, 1534 insertions(+), 1 deletion(-) create mode 100644 anti_stomp_preflight.py create mode 100644 tests/test_anti_stomp_preflight.py diff --git a/anti_stomp_preflight.py b/anti_stomp_preflight.py new file mode 100644 index 0000000..bdc0f96 --- /dev/null +++ b/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, [] diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 669e9aa..d1b587f 100644 --- a/gitea_mcp_server.py +++ b/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)" diff --git a/tests/test_anti_stomp_preflight.py b/tests/test_anti_stomp_preflight.py new file mode 100644 index 0000000..cf53177 --- /dev/null +++ b/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() -- 2.43.7 From 3fd02a3c633fb539850ce202c68b2043cbf7e4ca Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Sun, 12 Jul 2026 09:00:45 -0400 Subject: [PATCH 2/2] fix: anti-stomp capability auth, inventory wiring, side-effect proof (#604) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address REQUEST_CHANGES on PR #680: Blocker A — role vs capability agreement - authorization_compatible allows reviewer/merger when they hold the task's required permission (e.g. gitea.issue.comment on comment_issue/mark_issue/ set_issue_labels/lock_issue) without broad role-bypass. - Unauthorized escalation without the permission remains fail closed. - Regression coverage for allowed and denied reviewer/merger cases. Blocker B — MUTATION_TASKS matches runtime wiring - Remove unenforced entries; document dedicated-gate exclusions (mark_final_review_decision, save/resume_review_draft, etc.). - Wire non-closing gitea_edit_pr as edit_pr; acquire lease uses acquire_reviewer_pr_lease task through verify_preflight_purity. - Inventory↔wiring consistency tests replace set-membership-only coverage. Blocker C — entrypoint side-effect ordering - Behavioral tests for merge_pr, submit_pr_review, and comment_issue prove the assessor block aborts before Gitea API mutation and local durable writes. Validation: 43 focused anti-stomp + related suites green; full suite 2639 passed / 6 skipped. Closes #604 --- anti_stomp_preflight.py | 156 ++++++++-- gitea_mcp_server.py | 23 +- task_capability_map.py | 21 ++ tests/test_anti_stomp_preflight.py | 468 ++++++++++++++++++++++++++++- 4 files changed, 637 insertions(+), 31 deletions(-) diff --git a/anti_stomp_preflight.py b/anti_stomp_preflight.py index bdc0f96..83f1dae 100644 --- a/anti_stomp_preflight.py +++ b/anti_stomp_preflight.py @@ -67,50 +67,79 @@ BLOCKER_KINDS = frozenset({ 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 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", - "claim_issue", "mark_issue", "lock_issue", "set_issue_labels", "create_label", "create_pr", - "comment_pr", "close_pr", + "edit_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", }) +# 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"}) @@ -122,13 +151,9 @@ _LEASE_AWARE_TASKS = frozenset({ "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] = { @@ -199,6 +224,10 @@ def roles_compatible(active_role: str | None, required_role: str | None) -> bool (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() @@ -211,6 +240,43 @@ def roles_compatible(active_role: str | None, required_role: str | None) -> bool 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], @@ -272,10 +338,12 @@ def assess_anti_stomp_preflight( org_explicit: bool = False, repo_explicit: bool = False, check_repo: bool = True, - # profile/role + # 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, @@ -327,11 +395,13 @@ def assess_anti_stomp_preflight( 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]] = [] @@ -376,15 +446,33 @@ def assess_anti_stomp_preflight( else: checks["repo"] = {"block": False, "skipped": True} - # ── profile/role ───────────────────────────────────────────────────────── + # ── 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: - if not roles_compatible(role, req_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}' does not match required role " - f"'{req_role}' for task '{task_name or '(unknown)'}' " - f"(profile={profile_name or '(unknown)'})" + 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} + checks["role"] = { + "block": True, + "reasons": reasons, + "required_permission": req_perm, + "capability_authorized": False, + } blockers.append( _blocker( BLOCKER_WRONG_ROLE, @@ -393,11 +481,25 @@ def assess_anti_stomp_preflight( "profile_name": profile_name, "profile_role": role, "required_role": req_role, + "required_permission": req_perm, }, ) ) else: - checks["role"] = {"block": False, "reasons": []} + 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} diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index d1b587f..43b018c 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -671,12 +671,18 @@ def _run_anti_stomp_preflight( 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"] @@ -734,7 +740,6 @@ def _run_anti_stomp_preflight( "approve_pr", "request_changes_pr", "merge_pr", - "mark_final_review_decision", }: try: wf_reasons = _review_workflow_load_gate_reasons() @@ -765,6 +770,8 @@ def _run_anti_stomp_preflight( 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"), @@ -5030,7 +5037,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 @@ -7650,7 +7662,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() @@ -7791,6 +7807,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" ) diff --git a/task_capability_map.py b/task_capability_map.py index 32d6ab0..4477747 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -60,6 +60,15 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.close", "role": "author", }, + # Non-closing PR metadata edits (title/body/base). Closing uses close_pr. + "edit_pr": { + "permission": "gitea.pr.create", + "role": "author", + }, + "gitea_edit_pr": { + "permission": "gitea.pr.create", + "role": "author", + }, "address_pr_change_requests": { "permission": "gitea.branch.push", "role": "author", @@ -68,10 +77,22 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.review", "role": "reviewer", }, + "submit_pr_review": { + "permission": "gitea.pr.review", + "role": "reviewer", + }, "merge_pr": { "permission": "gitea.pr.merge", "role": "merger", }, + "acquire_reviewer_pr_lease": { + "permission": "gitea.pr.comment", + "role": "reviewer", + }, + "gitea_acquire_reviewer_pr_lease": { + "permission": "gitea.pr.comment", + "role": "reviewer", + }, "adopt_merger_pr_lease": { "permission": "gitea.pr.comment", "role": "reviewer", diff --git a/tests/test_anti_stomp_preflight.py b/tests/test_anti_stomp_preflight.py index cf53177..93b6865 100644 --- a/tests/test_anti_stomp_preflight.py +++ b/tests/test_anti_stomp_preflight.py @@ -406,6 +406,141 @@ class TestWrongRole(unittest.TestCase): self.assertFalse(asp.roles_compatible("author", "merger")) +class TestCapabilityAuthorizedRoles(unittest.TestCase): + """Blocker A: reviewer/merger holding issue-comment capability may mutate. + + Nominal task role is still ``author`` in task_capability_map, but the + shared anti-stomp gate must not WRONG_ROLE-block when the active profile + holds ``gitea.issue.comment``. Unauthorized escalation without the + permission remains fail closed. + """ + + _ISSUE_COMMENT_TASKS = ( + "comment_issue", + "mark_issue", + "set_issue_labels", + "lock_issue", + ) + _ISSUE_COMMENT_PERM = "gitea.issue.comment" + + def _role_kwargs(self, task, role, *, allowed_ops, permission=None): + return _happy_kwargs( + task=task, + profile_name=f"prgs-{role}", + profile_role=role, + required_role="author", + required_permission=permission if permission is not None else self._ISSUE_COMMENT_PERM, + allowed_operations=allowed_ops, + workspace_path=f"/repo/branches/{role}-ws", + project_root="/repo", + current_branch=f"review/{role}-task", + ) + + def test_reviewer_allowed_with_issue_comment_capability(self): + ops = ["gitea.read", "gitea.issue.comment", "gitea.pr.review"] + for task in self._ISSUE_COMMENT_TASKS: + with self.subTest(task=task): + result = asp.assess_anti_stomp_preflight( + **self._role_kwargs(task, "reviewer", allowed_ops=ops) + ) + self.assertTrue(result["allowed"], result.get("reasons")) + self.assertFalse(result["block"]) + self.assertTrue( + result["checks"]["role"].get("capability_authorized") + ) + + def test_merger_allowed_with_issue_comment_capability(self): + ops = ["gitea.read", "gitea.issue.comment", "gitea.pr.merge"] + for task in self._ISSUE_COMMENT_TASKS: + with self.subTest(task=task): + result = asp.assess_anti_stomp_preflight( + **self._role_kwargs(task, "merger", allowed_ops=ops) + ) + self.assertTrue(result["allowed"], result.get("reasons")) + self.assertFalse(result["block"]) + + def test_reviewer_denied_without_issue_comment_capability(self): + # Holds review permission only — not issue comment. + ops = ["gitea.read", "gitea.pr.review", "gitea.pr.approve"] + for task in self._ISSUE_COMMENT_TASKS: + with self.subTest(task=task): + result = asp.assess_anti_stomp_preflight( + **self._role_kwargs(task, "reviewer", allowed_ops=ops) + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE) + + def test_merger_denied_without_issue_comment_capability(self): + ops = ["gitea.read", "gitea.pr.merge"] + for task in self._ISSUE_COMMENT_TASKS: + with self.subTest(task=task): + result = asp.assess_anti_stomp_preflight( + **self._role_kwargs(task, "merger", allowed_ops=ops) + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE) + + def test_not_broad_reviewer_to_author_bypass(self): + # Reviewer with only issue.comment must not pass create_pr (needs create). + result = asp.assess_anti_stomp_preflight( + **_happy_kwargs( + task="create_pr", + profile_name="prgs-reviewer", + profile_role="reviewer", + required_role="author", + required_permission="gitea.pr.create", + allowed_operations=["gitea.read", "gitea.issue.comment", "gitea.pr.review"], + workspace_path="/repo/branches/review-ws", + project_root="/repo", + ) + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE) + + def test_not_broad_merger_to_author_bypass(self): + result = asp.assess_anti_stomp_preflight( + **_happy_kwargs( + task="create_issue", + profile_name="prgs-merger", + profile_role="merger", + required_role="author", + required_permission="gitea.issue.create", + allowed_operations=["gitea.read", "gitea.issue.comment", "gitea.pr.merge"], + workspace_path="/repo/branches/merge-ws", + project_root="/repo", + ) + ) + self.assertTrue(result["block"]) + self.assertEqual(result["blocker_kind"], asp.BLOCKER_WRONG_ROLE) + + def test_authorization_compatible_helper(self): + self.assertTrue( + asp.authorization_compatible( + "reviewer", + "author", + required_permission="gitea.issue.comment", + allowed_operations=["gitea.issue.comment"], + ) + ) + self.assertFalse( + asp.authorization_compatible( + "reviewer", + "author", + required_permission="gitea.issue.comment", + allowed_operations=["gitea.pr.review"], + ) + ) + # Missing ops list fails closed when permission is required and roles differ. + self.assertFalse( + asp.authorization_compatible( + "reviewer", + "author", + required_permission="gitea.issue.comment", + allowed_operations=None, + ) + ) + + class TestWorkflowHash(unittest.TestCase): def test_stale_workflow_hash_blocks(self): result = asp.assess_anti_stomp_preflight( @@ -483,7 +618,7 @@ class TestRootCheckoutContamination(unittest.TestCase): class TestMutationTaskInventory(unittest.TestCase): - """AC1: mutation task set covers issue-listed paths.""" + """AC1 + Blocker B: declared inventory matches runtime wiring.""" def test_issue_required_paths_covered(self): required = { @@ -497,10 +632,74 @@ class TestMutationTaskInventory(unittest.TestCase): "merge_pr", "cleanup_merged_pr_branch", "cleanup_stale_claims", + "edit_pr", } missing = required - asp.MUTATION_TASKS self.assertFalse(missing, f"missing mutation tasks: {missing}") + def test_disputed_helpers_classified_as_dedicated_or_shared(self): + """Blocker B explicit classification for disputed mutation helpers.""" + # Shared preflight inventory (wired). + self.assertIn("edit_pr", asp.MUTATION_TASKS) + # Dedicated fail-closed gates — must NOT claim shared preflight. + for name in ( + "mark_final_review_decision", + "save_review_draft", + "resume_review_draft", + ): + self.assertNotIn(name, asp.MUTATION_TASKS, name) + self.assertIn(name, asp.DEDICATED_GATE_MUTATIONS, name) + self.assertTrue(asp.DEDICATED_GATE_MUTATIONS[name].strip()) + + def test_inventory_and_wiring_consistent(self): + """Every MUTATION_TASKS member is referenced as a live task kwarg. + + Accepts: + * task=\"name\" / task='name' on verify_preflight_purity / _run_anti_stomp + * review anti_task map values (approve_pr / request_changes_pr / …) + * gitea_ alias form when the bare name is also declared + """ + import pathlib + import re + + server_src = pathlib.Path(__file__).resolve().parents[1] / "gitea_mcp_server.py" + text = server_src.read_text(encoding="utf-8") + + # Collect string literals used as task= kwargs. + task_kw = set(re.findall(r'task\s*=\s*["\']([a-z0-9_]+)["\']', text)) + # anti_task map values: "approve": "approve_pr", + anti_map = set( + re.findall( + r'"(?:approve|request_changes|comment)"\s*:\s*"([a-z0-9_]+)"', + text, + ) + ) + wired = task_kw | anti_map + # Also accept f-string / conditional close_pr|edit_pr style already in task_kw. + + missing = set() + for task in asp.MUTATION_TASKS: + bare = task.removeprefix("gitea_") + if task in wired or bare in wired or f"gitea_{bare}" in wired: + continue + # Alias pairs: gitea_commit_files ↔ commit_files + if task.startswith("gitea_") and bare in asp.MUTATION_TASKS and bare in wired: + continue + if f"gitea_{task}" in asp.MUTATION_TASKS and task in wired: + continue + missing.add(task) + self.assertFalse( + missing, + f"MUTATION_TASKS declared but not wired in gitea_mcp_server.py: {sorted(missing)}", + ) + + # Dedicated exclusions must not appear as shared anti-stomp task kwargs + # for the three disputed local helpers (except resume→submit_pr_review). + for name in ("mark_final_review_decision", "save_review_draft"): + # May appear as string metadata but not as task="..." anti-stomp target. + # Allow mention in comments; assert not as task= kwarg. + self.assertNotIn(name, task_kw, f"{name} must not be shared-preflight task=") + class TestServerWiring(unittest.TestCase): """Smoke: MCP server imports anti_stomp and exposes the runner.""" @@ -562,5 +761,272 @@ class TestServerWiring(unittest.TestCase): self.assertIn(asp.BLOCKER_STALE_RUNTIME, str(ctx.exception)) +class TestEntrypointSideEffectOrdering(unittest.TestCase): + """Blocker C / AC4: anti-stomp aborts before Gitea mutation side effects. + + Invokes real entrypoint functions with external boundaries mocked. + Forces the assessor to block and proves api_request POST/merge paths + and durable local writes never run. + """ + + def _blocked_assessment(self, kind=asp.BLOCKER_FOREIGN_LEASE, next_action="stop"): + return { + "allowed": False, + "block": True, + "blockers": [{ + "kind": kind, + "reasons": ["forced block for ordering test"], + "exact_next_action": next_action, + "detail": {}, + }], + "reasons": ["forced block for ordering test"], + "exact_next_action": next_action, + "blocker_kind": kind, + "checks": {}, + } + + def test_merge_pr_blocks_before_merge_api(self): + import gitea_mcp_server as server + + blocked = self._blocked_assessment( + kind=asp.BLOCKER_HEAD_SHA, + next_action="re-pin expected_head_sha and retry", + ) + api_mock = mock.Mock(side_effect=AssertionError("Gitea API must not be called")) + save_lock = mock.Mock(side_effect=AssertionError("local lock must not mutate")) + + with mock.patch.dict( + os.environ, + {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, + clear=False, + ): + with mock.patch.object( + server, "_verify_role_mutation_workspace", return_value="/repo/branches/x" + ), mock.patch.object( + server, "_review_workflow_load_gate_reasons", return_value=[] + ), mock.patch.object( + server, "_live_namespace_health_gate", return_value=None + ), mock.patch.object( + server, "terminal_review_hard_stop_reasons", return_value=[] + ), mock.patch.object( + server, + "gitea_check_pr_eligibility", + return_value={ + "eligible": True, + "authenticated_user": "merger-bot", + "profile_name": "prgs-merger", + "pr_author": "other-user", + "head_sha": HEAD_A, + "mergeable": True, + "reasons": [], + }, + ), mock.patch.object( + server, "_reviewer_pr_lease_gate", return_value=[] + ), mock.patch.object( + server, "_pr_work_lease_reviewer_block", return_value={"block": False} + ), mock.patch.object( + server, "get_profile", return_value={ + "profile_name": "prgs-merger", + "allowed_operations": ["gitea.pr.merge", "gitea.read"], + "forbidden_operations": [], + } + ), mock.patch.object( + server, "_role_kind", return_value="merger" + ), mock.patch.object( + asp, "assess_anti_stomp_preflight", return_value=blocked + ) as assess_mock, mock.patch.object( + server, "api_request", api_mock + ), mock.patch.object( + server, "_save_review_decision_lock", save_lock + ): + result = server.gitea_merge_pr( + pr_number=680, + confirmation="MERGE PR 680", + expected_head_sha=HEAD_A, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + worktree_path="/repo/branches/merge-680", + ) + + self.assertFalse(result.get("performed")) + self.assertTrue(any("#604" in r or "anti-stomp" in r.lower() or asp.BLOCKER_HEAD_SHA in r + for r in (result.get("reasons") or [])), + result.get("reasons")) + joined = " ".join(result.get("reasons") or []) + self.assertIn(asp.BLOCKER_HEAD_SHA, joined) + self.assertIn("exact_next_action", joined) + assess_mock.assert_called() + api_mock.assert_not_called() + save_lock.assert_not_called() + + def test_submit_pr_review_blocks_before_review_api(self): + import gitea_mcp_server as server + + blocked = self._blocked_assessment( + kind=asp.BLOCKER_FOREIGN_LEASE, + next_action="stop; do not stomp foreign lease", + ) + api_mock = mock.Mock(side_effect=AssertionError("Gitea review API must not be called")) + save_lock = mock.Mock(side_effect=AssertionError("decision lock must not mutate")) + + with mock.patch.dict( + os.environ, + {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, + clear=False, + ): + with mock.patch.object( + server, "_verify_role_mutation_workspace", return_value="/repo/branches/r" + ), mock.patch.object( + server, "_review_workflow_load_gate_reasons", return_value=[] + ), mock.patch.object( + server, "_live_namespace_health_gate", return_value=None + ), mock.patch.object( + server, "check_review_decision_gate", return_value=[] + ), mock.patch.object( + server, + "gitea_check_pr_eligibility", + return_value={ + "eligible": True, + "authenticated_user": "reviewer-bot", + "profile_name": "prgs-reviewer", + "pr_author": "other-user", + "head_sha": HEAD_A, + "reasons": [], + }, + ), mock.patch.object( + server, "_reviewer_pr_lease_gate", return_value=[] + ), mock.patch.object( + server, "_pr_work_lease_reviewer_block", return_value={"block": False} + ), mock.patch.object( + server, "_load_review_decision_lock", return_value={ + "ready_expected_head_sha": HEAD_A, + "final_review_decision_ready": True, + "ready_pr_number": 680, + "ready_action": "request_changes", + } + ), mock.patch.object( + server, "get_profile", return_value={ + "profile_name": "prgs-reviewer", + "allowed_operations": [ + "gitea.pr.review", + "gitea.pr.request_changes", + "gitea.read", + ], + "forbidden_operations": [], + } + ), mock.patch.object( + asp, "assess_anti_stomp_preflight", return_value=blocked + ) as assess_mock, mock.patch.object( + server, "api_request", api_mock + ), mock.patch.object( + server, "_save_review_decision_lock", save_lock + ), mock.patch.object( + server, "_submit_pending_pull_review", api_mock + ): + result = server.gitea_submit_pr_review( + pr_number=680, + action="request_changes", + body="needs work", + expected_head_sha=HEAD_A, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + final_review_decision_ready=True, + worktree_path="/repo/branches/review-680", + ) + + self.assertFalse(result.get("performed")) + joined = " ".join(result.get("reasons") or []) + self.assertIn(asp.BLOCKER_FOREIGN_LEASE, joined) + self.assertIn("exact_next_action", joined) + assess_mock.assert_called() + # Assessor must run for request_changes_pr (anti_task map). + call_task = None + for c in assess_mock.call_args_list: + kwargs = c.kwargs if c.kwargs else {} + if "task" in kwargs: + call_task = kwargs["task"] + elif c.args: + call_task = c.args[0] if False else kwargs.get("task") + # assess_anti_stomp is called with keyword task= + if c.kwargs.get("task"): + call_task = c.kwargs["task"] + # _run_anti_stomp passes task as first positional to assess via keyword. + self.assertTrue(assess_mock.called) + api_mock.assert_not_called() + save_lock.assert_not_called() + + def test_comment_issue_blocks_before_comment_api(self): + """Representative issue-mutation class for AC4 coverage.""" + import gitea_mcp_server as server + + blocked = self._blocked_assessment( + kind=asp.BLOCKER_WRONG_ROLE, + next_action="switch namespace", + ) + api_mock = mock.Mock(side_effect=AssertionError("comment API must not be called")) + + with mock.patch.dict( + os.environ, + {"GITEA_TEST_FORCE_ANTI_STOMP": "1"}, + clear=False, + ): + with mock.patch.object( + server, "verify_preflight_purity", wraps=None + ) as purity, mock.patch.object( + server, "api_request", api_mock + ): + # Drive verify_preflight_purity → _run_anti_stomp by calling purity + # path: patch _run_anti_stomp to raise via assessor. + def _purity_side_effect(*args, **kwargs): + # Call real runner under force so assessor block surfaces. + return server._run_anti_stomp_preflight( + kwargs.get("task") or (args[2] if len(args) > 2 else None), + remote=args[0] if args else kwargs.get("remote"), + worktree_path=kwargs.get("worktree_path"), + org=kwargs.get("org"), + repo=kwargs.get("repo"), + raise_on_block=True, + ) + + purity.side_effect = None + with mock.patch.object( + asp, "assess_anti_stomp_preflight", return_value=blocked + ), mock.patch.object( + server, "verify_preflight_purity", side_effect=lambda *a, **k: ( + server._run_anti_stomp_preflight( + k.get("task"), + remote=a[0] if a else k.get("remote"), + worktree_path=k.get("worktree_path"), + org=k.get("org"), + repo=k.get("repo"), + ) + ) + ), mock.patch.object( + server, "_profile_operation_gate", return_value=[] + ), mock.patch.object( + server, "_canonical_comment_gate", return_value={"blocked": False} + ), mock.patch.object( + server, "get_profile", return_value={ + "profile_name": "prgs-author", + "allowed_operations": ["gitea.issue.comment", "gitea.read"], + "forbidden_operations": [], + } + ): + with self.assertRaises(RuntimeError) as ctx: + server.gitea_create_issue_comment( + issue_number=604, + body="handoff", + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + worktree_path="/repo/branches/issue-604", + ) + self.assertIn(asp.BLOCKER_WRONG_ROLE, str(ctx.exception)) + self.assertIn("exact_next_action", str(ctx.exception)) + api_mock.assert_not_called() + + if __name__ == "__main__": unittest.main() -- 2.43.7