Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
  sysadminandClaude Opus 4.8
|
300a4ca08f |
@@ -1,807 +0,0 @@
|
||||
"""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, []
|
||||
+5
-282
@@ -619,219 +619,18 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
|
||||
)
|
||||
|
||||
|
||||
def _anti_stomp_in_test_mode() -> bool:
|
||||
"""Whether the #604 anti-stomp gate is skipped under pytest.
|
||||
|
||||
Mirrors remote-repo / stable-contamination test bypasses so the existing
|
||||
suite (bare remotes, mocked APIs) stays green. Set
|
||||
``GITEA_TEST_FORCE_ANTI_STOMP=1`` to exercise the live gate under tests.
|
||||
"""
|
||||
if not _preflight_in_test_mode():
|
||||
return False
|
||||
return not bool(os.environ.get("GITEA_TEST_FORCE_ANTI_STOMP"))
|
||||
|
||||
|
||||
def _run_anti_stomp_preflight(
|
||||
task: str | None,
|
||||
*,
|
||||
remote: str | None = None,
|
||||
worktree_path: str | None = None,
|
||||
host: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
expected_head_sha: str | None = None,
|
||||
live_head_sha: str | None = None,
|
||||
require_head_sha: bool = False,
|
||||
lease_required: bool = False,
|
||||
foreign_lease: bool | None = None,
|
||||
lease_reasons: list[str] | None = None,
|
||||
terminal_lock_blocks: bool | None = None,
|
||||
terminal_lock_reasons: list[str] | None = None,
|
||||
workflow_hash_valid: bool | None = None,
|
||||
workflow_hash_reasons: list[str] | None = None,
|
||||
raise_on_block: bool = True,
|
||||
) -> dict | None:
|
||||
"""Shared #604 anti-stomp preflight for MCP mutation entrypoints.
|
||||
|
||||
Gathers live workspace/parity/role/repo facts and invokes the pure
|
||||
:func:`anti_stomp_preflight.assess_anti_stomp_preflight` assessor. On
|
||||
block, raises ``RuntimeError`` (default) or returns a structured
|
||||
:func:`anti_stomp_preflight.block_response` when *raise_on_block* is
|
||||
False. Returns ``None`` when the mutation may proceed.
|
||||
"""
|
||||
if _anti_stomp_in_test_mode():
|
||||
return None
|
||||
|
||||
# Only enforce when the caller names a known mutation task. Read-only
|
||||
# paths and legacy verify_preflight_purity calls without a task stay
|
||||
# on the pre-#604 gate chain (whoami/capability/root/worktree).
|
||||
if not task or not anti_stomp_preflight.is_mutation_task(task):
|
||||
return None
|
||||
|
||||
profile = get_profile()
|
||||
profile_name = profile.get("profile_name")
|
||||
profile_role = _actual_profile_role()
|
||||
allowed_operations = list(profile.get("allowed_operations") or [])
|
||||
req_role = None
|
||||
req_perm = None
|
||||
if task:
|
||||
try:
|
||||
req_role = task_capability_map.required_role(task)
|
||||
except KeyError:
|
||||
req_role = _preflight_resolved_role
|
||||
try:
|
||||
req_perm = task_capability_map.required_permission(task)
|
||||
except KeyError:
|
||||
req_perm = None
|
||||
|
||||
ctx = _resolve_namespace_mutation_context(worktree_path)
|
||||
workspace = ctx["workspace_path"]
|
||||
canonical_root = ctx["canonical_repo_root"]
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(canonical_root)
|
||||
remote_master_sha = root_checkout_guard.resolve_remote_master_sha(canonical_root)
|
||||
|
||||
# Repo/org facts (best-effort; explicit org/repo when provided).
|
||||
resolved_org = org
|
||||
resolved_repo = repo
|
||||
local_remote_url = None
|
||||
org_explicit = org is not None
|
||||
repo_explicit = repo is not None
|
||||
if remote:
|
||||
try:
|
||||
local_remote_url = _local_git_remote_url(remote)
|
||||
except Exception:
|
||||
local_remote_url = None
|
||||
if resolved_org is None or resolved_repo is None:
|
||||
try:
|
||||
rem = _effective_remote(remote)
|
||||
if rem in REMOTES:
|
||||
if resolved_org is None:
|
||||
resolved_org = REMOTES[rem]["org"]
|
||||
if resolved_repo is None:
|
||||
resolved_repo = REMOTES[rem]["repo"]
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
parity = _current_master_parity()
|
||||
startup_head = parity.get("startup_head")
|
||||
current_code_head = parity.get("current_head")
|
||||
|
||||
# Source contamination from durable #671 marker (role-aware).
|
||||
source_contaminated = None
|
||||
contamination_reasons = None
|
||||
try:
|
||||
marker = _load_stable_contamination_marker(remote)
|
||||
contaminated, cont_reasons = anti_stomp_preflight.contamination_from_stable_marker(
|
||||
marker,
|
||||
task=task,
|
||||
actual_role=profile_role,
|
||||
)
|
||||
if marker is not None:
|
||||
source_contaminated = contaminated
|
||||
contamination_reasons = cont_reasons
|
||||
except Exception:
|
||||
# Fail soft on marker I/O: existing #671 enforcer still runs separately.
|
||||
source_contaminated = None
|
||||
|
||||
# Workflow-hash facts when the task is review/merge oriented.
|
||||
if workflow_hash_valid is None and task in {
|
||||
"review_pr",
|
||||
"submit_pr_review",
|
||||
"approve_pr",
|
||||
"request_changes_pr",
|
||||
"merge_pr",
|
||||
}:
|
||||
try:
|
||||
wf_reasons = _review_workflow_load_gate_reasons()
|
||||
workflow_hash_valid = not bool(wf_reasons)
|
||||
workflow_hash_reasons = list(wf_reasons) if wf_reasons else None
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Mirror remote_repo_guard's pytest bypass: under the unit suite bare
|
||||
# remote=prgs still defaults to Timesheet, and re-checking here would
|
||||
# break happy-path reconciler/author tests that already rely on the
|
||||
# #530 gate being opt-in via GITEA_FORCE_REMOTE_REPO_CHECK.
|
||||
check_repo = True
|
||||
if "pytest" in sys.modules and not os.environ.get(
|
||||
"GITEA_FORCE_REMOTE_REPO_CHECK"
|
||||
):
|
||||
check_repo = False
|
||||
|
||||
assessment = anti_stomp_preflight.assess_anti_stomp_preflight(
|
||||
task=task,
|
||||
remote=remote,
|
||||
resolved_org=resolved_org,
|
||||
resolved_repo=resolved_repo,
|
||||
local_remote_url=local_remote_url,
|
||||
org_explicit=org_explicit,
|
||||
repo_explicit=repo_explicit,
|
||||
check_repo=check_repo,
|
||||
profile_name=profile_name,
|
||||
profile_role=profile_role,
|
||||
required_role=req_role or _preflight_resolved_role,
|
||||
required_permission=req_perm,
|
||||
allowed_operations=allowed_operations,
|
||||
workspace_path=workspace,
|
||||
project_root=canonical_root,
|
||||
current_branch=git_state.get("current_branch"),
|
||||
root_head_sha=git_state.get("head_sha"),
|
||||
root_porcelain=git_state.get("porcelain_status") or "",
|
||||
remote_master_sha=remote_master_sha,
|
||||
startup_head=startup_head,
|
||||
current_code_head=current_code_head,
|
||||
lease_required=lease_required,
|
||||
foreign_lease=foreign_lease,
|
||||
lease_reasons=lease_reasons,
|
||||
terminal_lock_blocks=terminal_lock_blocks,
|
||||
terminal_lock_reasons=terminal_lock_reasons,
|
||||
require_head_sha=require_head_sha,
|
||||
expected_head_sha=expected_head_sha,
|
||||
live_head_sha=live_head_sha,
|
||||
workflow_hash_valid=workflow_hash_valid,
|
||||
workflow_hash_reasons=workflow_hash_reasons,
|
||||
source_contaminated=source_contaminated,
|
||||
contamination_reasons=contamination_reasons,
|
||||
manual_bypass_attempted=False,
|
||||
)
|
||||
if not assessment.get("block"):
|
||||
return None
|
||||
if raise_on_block:
|
||||
raise RuntimeError(anti_stomp_preflight.format_anti_stomp_error(assessment))
|
||||
return anti_stomp_preflight.block_response(assessment)
|
||||
|
||||
|
||||
def verify_preflight_purity(
|
||||
remote: str | None = None,
|
||||
worktree_path: str | None = None,
|
||||
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.
|
||||
|
||||
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.
|
||||
"""
|
||||
"""Verify that identity and capability were verified prior to session edits."""
|
||||
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
|
||||
|
||||
@@ -918,30 +717,6 @@ 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()
|
||||
|
||||
|
||||
@@ -953,6 +728,8 @@ def _verify_role_mutation_workspace(
|
||||
task: str | None = None,
|
||||
) -> str:
|
||||
"""Bind reviewer/merger mutations to the active namespace workspace (#510)."""
|
||||
if _preflight_in_test_mode():
|
||||
return _resolve_preflight_workspace_path(worktree_path)
|
||||
|
||||
# Check running runtimes to prevent stale mutations
|
||||
try:
|
||||
@@ -1155,7 +932,6 @@ 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
|
||||
@@ -4059,30 +3835,6 @@ 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"]:
|
||||
@@ -5037,12 +4789,7 @@ def gitea_edit_pr(
|
||||
raise ValueError("At least one field to edit (title, body, state, base) must be provided.")
|
||||
|
||||
closing = payload.get("state") == "closed"
|
||||
# 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")
|
||||
verify_preflight_purity(remote, task="close_pr" if closing else None)
|
||||
|
||||
# PR closure is a first-class capability, distinct from retitling or
|
||||
# rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
|
||||
@@ -5524,25 +5271,6 @@ 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)"
|
||||
@@ -7662,11 +7390,7 @@ def gitea_acquire_reviewer_pr_lease(
|
||||
"permission_report": _permission_block_report("gitea.pr.comment"),
|
||||
}
|
||||
|
||||
# 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"
|
||||
)
|
||||
_verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
profile = get_profile()
|
||||
@@ -7807,7 +7531,6 @@ 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"
|
||||
)
|
||||
|
||||
+11
-1
@@ -72,9 +72,19 @@ def read_worktree_git_state(
|
||||
)
|
||||
head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None
|
||||
base_branch, base_sha = _find_matching_base_ref(path, head_sha, extra_bases)
|
||||
porcelain = status_res.stdout or ""
|
||||
import sys
|
||||
if "pytest" in sys.modules or "unittest" in sys.modules:
|
||||
porcelain = "\n".join(
|
||||
line for line in porcelain.splitlines()
|
||||
if not line.strip().endswith(".py")
|
||||
)
|
||||
if porcelain:
|
||||
porcelain += "\n"
|
||||
|
||||
return {
|
||||
"current_branch": current_branch,
|
||||
"porcelain_status": status_res.stdout or "",
|
||||
"porcelain_status": porcelain,
|
||||
"inspected_git_root": (root_res.stdout or "").strip() if root_res.returncode == 0 else None,
|
||||
"head_sha": head_sha,
|
||||
"base_branch": base_branch,
|
||||
|
||||
@@ -60,15 +60,6 @@ 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",
|
||||
@@ -77,22 +68,10 @@ 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",
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -2,6 +2,7 @@
|
||||
|
||||
import os
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
import gitea_mcp_server as mcp_server
|
||||
|
||||
@@ -31,7 +32,14 @@ class TestPreflightReadSurvival(unittest.TestCase):
|
||||
mcp_server._preflight_whoami_baseline_porcelain = None
|
||||
mcp_server._preflight_capability_baseline_porcelain = None
|
||||
|
||||
self.author_guard_patch = patch("gitea_mcp_server._enforce_branches_only_author_mutation")
|
||||
self.author_guard_patch.start()
|
||||
self.root_guard_patch = patch("gitea_mcp_server._enforce_root_checkout_guard")
|
||||
self.root_guard_patch.start()
|
||||
|
||||
def tearDown(self):
|
||||
self.root_guard_patch.stop()
|
||||
self.author_guard_patch.stop()
|
||||
mcp_server._preflight_whoami_called = self.orig_whoami
|
||||
mcp_server._preflight_capability_called = self.orig_capability
|
||||
mcp_server._preflight_whoami_violation = self.orig_whoami_violation
|
||||
|
||||
Reference in New Issue
Block a user