Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity
Files
3fd02a3c633fb539850ce202c68b2043cbf7e4ca
Gitea-Tools/anti_stomp_preflight.py
T
sysadmin 3fd02a3c63 fix: anti-stomp capability auth, inventory wiring, side-effect proof (#604) 
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
2026-07-12 09:00:45 -04:00

808 lines
30 KiB
Python
Raw Blame History

"""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, []
Reference in New Issue View Git Blame Copy Permalink