Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 sysadmin
|
f1ad665242 |
@@ -12,7 +12,6 @@ import re
|
||||
from typing import Any, Callable
|
||||
|
||||
import issue_lock_provenance
|
||||
from post_merge_cleanup_proof import assess_post_merge_cleanup_proof
|
||||
from review_proofs import (
|
||||
HANDOFF_HEADING,
|
||||
assess_controller_handoff,
|
||||
@@ -1043,20 +1042,6 @@ def _rule_reviewer_review_mutation(
|
||||
)
|
||||
|
||||
|
||||
def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str, str]]:
|
||||
result = assess_post_merge_cleanup_proof(report_text)
|
||||
if not result.get("block"):
|
||||
return []
|
||||
return _findings_from_reasons(
|
||||
"reviewer.post_merge_cleanup_proof",
|
||||
result.get("reasons") or [],
|
||||
field="Cleanup status",
|
||||
severity="block",
|
||||
safe_next_action=result.get("safe_next_action")
|
||||
or "report CLEANUP_SKIPPED with blocker or full cleanup checklist",
|
||||
)
|
||||
|
||||
|
||||
_SHARED_ISSUE_LOCK_RULES = (
|
||||
_rule_shared_issue_lock_external_state,
|
||||
_rule_shared_manual_lock_pr_override,
|
||||
@@ -1086,7 +1071,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
_rule_reviewer_target_branch_freshness,
|
||||
_rule_reviewer_mutation_ledger,
|
||||
_rule_reviewer_review_mutation,
|
||||
_rule_reviewer_post_merge_cleanup_proof,
|
||||
_rule_reviewer_stale_head_proof,
|
||||
],
|
||||
"reconcile_already_landed": [
|
||||
|
||||
+31
-98
@@ -165,7 +165,6 @@ _preflight_capability_called = False
|
||||
_preflight_whoami_violation = False
|
||||
_preflight_capability_violation = False
|
||||
_preflight_resolved_role = None
|
||||
_preflight_resolved_task: str | None = None
|
||||
_process_start_porcelain: str | None = None
|
||||
_preflight_whoami_baseline_porcelain: str | None = None
|
||||
_preflight_capability_baseline_porcelain: str | None = None
|
||||
@@ -452,30 +451,11 @@ def assess_preflight_status(worktree_path: str | None = None) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def _clear_preflight_capability_state() -> None:
|
||||
"""Drop resolved capability proof (consumed by a mutation or fresh resolve)."""
|
||||
global _preflight_capability_called, _preflight_capability_violation
|
||||
global _preflight_resolved_task
|
||||
global _preflight_capability_baseline_porcelain, _preflight_capability_violation_files
|
||||
global _preflight_reviewer_violation_files
|
||||
|
||||
_preflight_capability_called = False
|
||||
_preflight_capability_violation = False
|
||||
_preflight_capability_violation_files = []
|
||||
_preflight_capability_baseline_porcelain = None
|
||||
_preflight_resolved_task = None
|
||||
_preflight_reviewer_violation_files = []
|
||||
|
||||
|
||||
def record_preflight_check(
|
||||
type_name: str,
|
||||
resolved_role: str | None = None,
|
||||
resolved_task: str | None = None,
|
||||
):
|
||||
def record_preflight_check(type_name: str, resolved_role: str | None = None):
|
||||
"""Record a pre-flight check (whoami or capability) with session-scoped deltas."""
|
||||
global _preflight_whoami_called, _preflight_capability_called
|
||||
global _preflight_whoami_violation, _preflight_capability_violation
|
||||
global _preflight_resolved_role, _preflight_resolved_task
|
||||
global _preflight_resolved_role
|
||||
global _preflight_whoami_baseline_porcelain, _preflight_capability_baseline_porcelain
|
||||
global _preflight_whoami_violation_files, _preflight_capability_violation_files
|
||||
global _preflight_reviewer_violation_files
|
||||
@@ -483,24 +463,13 @@ def record_preflight_check(
|
||||
current = _get_workspace_porcelain()
|
||||
|
||||
if type_name == "whoami":
|
||||
# Re-evaluate whoami violations instead of replaying sticky state (#252).
|
||||
# Interleaved read-only whoami must not clear a valid capability proof (#469).
|
||||
saved_capability = None
|
||||
preserve_capability = (
|
||||
_preflight_capability_called
|
||||
and _preflight_whoami_called
|
||||
and not _preflight_whoami_violation
|
||||
)
|
||||
if preserve_capability:
|
||||
saved_capability = (
|
||||
_preflight_capability_violation,
|
||||
list(_preflight_capability_violation_files),
|
||||
_preflight_capability_baseline_porcelain,
|
||||
_preflight_resolved_role,
|
||||
_preflight_resolved_task,
|
||||
)
|
||||
else:
|
||||
_clear_preflight_capability_state()
|
||||
# Fresh whoami restarts the capability step and re-evaluates violations
|
||||
# instead of replaying a sticky record (#252).
|
||||
_preflight_capability_called = False
|
||||
_preflight_capability_violation = False
|
||||
_preflight_capability_violation_files = []
|
||||
_preflight_capability_baseline_porcelain = None
|
||||
_preflight_reviewer_violation_files = []
|
||||
|
||||
process_start = _ensure_process_start_porcelain()
|
||||
whoami_delta = _new_tracked_changes_since(process_start, current)
|
||||
@@ -508,20 +477,6 @@ def record_preflight_check(
|
||||
_preflight_whoami_violation_files = whoami_delta
|
||||
_preflight_whoami_baseline_porcelain = current
|
||||
_preflight_whoami_called = True
|
||||
|
||||
if (
|
||||
preserve_capability
|
||||
and saved_capability is not None
|
||||
and not whoami_delta
|
||||
):
|
||||
(
|
||||
_preflight_capability_violation,
|
||||
_preflight_capability_violation_files,
|
||||
_preflight_capability_baseline_porcelain,
|
||||
_preflight_resolved_role,
|
||||
_preflight_resolved_task,
|
||||
) = saved_capability
|
||||
_preflight_capability_called = True
|
||||
elif type_name == "capability":
|
||||
baseline = _preflight_whoami_baseline_porcelain or ""
|
||||
capability_delta = _new_tracked_changes_since(baseline, current)
|
||||
@@ -531,8 +486,6 @@ def record_preflight_check(
|
||||
_preflight_capability_called = True
|
||||
if resolved_role:
|
||||
_preflight_resolved_role = resolved_role
|
||||
if resolved_task:
|
||||
_preflight_resolved_task = resolved_task
|
||||
|
||||
|
||||
def _enforce_branches_only_author_mutation(worktree_path: str | None = None) -> None:
|
||||
@@ -559,11 +512,7 @@ def _enforce_branches_only_author_mutation(worktree_path: str | None = None) ->
|
||||
)
|
||||
|
||||
|
||||
def verify_preflight_purity(
|
||||
remote: str | None = None,
|
||||
worktree_path: str | None = None,
|
||||
task: str | None = None,
|
||||
):
|
||||
def verify_preflight_purity(remote: str | None = None, worktree_path: str | None = None):
|
||||
"""Verify that identity and capability were verified prior to session edits."""
|
||||
global _preflight_reviewer_violation_files
|
||||
|
||||
@@ -582,16 +531,6 @@ def verify_preflight_purity(
|
||||
raise RuntimeError(
|
||||
"Pre-flight order violation: Task capability (gitea_resolve_task_capability) has not been resolved (fail closed)"
|
||||
)
|
||||
if (
|
||||
task is not None
|
||||
and _preflight_resolved_task is not None
|
||||
and task != _preflight_resolved_task
|
||||
):
|
||||
raise RuntimeError(
|
||||
"Pre-flight task mismatch: "
|
||||
f"resolved '{_preflight_resolved_task}' but mutation requires "
|
||||
f"'{task}' (fail closed)"
|
||||
)
|
||||
|
||||
ctx = _resolve_namespace_mutation_context(worktree_path)
|
||||
workspace = ctx["workspace_path"]
|
||||
@@ -652,7 +591,6 @@ def verify_preflight_purity(
|
||||
)
|
||||
|
||||
_enforce_branches_only_author_mutation(worktree_path)
|
||||
_clear_preflight_capability_state()
|
||||
|
||||
|
||||
def _verify_role_mutation_workspace(
|
||||
@@ -660,7 +598,6 @@ def _verify_role_mutation_workspace(
|
||||
*,
|
||||
worktree_path: str | None = None,
|
||||
worktree: str | None = None,
|
||||
task: str | None = None,
|
||||
) -> str:
|
||||
"""Bind reviewer/merger mutations to the active namespace workspace (#510)."""
|
||||
role = _effective_workspace_role()
|
||||
@@ -690,7 +627,7 @@ def _verify_role_mutation_workspace(
|
||||
)
|
||||
)
|
||||
resolved = assessment["mutation_workspace"]
|
||||
verify_preflight_purity(remote, worktree_path=resolved, task=task)
|
||||
verify_preflight_purity(remote, worktree_path=resolved)
|
||||
return resolved
|
||||
|
||||
from mcp.server.fastmcp import FastMCP # noqa: E402
|
||||
@@ -1407,7 +1344,7 @@ def gitea_create_issue(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="create_issue")
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||
base = repo_api_url(h, o, r)
|
||||
open_issues = api_get_all(f"{base}/issues?state=open&type=issues", auth)
|
||||
closed_issues = api_get_all(
|
||||
@@ -1540,7 +1477,7 @@ def gitea_lock_issue(
|
||||
)
|
||||
else:
|
||||
git_state = issue_lock_worktree.read_worktree_git_state(resolved_worktree)
|
||||
verify_preflight_purity(remote, worktree_path=resolved_worktree, task="lock_issue")
|
||||
verify_preflight_purity(remote, worktree_path=resolved_worktree)
|
||||
lock_assessment = issue_lock_worktree.assess_issue_lock_worktree(
|
||||
worktree_path=resolved_worktree,
|
||||
current_branch=git_state.get("current_branch"),
|
||||
@@ -1764,7 +1701,7 @@ def gitea_create_pr(
|
||||
)
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="create_pr")
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
|
||||
# ── Issue Lock Validation (Issue #194 / #196 / #443) ──
|
||||
@@ -2831,9 +2768,7 @@ def _evaluate_pr_review_submission(
|
||||
worktree_path: str | None = None,
|
||||
) -> dict:
|
||||
"""Shared gate chain for live submit and dry-run review tools."""
|
||||
_verify_role_mutation_workspace(
|
||||
remote, worktree_path=worktree_path, task="review_pr"
|
||||
)
|
||||
_verify_role_mutation_workspace(remote, worktree_path=worktree_path)
|
||||
action = (action or "").strip().lower()
|
||||
result = {
|
||||
"requested_action": action,
|
||||
@@ -3344,12 +3279,12 @@ def gitea_edit_pr(
|
||||
if not payload:
|
||||
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)
|
||||
verify_preflight_purity(remote)
|
||||
|
||||
# PR closure is a first-class capability, distinct from retitling or
|
||||
# rebasing edits (#216). Gate BEFORE auth/API setup so a blocked close
|
||||
# never touches the network.
|
||||
closing = payload.get("state") == "closed"
|
||||
if closing:
|
||||
gate_reasons = _profile_operation_gate("gitea.pr.close")
|
||||
if gate_reasons:
|
||||
@@ -3594,7 +3529,7 @@ def gitea_commit_files(
|
||||
branch="",
|
||||
)
|
||||
|
||||
verify_preflight_purity(remote, task="commit_files")
|
||||
verify_preflight_purity(remote)
|
||||
processed_files, source_proofs = _prepare_commit_payload_files(files)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
@@ -3700,9 +3635,7 @@ def gitea_merge_pr(
|
||||
reasons/gates passed or blocked, and merge result / merge commit if
|
||||
available. Never secrets.
|
||||
"""
|
||||
_verify_role_mutation_workspace(
|
||||
remote, worktree_path=worktree_path, task="merge_pr"
|
||||
)
|
||||
_verify_role_mutation_workspace(remote, worktree_path=worktree_path)
|
||||
do = (do or "").strip().lower()
|
||||
result = {
|
||||
"performed": False,
|
||||
@@ -4231,7 +4164,7 @@ def gitea_delete_branch(
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote, task="delete_branch")
|
||||
verify_preflight_purity(remote)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
import urllib.parse
|
||||
@@ -4340,7 +4273,7 @@ def gitea_reconcile_merged_cleanups(
|
||||
report["executed"] = False
|
||||
return {"success": True, "performed": False, **report}
|
||||
|
||||
verify_preflight_purity(remote, task="reconcile_merged_cleanups")
|
||||
verify_preflight_purity(remote)
|
||||
actions: list[dict] = []
|
||||
for entry in report.get("entries") or []:
|
||||
head_branch = entry.get("head_branch") or ""
|
||||
@@ -4654,7 +4587,7 @@ def gitea_reconcile_already_landed_pr(
|
||||
)
|
||||
return result
|
||||
|
||||
verify_preflight_purity(remote, task="reconcile_already_landed_pr")
|
||||
verify_preflight_purity(remote)
|
||||
|
||||
if post_comment and comment_body.strip():
|
||||
comment_block = _profile_operation_gate("gitea.pr.comment")
|
||||
@@ -4757,7 +4690,7 @@ def gitea_close_issue(
|
||||
task_capability_map.required_permission("close_issue"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, task="close_issue")
|
||||
verify_preflight_purity(remote)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
url = f"{repo_api_url(h, o, r)}/issues/{issue_number}"
|
||||
@@ -5185,7 +5118,7 @@ def gitea_acquire_reviewer_pr_lease(
|
||||
"permission_report": _permission_block_report("gitea.pr.comment"),
|
||||
}
|
||||
|
||||
_verify_role_mutation_workspace(remote, worktree=worktree, task="review_pr")
|
||||
_verify_role_mutation_workspace(remote, worktree=worktree)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
profile = get_profile()
|
||||
@@ -5286,7 +5219,7 @@ def gitea_heartbeat_reviewer_pr_lease(
|
||||
],
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote, task="review_pr")
|
||||
verify_preflight_purity(remote)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
body = reviewer_pr_lease.format_lease_body(
|
||||
@@ -5462,7 +5395,7 @@ def gitea_create_issue_comment(
|
||||
(permission blocks also carry a structured 'permission_report',
|
||||
#142).
|
||||
"""
|
||||
verify_preflight_purity(remote, task="comment_issue")
|
||||
verify_preflight_purity(remote)
|
||||
gate_reasons = _profile_operation_gate("gitea.issue.comment")
|
||||
reasons = list(gate_reasons)
|
||||
if not (body or "").strip():
|
||||
@@ -6996,7 +6929,7 @@ def gitea_mark_issue(
|
||||
task_capability_map.required_permission("mark_issue"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path, task="mark_issue")
|
||||
verify_preflight_purity(remote, worktree_path=worktree_path)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
@@ -7074,7 +7007,7 @@ def gitea_post_heartbeat(
|
||||
task_capability_map.required_permission("post_heartbeat"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, task="post_heartbeat")
|
||||
verify_preflight_purity(remote)
|
||||
active_profile = profile or get_profile().get("profile_name")
|
||||
body = issue_claim_heartbeat.format_heartbeat_body(
|
||||
kind="progress",
|
||||
@@ -7317,7 +7250,7 @@ def gitea_cleanup_stale_claims(
|
||||
task_capability_map.required_permission("cleanup_stale_claims"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, task="cleanup_stale_claims")
|
||||
verify_preflight_purity(remote)
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
@@ -7471,7 +7404,7 @@ def gitea_set_issue_labels(
|
||||
task_capability_map.required_permission("set_issue_labels"))
|
||||
if blocked:
|
||||
return blocked
|
||||
verify_preflight_purity(remote, task="set_issue_labels")
|
||||
verify_preflight_purity(remote)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
@@ -7708,7 +7641,7 @@ def gitea_resolve_task_capability(
|
||||
"exact_safe_next_action": next_safe_action,
|
||||
}
|
||||
|
||||
record_preflight_check("capability", required_role, resolved_task=task)
|
||||
record_preflight_check("capability", required_role)
|
||||
|
||||
# Try automatic dispatch switching
|
||||
_ensure_matching_profile(required_permission, required_role, remote, host)
|
||||
|
||||
@@ -1,239 +0,0 @@
|
||||
"""Post-merge cleanup proof verifier for reviewer final reports (#402)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
CLEANUP_SKIPPED = "CLEANUP_SKIPPED"
|
||||
CLEANUP_PERFORMED = "CLEANUP_PERFORMED"
|
||||
|
||||
_CLEANUP_SECTION_HINT = re.compile(
|
||||
r"(?:cleanup (?:status|result|mutations)|post-merge cleanup|"
|
||||
r"gitea_delete_branch|remote branch.*deleted|worktree remove)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CLEANUP_SKIPPED_RE = re.compile(r"\bCLEANUP_SKIPPED\b", re.IGNORECASE)
|
||||
_CLEANUP_BLOCKER_RE = re.compile(
|
||||
r"(?:cleanup blocker|cleanup skip(?:ped)? reason)\s*:\s*(.+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_REMOTE_DELETE_CLAIM_RE = re.compile(
|
||||
r"(?:gitea_delete_branch|remote (?:head )?branch (?:was )?deleted|"
|
||||
r"deleted remote branch|delete_branch)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_WORKTREE_REMOVE_CLAIM_RE = re.compile(
|
||||
r"(?:git worktree remove|worktree (?:was )?removed|removed (?:local )?worktree|"
|
||||
r"worktree cleanup performed)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_DELETE_CAPABILITY_RE = re.compile(
|
||||
r"(?:delete[- ]branch capability resolved|gitea\.branch\.delete)\s*:\s*"
|
||||
r".*(?:gitea\.branch\.delete|delete_branch).*(?:resolved|allowed|proven)|"
|
||||
r"gitea\.branch\.delete\s+(?:resolved|allowed|proven)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_DELETE_TASK_RE = re.compile(
|
||||
r"(?:delete_branch|cleanup_branch|reconcile_merged_cleanups)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_MERGE_RESULT_RE = re.compile(
|
||||
r"merge result\s*:\s*(?:merged|success|performed)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_MERGE_COMMIT_SHA_RE = re.compile(
|
||||
r"(?:merge commit sha|merged commit sha|merge commit)\s*:\s*([0-9a-f]{7,40})",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_PR_HEAD_BRANCH_RE = re.compile(
|
||||
r"(?:merged pr head branch|pr head branch|deleted branch)\s*:\s*(\S+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BRANCH_NOT_PROTECTED_RE = re.compile(
|
||||
r"branch (?:is )?not protected|branch protection\s*:\s*(?:none|false|no)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_OPEN_PR_INVENTORY_RE = re.compile(
|
||||
r"(?:no other open pr(?:\s+references)?(?:\s+\S+)?|open pr inventory proof|"
|
||||
r"open pr references).*(?:none|zero|0|clear|inventory complete)|"
|
||||
r"no other open pr references branch",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_ACTIVE_CLAIM_LEASE_RE = re.compile(
|
||||
r"(?:no active (?:heartbeat|claim|lease)|"
|
||||
r"(?:active )?(?:heartbeat|claim|lease)(?:/(?:claim|lease))*\s*:\s*none)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_SESSION_OWNED_WORKTREE_RE = re.compile(
|
||||
r"(?:removed worktree path|cleanup worktree path|session-owned worktree)\s*:\s*(\S+)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BRANCHES_PATH_RE = re.compile(r"\bbranches/", re.IGNORECASE)
|
||||
_CLEAN_TRACKED_RE = re.compile(
|
||||
r"(?:pre-removal tracked state|tracked state before removal)\s*:\s*clean",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CLEAN_UNTRACKED_RE = re.compile(
|
||||
r"(?:pre-removal untracked state|untracked state before removal)\s*:\s*clean",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_WORKTREE_LIST_AFTER_RE = re.compile(
|
||||
r"(?:git worktree list after|post-removal worktree list|worktree list after)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_WRONG_BRANCH_RE = re.compile(
|
||||
r"deleted branch (?:does not match|!=|differs from) (?:merged )?pr head",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def _claims_remote_delete(text: str) -> bool:
|
||||
return bool(_REMOTE_DELETE_CLAIM_RE.search(text))
|
||||
|
||||
|
||||
def _claims_worktree_remove(text: str) -> bool:
|
||||
return bool(_WORKTREE_REMOVE_CLAIM_RE.search(text))
|
||||
|
||||
|
||||
def _branch_safety_fields_present(text: str) -> list[str]:
|
||||
missing: list[str] = []
|
||||
if not _DELETE_CAPABILITY_RE.search(text):
|
||||
missing.append("delete-branch capability resolved (gitea.branch.delete)")
|
||||
if not _DELETE_TASK_RE.search(text):
|
||||
missing.append("delete-branch task named (delete_branch or cleanup)")
|
||||
if not _MERGE_RESULT_RE.search(text):
|
||||
missing.append("merge result: merged")
|
||||
if not _MERGE_COMMIT_SHA_RE.search(text):
|
||||
missing.append("merge commit SHA")
|
||||
if not _PR_HEAD_BRANCH_RE.search(text):
|
||||
missing.append("merged PR head branch / deleted branch name")
|
||||
if not _BRANCH_NOT_PROTECTED_RE.search(text):
|
||||
missing.append("branch not protected proof")
|
||||
if not _OPEN_PR_INVENTORY_RE.search(text):
|
||||
missing.append("open PR inventory proof (no other PR references branch)")
|
||||
if not _ACTIVE_CLAIM_LEASE_RE.search(text):
|
||||
missing.append("no active heartbeat/claim/lease proof")
|
||||
return missing
|
||||
|
||||
|
||||
def _worktree_cleanup_fields_present(text: str) -> list[str]:
|
||||
missing: list[str] = []
|
||||
match = _SESSION_OWNED_WORKTREE_RE.search(text)
|
||||
path = match.group(1).strip() if match else ""
|
||||
if not path:
|
||||
missing.append("session-owned worktree path")
|
||||
elif not _BRANCHES_PATH_RE.search(path.replace("\\", "/")):
|
||||
missing.append("worktree path under branches/")
|
||||
if not _CLEAN_TRACKED_RE.search(text):
|
||||
missing.append("pre-removal tracked state: clean")
|
||||
if not _CLEAN_UNTRACKED_RE.search(text):
|
||||
missing.append("pre-removal untracked state: clean")
|
||||
if not _WORKTREE_LIST_AFTER_RE.search(text):
|
||||
missing.append("git worktree list after removal")
|
||||
return missing
|
||||
|
||||
|
||||
def assess_post_merge_cleanup_proof(
|
||||
report_text: str,
|
||||
*,
|
||||
cleanup_session: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Validate post-merge cleanup claims carry safety-gate proof (#402)."""
|
||||
text = report_text or ""
|
||||
session = dict(cleanup_session or {})
|
||||
reasons: list[str] = []
|
||||
|
||||
if _CLEANUP_SKIPPED_RE.search(text) or session.get("outcome") == CLEANUP_SKIPPED:
|
||||
blocker = (session.get("blocker") or "").strip()
|
||||
if not blocker:
|
||||
match = _CLEANUP_BLOCKER_RE.search(text)
|
||||
blocker = match.group(1).strip() if match else ""
|
||||
if blocker.upper() == CLEANUP_SKIPPED:
|
||||
blocker = ""
|
||||
if not blocker:
|
||||
reasons.append(
|
||||
"CLEANUP_SKIPPED requires exact cleanup blocker reason (#402)"
|
||||
)
|
||||
return {
|
||||
"block": bool(reasons),
|
||||
"proven": not reasons,
|
||||
"outcome": CLEANUP_SKIPPED,
|
||||
"remote_delete_claimed": False,
|
||||
"worktree_remove_claimed": False,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"report CLEANUP_SKIPPED with exact blocker; do not claim performed cleanup"
|
||||
if reasons
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
|
||||
if not _CLEANUP_SECTION_HINT.search(text) and not session.get("cleanup_claimed"):
|
||||
return {
|
||||
"block": False,
|
||||
"proven": True,
|
||||
"outcome": None,
|
||||
"remote_delete_claimed": False,
|
||||
"worktree_remove_claimed": False,
|
||||
"reasons": [],
|
||||
"safe_next_action": "proceed",
|
||||
}
|
||||
|
||||
remote_delete = bool(
|
||||
session.get("remote_delete_claimed") or _claims_remote_delete(text)
|
||||
)
|
||||
worktree_remove = bool(
|
||||
session.get("worktree_remove_claimed") or _claims_worktree_remove(text)
|
||||
)
|
||||
|
||||
if _WRONG_BRANCH_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup report claims deleted branch that is not the merged PR head branch"
|
||||
)
|
||||
|
||||
if remote_delete:
|
||||
reasons.extend(
|
||||
f"remote branch deletion missing {field}"
|
||||
for field in _branch_safety_fields_present(text)
|
||||
)
|
||||
|
||||
if worktree_remove:
|
||||
reasons.extend(
|
||||
f"worktree removal missing {field}"
|
||||
for field in _worktree_cleanup_fields_present(text)
|
||||
)
|
||||
|
||||
if (remote_delete or worktree_remove) and not (remote_delete or worktree_remove):
|
||||
pass
|
||||
|
||||
if not remote_delete and not worktree_remove:
|
||||
cleanup_mutations = re.search(
|
||||
r"cleanup mutations\s*:\s*(?!none\b)\S",
|
||||
text,
|
||||
re.IGNORECASE,
|
||||
)
|
||||
if cleanup_mutations:
|
||||
reasons.append(
|
||||
"cleanup mutations reported without post-merge cleanup proof checklist"
|
||||
)
|
||||
|
||||
outcome = CLEANUP_PERFORMED if (remote_delete or worktree_remove) and not reasons else None
|
||||
if remote_delete or worktree_remove:
|
||||
outcome = CLEANUP_PERFORMED if not reasons else "CLEANUP_CLAIMED_UNPROVEN"
|
||||
|
||||
block = bool(reasons)
|
||||
return {
|
||||
"block": block,
|
||||
"proven": not block,
|
||||
"outcome": outcome,
|
||||
"remote_delete_claimed": remote_delete,
|
||||
"worktree_remove_claimed": worktree_remove,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"report CLEANUP_SKIPPED with exact blocker or include the full cleanup "
|
||||
"checklist before claiming remote delete or worktree removal"
|
||||
if reasons
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
@@ -881,40 +881,6 @@ Do not update the main checkout if merge failed, was blocked, or produced reconc
|
||||
|
||||
If any local artifact is created after final cleanup, run and report a new final status check.
|
||||
|
||||
## 28A. Post-merge cleanup proof checklist (#402)
|
||||
|
||||
Successful tool execution is not proof that cleanup was authorized. Before claiming remote branch deletion or local worktree removal, the final report must carry the full safety checklist below. If any gate is missing, report `CLEANUP_SKIPPED` with the exact blocker — never perform cleanup and never claim it was performed.
|
||||
|
||||
### Remote branch deletion checklist
|
||||
|
||||
When `gitea_delete_branch` (or equivalent) deletes the merged PR head branch, report:
|
||||
|
||||
* Delete-branch capability resolved: name the task (`delete_branch` / `cleanup_branch` / `reconcile_merged_cleanups`) and permission (`gitea.branch.delete`) with resolver proof before the delete call
|
||||
* Merge result: merged
|
||||
* Merge commit SHA: full 40-character SHA
|
||||
* Merged PR head branch / deleted branch: exact branch name (must match)
|
||||
* Branch protection: none / branch is not protected
|
||||
* Open PR inventory proof: no other open PR references the branch
|
||||
* Active heartbeat/claim/lease: none
|
||||
|
||||
### Local worktree removal checklist
|
||||
|
||||
When removing session-owned review/simulation worktrees under `branches/`, report:
|
||||
|
||||
* Session-owned worktree path: exact path under `branches/`
|
||||
* Pre-removal tracked state: clean
|
||||
* Pre-removal untracked state: clean
|
||||
* Git worktree list after removal: command output or equivalent proof
|
||||
|
||||
### Skipped cleanup
|
||||
|
||||
If any gate fails, report:
|
||||
|
||||
* Cleanup outcome: `CLEANUP_SKIPPED`
|
||||
* Cleanup blocker: exact missing gate (for example `gitea.branch.delete capability not resolved`)
|
||||
|
||||
Skipped cleanup with an exact blocker passes validation. Performed-cleanup claims without the checklist fail validation.
|
||||
|
||||
## 29. Recovery handoff rules
|
||||
|
||||
If blocked, produce a recovery handoff with:
|
||||
|
||||
@@ -1,169 +0,0 @@
|
||||
"""Tests for post-merge cleanup proof enforcement (#402)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from final_report_validator import assess_final_report_validator # noqa: E402
|
||||
from post_merge_cleanup_proof import ( # noqa: E402
|
||||
CLEANUP_SKIPPED,
|
||||
assess_post_merge_cleanup_proof,
|
||||
)
|
||||
|
||||
MERGE_SHA = "a" * 40
|
||||
HEAD_BRANCH = "feat/issue-274-branches-only-worktrees"
|
||||
WORKTREE = "branches/review-pr374-conflicts"
|
||||
|
||||
|
||||
def _full_remote_cleanup_report(**overrides):
|
||||
fields = {
|
||||
"Task": "review PR #374",
|
||||
"Merge result": "merged",
|
||||
"Merge commit SHA": MERGE_SHA,
|
||||
"Cleanup status": "remote branch deleted",
|
||||
"Delete-branch capability resolved": "gitea.branch.delete allowed via delete_branch task",
|
||||
"Merged PR head branch": HEAD_BRANCH,
|
||||
"Deleted branch": HEAD_BRANCH,
|
||||
"Branch protection": "none",
|
||||
"Open PR inventory proof": "no other open PR references branch (inventory complete)",
|
||||
"Active heartbeat/claim/lease": "none",
|
||||
"Cleanup mutations": "gitea_delete_branch on remote head branch",
|
||||
}
|
||||
fields.update(overrides)
|
||||
lines = ["## Controller Handoff", ""]
|
||||
lines.extend(f"- {key}: {value}" for key, value in fields.items())
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _full_worktree_cleanup_report(**overrides):
|
||||
fields = {
|
||||
"Task": "review PR #374",
|
||||
"Merge result": "merged",
|
||||
"Cleanup status": "local worktree removed",
|
||||
"Removed worktree path": WORKTREE,
|
||||
"Pre-removal tracked state": "clean",
|
||||
"Pre-removal untracked state": "clean",
|
||||
"Git worktree list after removal": "only main checkout listed",
|
||||
"Cleanup mutations": "git worktree remove on session-owned review worktree",
|
||||
}
|
||||
fields.update(overrides)
|
||||
lines = ["## Controller Handoff", ""]
|
||||
lines.extend(f"- {key}: {value}" for key, value in fields.items())
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
class TestPostMergeCleanupProof(unittest.TestCase):
|
||||
def test_no_cleanup_claim_passes(self):
|
||||
report = "## Controller Handoff\n- Task: review PR #1\n- Merge result: merged"
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_missing_capability_proof_blocked(self):
|
||||
report = _full_remote_cleanup_report(
|
||||
**{"Delete-branch capability resolved": "delete succeeded"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("capability" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_unmerged_pr_blocked(self):
|
||||
report = _full_remote_cleanup_report(**{"Merge result": "not merged"})
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("merge result" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_wrong_branch_blocked(self):
|
||||
report = _full_remote_cleanup_report(
|
||||
**{"Deleted branch": "feat/other-branch"}
|
||||
)
|
||||
report += "\nDeleted branch does not match merged PR head branch"
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_protected_branch_blocked(self):
|
||||
report = _full_remote_cleanup_report(**{"Branch protection": "enabled"})
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_open_pr_reference_blocked(self):
|
||||
report = _full_remote_cleanup_report(
|
||||
**{"Open PR inventory proof": "PR #999 still references branch"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_active_claim_blocked(self):
|
||||
report = _full_remote_cleanup_report(
|
||||
**{"Active heartbeat/claim/lease": "status:in-progress on linked issue"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_dirty_worktree_blocked(self):
|
||||
report = _full_worktree_cleanup_report(
|
||||
**{"Pre-removal tracked state": "dirty"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("tracked" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_foreign_worktree_blocked(self):
|
||||
report = _full_worktree_cleanup_report(
|
||||
**{"Removed worktree path": "/tmp/foreign-worktree"}
|
||||
)
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertTrue(any("branches/" in r.lower() for r in result["reasons"]))
|
||||
|
||||
def test_cleanup_skipped_with_blocker_passes(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Cleanup outcome: CLEANUP_SKIPPED",
|
||||
"- Cleanup blocker: gitea.branch.delete capability not resolved in active profile",
|
||||
])
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["outcome"], CLEANUP_SKIPPED)
|
||||
|
||||
def test_cleanup_skipped_without_blocker_blocked(self):
|
||||
report = "## Controller Handoff\n- Cleanup outcome: CLEANUP_SKIPPED"
|
||||
result = assess_post_merge_cleanup_proof(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_full_remote_cleanup_passes(self):
|
||||
result = assess_post_merge_cleanup_proof(_full_remote_cleanup_report())
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["remote_delete_claimed"])
|
||||
|
||||
def test_full_worktree_cleanup_passes(self):
|
||||
result = assess_post_merge_cleanup_proof(_full_worktree_cleanup_report())
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["worktree_remove_claimed"])
|
||||
|
||||
def test_validator_integration_blocks_unproven_delete(self):
|
||||
report = (
|
||||
"## Controller Handoff\n"
|
||||
"- Cleanup mutations: gitea_delete_branch deleted remote branch\n"
|
||||
)
|
||||
result = assess_final_report_validator(report, task_kind="review_pr")
|
||||
self.assertTrue(result["blocked"])
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("reviewer.post_merge_cleanup_proof", rule_ids)
|
||||
|
||||
def test_validator_integration_allows_skipped(self):
|
||||
report = "\n".join([
|
||||
"## Controller Handoff",
|
||||
"- Cleanup outcome: CLEANUP_SKIPPED",
|
||||
"- Cleanup blocker: branch still referenced by open PR #414",
|
||||
])
|
||||
result = assess_final_report_validator(report, task_kind="review_pr")
|
||||
cleanup_findings = [
|
||||
f for f in result["findings"]
|
||||
if f["rule_id"] == "reviewer.post_merge_cleanup_proof"
|
||||
]
|
||||
self.assertEqual(cleanup_findings, [])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -1,106 +0,0 @@
|
||||
"""#469: capability preflight survives interleaved read-only whoami calls."""
|
||||
|
||||
import os
|
||||
import unittest
|
||||
|
||||
import gitea_mcp_server as mcp_server
|
||||
|
||||
|
||||
class TestPreflightReadSurvival(unittest.TestCase):
|
||||
def setUp(self):
|
||||
self.orig_whoami = mcp_server._preflight_whoami_called
|
||||
self.orig_capability = mcp_server._preflight_capability_called
|
||||
self.orig_whoami_violation = mcp_server._preflight_whoami_violation
|
||||
self.orig_capability_violation = mcp_server._preflight_capability_violation
|
||||
self.orig_resolved_role = mcp_server._preflight_resolved_role
|
||||
self.orig_resolved_task = mcp_server._preflight_resolved_task
|
||||
self.orig_process_start = mcp_server._process_start_porcelain
|
||||
self.orig_whoami_baseline = mcp_server._preflight_whoami_baseline_porcelain
|
||||
self.orig_capability_baseline = mcp_server._preflight_capability_baseline_porcelain
|
||||
for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
|
||||
if key in os.environ:
|
||||
del os.environ[key]
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = ""
|
||||
mcp_server._preflight_whoami_called = False
|
||||
mcp_server._preflight_capability_called = False
|
||||
mcp_server._preflight_whoami_violation = False
|
||||
mcp_server._preflight_capability_violation = False
|
||||
mcp_server._preflight_resolved_role = None
|
||||
mcp_server._preflight_resolved_task = None
|
||||
mcp_server._process_start_porcelain = ""
|
||||
mcp_server._preflight_whoami_baseline_porcelain = None
|
||||
mcp_server._preflight_capability_baseline_porcelain = None
|
||||
|
||||
def tearDown(self):
|
||||
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
|
||||
mcp_server._preflight_capability_violation = self.orig_capability_violation
|
||||
mcp_server._preflight_resolved_role = self.orig_resolved_role
|
||||
mcp_server._preflight_resolved_task = self.orig_resolved_task
|
||||
mcp_server._process_start_porcelain = self.orig_process_start
|
||||
mcp_server._preflight_whoami_baseline_porcelain = self.orig_whoami_baseline
|
||||
mcp_server._preflight_capability_baseline_porcelain = self.orig_capability_baseline
|
||||
for key in ("GITEA_TEST_FORCE_DIRTY", "GITEA_TEST_PORCELAIN"):
|
||||
if key in os.environ:
|
||||
del os.environ[key]
|
||||
|
||||
def test_interleaved_whoami_preserves_capability(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="reconciler", resolved_task="close_pr"
|
||||
)
|
||||
self.assertTrue(mcp_server._preflight_capability_called)
|
||||
self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
|
||||
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertTrue(mcp_server._preflight_capability_called)
|
||||
self.assertEqual(mcp_server._preflight_resolved_task, "close_pr")
|
||||
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
self.assertFalse(mcp_server._preflight_capability_called)
|
||||
|
||||
def test_missing_capability_still_fails_closed(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
self.assertIn("has not been resolved", str(ctx.exception))
|
||||
|
||||
def test_task_mismatch_fails_closed(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="author", resolved_task="create_issue"
|
||||
)
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="close_pr")
|
||||
self.assertIn("task mismatch", str(ctx.exception))
|
||||
|
||||
def test_capability_consumed_after_mutation_gate(self):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="author", resolved_task="create_issue"
|
||||
)
|
||||
mcp_server.verify_preflight_purity(task="create_issue")
|
||||
with self.assertRaises(RuntimeError) as ctx:
|
||||
mcp_server.verify_preflight_purity(task="create_issue")
|
||||
self.assertIn("has not been resolved", str(ctx.exception))
|
||||
|
||||
def test_whoami_recovery_after_violation_clears_capability(self):
|
||||
os.environ["GITEA_TEST_FORCE_DIRTY"] = "1"
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertTrue(mcp_server._preflight_whoami_violation)
|
||||
|
||||
del os.environ["GITEA_TEST_FORCE_DIRTY"]
|
||||
os.environ["GITEA_TEST_PORCELAIN"] = ""
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
self.assertFalse(mcp_server._preflight_whoami_violation)
|
||||
self.assertFalse(mcp_server._preflight_capability_called)
|
||||
|
||||
mcp_server.record_preflight_check(
|
||||
"capability", resolved_role="reviewer", resolved_task="review_pr"
|
||||
)
|
||||
mcp_server.verify_preflight_purity(task="review_pr")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user