Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
  sysadminandClaude Opus 4.8
|
7f97de1ed6 |
@@ -1,344 +0,0 @@
|
||||
"""Audit vs cleanup phase gates for reconciliation workflows (#419).
|
||||
|
||||
Audit/reconciliation tasks are read-only unless a separate cleanup phase is
|
||||
explicitly authorized with exact capability proof, safety proof, and
|
||||
before/after snapshots. Cleanup mutations must be classified in final reports;
|
||||
audit reports must not claim ``no mutations`` when cleanup occurred.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
RECONCILE_WORKFLOW_PATH = "workflows/reconcile-landed-pr.md"
|
||||
|
||||
PHASE_AUDIT = "audit"
|
||||
PHASE_CLEANUP = "cleanup"
|
||||
|
||||
# Tasks that enter audit phase on capability resolution (read-only default).
|
||||
AUDIT_PHASE_TASKS = frozenset({
|
||||
"reconcile-landed-pr",
|
||||
"reconcile_landed_pr",
|
||||
"reconcile_issue_claims",
|
||||
"reconcile_merged_cleanups",
|
||||
})
|
||||
|
||||
# Mutation tasks forbidden during audit phase (fail closed).
|
||||
AUDIT_FORBIDDEN_TASKS = frozenset({
|
||||
"delete_branch",
|
||||
"create_branch",
|
||||
"push_branch",
|
||||
"create_pr",
|
||||
"commit_files",
|
||||
"gitea_commit_files",
|
||||
"mark_issue",
|
||||
"lock_issue",
|
||||
"claim_issue",
|
||||
"close_pr",
|
||||
"close_issue",
|
||||
"create_issue",
|
||||
"merge_pr",
|
||||
"review_pr",
|
||||
"submit_pr_review",
|
||||
"comment_pr",
|
||||
"comment_issue",
|
||||
"set_issue_labels",
|
||||
})
|
||||
|
||||
# Shell/git commands audit phase must not run.
|
||||
AUDIT_FORBIDDEN_COMMAND_RE = re.compile(
|
||||
r"(?:^|\s)(?:git\s+(?:push|branch\s+-D|worktree\s+remove)|"
|
||||
r"gitea_delete_branch|delete_remote_branch)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
_NO_MUTATIONS_RE = re.compile(
|
||||
r"(?:no\s+mutations|mutations\s*:\s*none|no\s+unsafe\s+mutation)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_CLEANUP_OCCURRED_RE = re.compile(
|
||||
r"(?:delete_remote_branch|remove_local_worktree|git\s+branch\s+-D|"
|
||||
r"git\s+worktree\s+remove|remote branch.*deleted|worktree.*removed|"
|
||||
r"cleanup\s+phase\s*:\s*(?!none\b)\S)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_EXTERNAL_STATE_RE = re.compile(
|
||||
r"^\s*[-*]?\s*external[- ]state mutations\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_GIT_REF_RE = re.compile(
|
||||
r"^\s*[-*]?\s*git ref mutations\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_CLEANUP_MUTATIONS_RE = re.compile(
|
||||
r"^\s*[-*]?\s*cleanup mutations\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_CLEANUP_PHASE_AUTH_RE = re.compile(
|
||||
r"^\s*[-*]?\s*cleanup phase (?:authorized|authorization)\s*:\s*true",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_DELETE_CAPABILITY_RE = re.compile(
|
||||
r"^\s*[-*]?\s*delete.?branch capability(?: proven)?\s*:\s*true",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_BEFORE_AFTER_RE = re.compile(
|
||||
r"^\s*[-*]?\s*before/after (?:state )?snapshot\s*:",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_SAFETY_PROOF_RE = re.compile(
|
||||
r"^\s*[-*]?\s*(?:branch|worktree) safe to remove\s*:\s*true",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
|
||||
_session: dict[str, Any] | None = None
|
||||
|
||||
|
||||
def _blank_session() -> dict[str, Any]:
|
||||
return {
|
||||
"phase": PHASE_AUDIT,
|
||||
"entered_from_task": None,
|
||||
"cleanup_authorized": False,
|
||||
"cleanup_authorization": {},
|
||||
}
|
||||
|
||||
|
||||
def current_phase() -> str | None:
|
||||
"""Return active reconciliation phase or None when unset."""
|
||||
if not _session:
|
||||
return None
|
||||
return _session.get("phase")
|
||||
|
||||
|
||||
def active_record() -> dict[str, Any] | None:
|
||||
"""Return a copy of the session record, if any."""
|
||||
return dict(_session) if _session else None
|
||||
|
||||
|
||||
def clear_phase() -> None:
|
||||
"""Clear reconciliation phase state."""
|
||||
global _session
|
||||
_session = None
|
||||
|
||||
|
||||
def enter_audit_phase(task: str) -> dict[str, Any]:
|
||||
"""Enter read-only audit phase for a reconciliation task."""
|
||||
global _session
|
||||
normalized = (task or "").strip().lower()
|
||||
_session = _blank_session()
|
||||
_session["entered_from_task"] = normalized
|
||||
return dict(_session)
|
||||
|
||||
|
||||
def authorize_cleanup_phase(
|
||||
*,
|
||||
operator_approved: bool = False,
|
||||
workflow_authorized: bool = False,
|
||||
delete_capability_proven: bool = False,
|
||||
safety_proof: dict[str, Any] | None = None,
|
||||
before_after_snapshot: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Authorize cleanup phase after explicit approval and safety proofs."""
|
||||
reasons: list[str] = []
|
||||
if not (operator_approved or workflow_authorized):
|
||||
reasons.append(
|
||||
"cleanup phase requires operator approval or explicit workflow "
|
||||
"authorization"
|
||||
)
|
||||
if not delete_capability_proven:
|
||||
reasons.append(
|
||||
"cleanup phase requires exact delete_branch capability proof "
|
||||
"(gitea.branch.delete)"
|
||||
)
|
||||
safety = dict(safety_proof or {})
|
||||
if not safety.get("safe_to_delete_remote") and not safety.get(
|
||||
"safe_to_remove_worktree"
|
||||
):
|
||||
reasons.append(
|
||||
"cleanup phase requires proof that branch/worktree is safe to remove"
|
||||
)
|
||||
snapshot = dict(before_after_snapshot or {})
|
||||
if not snapshot.get("before") or not snapshot.get("after"):
|
||||
reasons.append(
|
||||
"cleanup phase requires before/after state snapshot"
|
||||
)
|
||||
|
||||
if reasons:
|
||||
return {
|
||||
"authorized": False,
|
||||
"phase": current_phase() or PHASE_AUDIT,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"remain in audit-only mode or supply operator approval, "
|
||||
"delete_branch capability proof, safety proof, and "
|
||||
"before/after snapshot before cleanup"
|
||||
),
|
||||
}
|
||||
|
||||
global _session
|
||||
if _session is None:
|
||||
_session = _blank_session()
|
||||
_session["phase"] = PHASE_CLEANUP
|
||||
_session["cleanup_authorized"] = True
|
||||
_session["cleanup_authorization"] = {
|
||||
"operator_approved": operator_approved,
|
||||
"workflow_authorized": workflow_authorized,
|
||||
"delete_capability_proven": delete_capability_proven,
|
||||
"safety_proof": safety,
|
||||
"before_after_snapshot": snapshot,
|
||||
}
|
||||
return {
|
||||
"authorized": True,
|
||||
"phase": PHASE_CLEANUP,
|
||||
"reasons": [],
|
||||
"cleanup_authorization": dict(_session["cleanup_authorization"]),
|
||||
"safe_next_action": "proceed with authorized cleanup mutations only",
|
||||
}
|
||||
|
||||
|
||||
def check_audit_task_enters_phase(task: str) -> bool:
|
||||
"""Return whether resolving *task* should enter audit phase."""
|
||||
return (task or "").strip().lower() in AUDIT_PHASE_TASKS
|
||||
|
||||
|
||||
def check_audit_mutation_allowed(task: str) -> tuple[bool, list[str]]:
|
||||
"""Fail closed when a mutation task runs during audit phase."""
|
||||
normalized = (task or "").strip().lower()
|
||||
phase = current_phase()
|
||||
if phase != PHASE_AUDIT:
|
||||
return True, []
|
||||
if normalized in AUDIT_FORBIDDEN_TASKS:
|
||||
return False, [
|
||||
f"task '{normalized}' is forbidden in audit-only reconciliation "
|
||||
"mode: switch to an explicit cleanup phase with operator approval "
|
||||
"and exact delete_branch capability proof before cleanup mutations"
|
||||
]
|
||||
return True, []
|
||||
|
||||
|
||||
def check_cleanup_execution_allowed() -> tuple[bool, list[str]]:
|
||||
"""Fail closed when cleanup execution is attempted without authorization."""
|
||||
phase = current_phase()
|
||||
if phase == PHASE_CLEANUP and (_session or {}).get("cleanup_authorized"):
|
||||
return True, []
|
||||
if phase is None:
|
||||
return False, [
|
||||
"cleanup execution requires an active reconciliation session; "
|
||||
"resolve a reconciliation audit task first"
|
||||
]
|
||||
return False, [
|
||||
"cleanup execution forbidden in audit-only reconciliation mode; "
|
||||
"call gitea_authorize_reconciliation_cleanup_phase with operator "
|
||||
"approval, delete_branch capability proof, safety proof, and "
|
||||
"before/after snapshot"
|
||||
]
|
||||
|
||||
|
||||
def classify_cleanup_mutation(action: str) -> str:
|
||||
"""Map a cleanup action to the required mutation ledger category (#419)."""
|
||||
normalized = (action or "").strip().lower()
|
||||
if "delete_remote" in normalized or normalized in {
|
||||
"delete_branch",
|
||||
"gitea_delete_branch",
|
||||
}:
|
||||
return "external-state"
|
||||
if "branch" in normalized and "delete" in normalized:
|
||||
return "git-ref"
|
||||
if "worktree" in normalized or "remove_local" in normalized:
|
||||
return "cleanup"
|
||||
return "cleanup"
|
||||
|
||||
|
||||
def assess_audit_reconciliation_report(report_text: str) -> dict[str, Any]:
|
||||
"""Validate audit/cleanup reconciliation reports (fail closed)."""
|
||||
text = report_text or ""
|
||||
reasons: list[str] = []
|
||||
|
||||
cleanup_occurred = bool(_CLEANUP_OCCURRED_RE.search(text))
|
||||
claims_no_mutations = bool(_NO_MUTATIONS_RE.search(text))
|
||||
|
||||
if cleanup_occurred and claims_no_mutations:
|
||||
reasons.append(
|
||||
"report claims no mutations but documents cleanup mutations; "
|
||||
"audit-only reports must not perform cleanup and cleanup reports "
|
||||
"must not claim no mutations"
|
||||
)
|
||||
|
||||
if cleanup_occurred:
|
||||
if not _CLEANUP_PHASE_AUTH_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup mutations reported without "
|
||||
"'Cleanup phase authorized: true'"
|
||||
)
|
||||
if not _DELETE_CAPABILITY_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup mutations reported without delete_branch capability "
|
||||
"proof"
|
||||
)
|
||||
if not _BEFORE_AFTER_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup mutations reported without before/after state snapshot"
|
||||
)
|
||||
if not _SAFETY_PROOF_RE.search(text):
|
||||
reasons.append(
|
||||
"cleanup mutations reported without branch/worktree safety proof"
|
||||
)
|
||||
|
||||
if re.search(r"delete_remote|remote branch.*delet", text, re.I):
|
||||
if not _EXTERNAL_STATE_RE.search(text):
|
||||
reasons.append(
|
||||
"remote branch deletion must be classified under "
|
||||
"External-state mutations"
|
||||
)
|
||||
if re.search(r"git\s+branch\s+-D|local branch.*delet", text, re.I):
|
||||
if not _GIT_REF_RE.search(text):
|
||||
reasons.append(
|
||||
"local branch deletion must be classified under "
|
||||
"Git ref mutations"
|
||||
)
|
||||
if re.search(r"worktree.*remov|remove_local_worktree", text, re.I):
|
||||
if not _CLEANUP_MUTATIONS_RE.search(text):
|
||||
reasons.append(
|
||||
"worktree removal must be classified under Cleanup mutations"
|
||||
)
|
||||
|
||||
if (
|
||||
RECONCILE_WORKFLOW_PATH.replace("workflows/", "") in text
|
||||
or "reconcile-landed-pr" in text.lower()
|
||||
):
|
||||
if cleanup_occurred and "audit phase" in text.lower():
|
||||
if "cleanup phase" not in text.lower():
|
||||
reasons.append(
|
||||
"report mixes audit phase with cleanup mutations without "
|
||||
"documenting cleanup phase transition"
|
||||
)
|
||||
|
||||
proven = not reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": reasons,
|
||||
"cleanup_occurred": cleanup_occurred,
|
||||
"claims_no_mutations": claims_no_mutations,
|
||||
"safe_next_action": (
|
||||
"proceed"
|
||||
if proven
|
||||
else "fix audit/cleanup report: separate audit from cleanup phase, "
|
||||
"classify mutations, and do not claim no mutations after cleanup"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def assess_audit_command_allowed(command: str) -> tuple[bool, list[str]]:
|
||||
"""Block shell commands that perform cleanup during audit phase."""
|
||||
phase = current_phase()
|
||||
if phase != PHASE_AUDIT:
|
||||
return True, []
|
||||
cmd = (command or "").strip()
|
||||
if AUDIT_FORBIDDEN_COMMAND_RE.search(cmd):
|
||||
return False, [
|
||||
f"command forbidden in audit-only reconciliation mode: {cmd!r}; "
|
||||
"authorize cleanup phase before branch/worktree deletion or push"
|
||||
]
|
||||
return True, []
|
||||
@@ -0,0 +1,198 @@
|
||||
"""Early duplicate-work detection for author work-issue flows (#400)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from issue_claim_heartbeat import (
|
||||
_linked_open_pr,
|
||||
_matching_branch_names,
|
||||
classify_issue_claim,
|
||||
)
|
||||
|
||||
STAGES = (
|
||||
"claim",
|
||||
"lock",
|
||||
"worktree",
|
||||
"edit",
|
||||
"commit",
|
||||
"push",
|
||||
"create_pr",
|
||||
)
|
||||
|
||||
ELIGIBILITY_OPEN_PR_EXISTS = "OPEN_PR_EXISTS"
|
||||
ELIGIBILITY_DUPLICATE_BRANCH_EXISTS = "DUPLICATE_BRANCH_EXISTS"
|
||||
ELIGIBILITY_ACTIVE_CLAIM = "ACTIVE_CLAIM_BY_OTHER"
|
||||
ELIGIBILITY_CLEAR = "CLEAR"
|
||||
|
||||
|
||||
def assess_author_duplicate_work(
|
||||
issue_number: int,
|
||||
*,
|
||||
stage: str,
|
||||
open_prs: list[dict] | None = None,
|
||||
branch_names: list[str] | None = None,
|
||||
claim_entry: dict | None = None,
|
||||
matching_branches: list[str] | None = None,
|
||||
allow_stale_takeover: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Fail closed when duplicate work is detected before author mutations (#400)."""
|
||||
stage_norm = (stage or "").strip().lower()
|
||||
if stage_norm not in STAGES:
|
||||
return {
|
||||
"allowed": False,
|
||||
"block": True,
|
||||
"eligibility_class": "INVALID_STAGE",
|
||||
"stage": stage_norm or None,
|
||||
"reasons": [f"unknown duplicate-work stage {stage!r}"],
|
||||
"safe_next_action": f"use one of: {', '.join(STAGES)}",
|
||||
}
|
||||
|
||||
prs = list(open_prs or [])
|
||||
linked_pr = _linked_open_pr(int(issue_number), prs)
|
||||
branches = list(
|
||||
matching_branches
|
||||
if matching_branches is not None
|
||||
else _matching_branch_names(int(issue_number), list(branch_names or []))
|
||||
)
|
||||
|
||||
reasons: list[str] = []
|
||||
eligibility = ELIGIBILITY_CLEAR
|
||||
|
||||
if linked_pr:
|
||||
eligibility = ELIGIBILITY_OPEN_PR_EXISTS
|
||||
reasons.append(
|
||||
f"open PR #{linked_pr.get('number')} already covers issue "
|
||||
f"#{issue_number}"
|
||||
)
|
||||
|
||||
if branches and stage_norm in {"claim", "lock", "worktree", "edit"}:
|
||||
if eligibility == ELIGIBILITY_CLEAR:
|
||||
eligibility = ELIGIBILITY_DUPLICATE_BRANCH_EXISTS
|
||||
reasons.append(
|
||||
f"remote branch(es) already exist for issue #{issue_number}: "
|
||||
f"{', '.join(branches)}"
|
||||
)
|
||||
|
||||
entry = claim_entry or {}
|
||||
claim_status = (entry.get("status") or "").strip()
|
||||
if claim_status in {"active", "awaiting_review"} and stage_norm == "claim":
|
||||
if entry.get("reclaimable") and allow_stale_takeover:
|
||||
pass
|
||||
elif claim_status == "active" and not entry.get("reclaimable"):
|
||||
if eligibility == ELIGIBILITY_CLEAR:
|
||||
eligibility = ELIGIBILITY_ACTIVE_CLAIM
|
||||
reasons.append(
|
||||
f"issue #{issue_number} has active claim "
|
||||
f"(status={claim_status})"
|
||||
)
|
||||
elif claim_status == "awaiting_review" and stage_norm == "claim":
|
||||
if not linked_pr:
|
||||
reasons.append(
|
||||
f"issue #{issue_number} is awaiting_review but no linked open PR "
|
||||
"was supplied for duplicate-work proof"
|
||||
)
|
||||
|
||||
allowed = not reasons
|
||||
outcome = "duplicate_work_prevented" if not allowed else "clear"
|
||||
if stage_norm == "create_pr" and not allowed:
|
||||
outcome = "duplicate_pr_prevented"
|
||||
elif stage_norm in {"commit", "push"} and not allowed:
|
||||
outcome = "duplicate_push_prevented" if stage_norm == "push" else "duplicate_commit_prevented"
|
||||
|
||||
return {
|
||||
"allowed": allowed,
|
||||
"block": not allowed,
|
||||
"eligibility_class": eligibility if not allowed else ELIGIBILITY_CLEAR,
|
||||
"stage": stage_norm,
|
||||
"linked_open_pr": linked_pr.get("number") if linked_pr else None,
|
||||
"matching_branches": branches,
|
||||
"claim_status": claim_status or None,
|
||||
"outcome": outcome,
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"stop without edits/commit/push/PR; produce reconciliation handoff "
|
||||
"preserving local work only"
|
||||
if not allowed
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def build_claim_entry_from_classification(classification: dict) -> dict:
|
||||
"""Map ``classify_issue_claim`` output to gate claim metadata."""
|
||||
return {
|
||||
"status": classification.get("status"),
|
||||
"reclaimable": classification.get("reclaimable"),
|
||||
"linked_open_pr": classification.get("linked_open_pr"),
|
||||
}
|
||||
|
||||
|
||||
_DUPLICATE_OUTCOME_RE = re.compile(
|
||||
r"(duplicate\s+(?:pr|branch|commit|push)\s+prevented|"
|
||||
r"duplicate\s+work\s+not\s+prevented|reconciliation\s+handoff)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
|
||||
|
||||
def assess_work_issue_duplicate_prevention_report(report_text: str) -> dict:
|
||||
"""#400: work-issue reports must state duplicate-work prevention outcome."""
|
||||
text = report_text or ""
|
||||
if "duplicate work" not in text.lower() and "duplicate pr" not in text.lower():
|
||||
return {
|
||||
"proven": True,
|
||||
"block": False,
|
||||
"reasons": [],
|
||||
"safe_next_action": "proceed",
|
||||
}
|
||||
if _DUPLICATE_OUTCOME_RE.search(text):
|
||||
return {
|
||||
"proven": True,
|
||||
"block": False,
|
||||
"reasons": [],
|
||||
"safe_next_action": "proceed",
|
||||
}
|
||||
return {
|
||||
"proven": False,
|
||||
"block": True,
|
||||
"reasons": [
|
||||
"duplicate-work discussion must name a prevention outcome "
|
||||
"(duplicate PR/branch/commit/push prevented, or duplicate work "
|
||||
"not prevented, or reconciliation handoff)"
|
||||
],
|
||||
"safe_next_action": "state exact duplicate-work prevention class in final report",
|
||||
}
|
||||
|
||||
|
||||
def classify_and_assess(
|
||||
issue: dict,
|
||||
*,
|
||||
stage: str,
|
||||
comments: list[dict] | None = None,
|
||||
open_prs: list[dict] | None = None,
|
||||
branch_names: list[str] | None = None,
|
||||
allow_stale_takeover: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Combine claim classification with duplicate-work gate assessment."""
|
||||
issue_number = int(issue.get("number") or 0)
|
||||
claim = classify_issue_claim(
|
||||
issue=issue,
|
||||
comments=comments or [],
|
||||
open_prs=open_prs or [],
|
||||
branch_names=branch_names or [],
|
||||
)
|
||||
assessment = assess_author_duplicate_work(
|
||||
issue_number,
|
||||
stage=stage,
|
||||
open_prs=open_prs,
|
||||
branch_names=branch_names,
|
||||
claim_entry=build_claim_entry_from_classification(claim),
|
||||
matching_branches=claim.get("matching_branches"),
|
||||
allow_stale_takeover=allow_stale_takeover,
|
||||
)
|
||||
return {
|
||||
"issue_number": issue_number,
|
||||
"claim": claim,
|
||||
"duplicate_work": assessment,
|
||||
}
|
||||
@@ -870,22 +870,6 @@ def _rule_reviewer_mutation_ledger(
|
||||
)
|
||||
|
||||
|
||||
def _rule_audit_reconciliation_boundary(report_text: str) -> list[dict[str, str]]:
|
||||
from audit_reconciliation_mode import assess_audit_reconciliation_report
|
||||
|
||||
result = assess_audit_reconciliation_report(report_text)
|
||||
if result.get("proven"):
|
||||
return []
|
||||
return _findings_from_reasons(
|
||||
"reconcile.audit_cleanup_boundary",
|
||||
result.get("reasons") or [],
|
||||
field="Audit/cleanup phase",
|
||||
severity="block",
|
||||
safe_next_action=result.get("safe_next_action")
|
||||
or "separate audit from authorized cleanup and classify mutations",
|
||||
)
|
||||
|
||||
|
||||
def _rule_reviewer_review_mutation(
|
||||
report_text: str,
|
||||
*,
|
||||
@@ -936,7 +920,6 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
_rule_reviewer_git_fetch_readonly,
|
||||
_rule_reviewer_legacy_workspace_mutations,
|
||||
_rule_reviewer_vague_mutations_none,
|
||||
_rule_audit_reconciliation_boundary,
|
||||
],
|
||||
"author_issue": [
|
||||
_rule_shared_controller_handoff,
|
||||
|
||||
+158
-116
@@ -503,10 +503,10 @@ import issue_lock_worktree # noqa: E402
|
||||
import already_landed_reconcile # noqa: E402
|
||||
import author_mutation_worktree # noqa: E402
|
||||
import issue_claim_heartbeat # noqa: E402
|
||||
import author_duplicate_work_gate # noqa: E402
|
||||
import merged_cleanup_reconcile # noqa: E402
|
||||
import reconciler_profile # noqa: E402
|
||||
import reconciliation_workflow # noqa: E402
|
||||
import audit_reconciliation_mode # noqa: E402
|
||||
import review_merge_state_machine # noqa: E402
|
||||
import native_mcp_preference # noqa: E402
|
||||
|
||||
@@ -1050,6 +1050,76 @@ def gitea_create_issue(
|
||||
return _with_optional_url({"number": data["number"]}, data.get("html_url"))
|
||||
|
||||
|
||||
def _list_repo_branch_names(h: str, o: str, r: str, auth: str, *, limit: int = 200) -> list[str]:
|
||||
branches = api_get_all(f"{repo_api_url(h, o, r)}/branches", auth, limit=limit)
|
||||
return [_branch_entry_name(branch) for branch in branches]
|
||||
|
||||
|
||||
def _gather_author_duplicate_work_context(
|
||||
issue_number: int,
|
||||
*,
|
||||
h: str,
|
||||
o: str,
|
||||
r: str,
|
||||
auth: str,
|
||||
exclude_branch_name: str | None = None,
|
||||
) -> dict:
|
||||
base = repo_api_url(h, o, r)
|
||||
issue = api_request("GET", f"{base}/issues/{issue_number}", auth)
|
||||
comments = api_request("GET", f"{base}/issues/{issue_number}/comments", auth) or []
|
||||
open_prs = api_get_all(f"{base}/pulls?state=open", auth)
|
||||
branch_names = _list_repo_branch_names(h, o, r, auth)
|
||||
if exclude_branch_name:
|
||||
branch_names = [
|
||||
name for name in branch_names
|
||||
if name != exclude_branch_name
|
||||
]
|
||||
return {
|
||||
"issue": issue,
|
||||
"comments": comments,
|
||||
"open_prs": open_prs,
|
||||
"branch_names": branch_names,
|
||||
}
|
||||
|
||||
|
||||
def _enforce_author_duplicate_work_gate(
|
||||
issue_number: int,
|
||||
stage: str,
|
||||
*,
|
||||
h: str,
|
||||
o: str,
|
||||
r: str,
|
||||
auth: str,
|
||||
allow_stale_takeover: bool = False,
|
||||
exclude_branch_name: str | None = None,
|
||||
) -> dict:
|
||||
"""Fail closed when duplicate work is detected (#400)."""
|
||||
ctx = _gather_author_duplicate_work_context(
|
||||
issue_number,
|
||||
h=h,
|
||||
o=o,
|
||||
r=r,
|
||||
auth=auth,
|
||||
exclude_branch_name=exclude_branch_name,
|
||||
)
|
||||
result = author_duplicate_work_gate.classify_and_assess(
|
||||
ctx["issue"],
|
||||
stage=stage,
|
||||
comments=ctx["comments"],
|
||||
open_prs=ctx["open_prs"],
|
||||
branch_names=ctx["branch_names"],
|
||||
allow_stale_takeover=allow_stale_takeover,
|
||||
)
|
||||
assessment = result.get("duplicate_work") or {}
|
||||
if assessment.get("block"):
|
||||
reasons = "; ".join(assessment.get("reasons") or ["duplicate work detected"])
|
||||
raise RuntimeError(
|
||||
f"Author duplicate-work gate (#400) blocked at stage '{stage}': "
|
||||
f"{reasons} (fail closed)"
|
||||
)
|
||||
return result
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_lock_issue(
|
||||
issue_number: int,
|
||||
@@ -1112,48 +1182,17 @@ def gitea_lock_issue(
|
||||
issue_lock_worktree.format_issue_lock_worktree_error(lock_assessment)
|
||||
)
|
||||
|
||||
# 2. Check if the issue already has an open PR (reuse protection)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
url = f"{repo_api_url(h, o, r)}/pulls?state=open"
|
||||
|
||||
try:
|
||||
prs = api_get_all(url, auth)
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Could not list open PRs to verify issue lock: {e}")
|
||||
|
||||
for pr in prs:
|
||||
pr_head = pr.get("head", {}).get("ref", "")
|
||||
pr_title = pr.get("title", "")
|
||||
pr_body = pr.get("body", "")
|
||||
|
||||
if expected_pattern in pr_head:
|
||||
raise ValueError(
|
||||
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}, branch '{pr_head}') (fail closed)"
|
||||
)
|
||||
|
||||
patterns = [
|
||||
f"closes #{issue_number}",
|
||||
f"fixes #{issue_number}",
|
||||
]
|
||||
text_to_check = f"{pr_title} {pr_body}".lower()
|
||||
if any(p in text_to_check for p in patterns):
|
||||
raise ValueError(
|
||||
f"Issue #{issue_number} is already tied to an open PR (PR #{pr.get('number')}) via Closes/Fixes reference (fail closed)"
|
||||
)
|
||||
|
||||
branch_url = f"{repo_api_url(h, o, r)}/branches"
|
||||
try:
|
||||
branches = api_get_all(branch_url, auth)
|
||||
except Exception as e:
|
||||
raise RuntimeError(f"Could not list branches to verify issue lock: {e}")
|
||||
for branch in branches:
|
||||
name = _branch_entry_name(branch)
|
||||
if expected_pattern in name:
|
||||
raise ValueError(
|
||||
f"Issue #{issue_number} already has matching branch '{name}' "
|
||||
"(fail closed)"
|
||||
)
|
||||
_enforce_author_duplicate_work_gate(
|
||||
issue_number,
|
||||
"lock",
|
||||
h=h,
|
||||
o=o,
|
||||
r=r,
|
||||
auth=auth,
|
||||
exclude_branch_name=branch_name,
|
||||
)
|
||||
|
||||
work_lease = _build_author_issue_work_lease(
|
||||
issue_number=issue_number,
|
||||
@@ -1274,6 +1313,17 @@ def gitea_create_pr(
|
||||
f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
|
||||
)
|
||||
|
||||
auth = _auth(h)
|
||||
_enforce_author_duplicate_work_gate(
|
||||
int(locked_issue),
|
||||
"create_pr",
|
||||
h=h,
|
||||
o=o,
|
||||
r=r,
|
||||
auth=auth,
|
||||
exclude_branch_name=locked_branch,
|
||||
)
|
||||
|
||||
# Check for forbidden terms anywhere in title/body
|
||||
forbidden_terms = ["equivalent", "related", "same as"]
|
||||
text_to_check = f"{title} {body}".lower()
|
||||
@@ -1290,7 +1340,6 @@ def gitea_create_pr(
|
||||
f"PR title or body must contain 'Closes #{locked_issue}' or 'Fixes #{locked_issue}' exactly to ensure durable tracking (fail closed)"
|
||||
)
|
||||
|
||||
auth = _auth(h)
|
||||
url = f"{repo_api_url(h, o, r)}/pulls"
|
||||
payload = {"title": title, "body": body, "head": head, "base": base}
|
||||
meta = {"title": title, "head": head, "base": base}
|
||||
@@ -3479,18 +3528,6 @@ def gitea_delete_branch(
|
||||
"permission_report": _permission_block_report("gitea.branch.delete"),
|
||||
}
|
||||
|
||||
audit_allowed, audit_reasons = (
|
||||
audit_reconciliation_mode.check_audit_mutation_allowed("delete_branch")
|
||||
)
|
||||
if not audit_allowed:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"required_permission": "gitea.branch.delete",
|
||||
"reasons": audit_reasons,
|
||||
"audit_phase": audit_reconciliation_mode.current_phase(),
|
||||
}
|
||||
|
||||
verify_preflight_purity(remote)
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
@@ -3560,18 +3597,6 @@ def gitea_reconcile_merged_cleanups(
|
||||
"execute_confirmed must be True when dry_run=False (fail closed)"
|
||||
)
|
||||
|
||||
if not dry_run:
|
||||
exec_allowed, exec_reasons = (
|
||||
audit_reconciliation_mode.check_cleanup_execution_allowed()
|
||||
)
|
||||
if not exec_allowed:
|
||||
return {
|
||||
"success": False,
|
||||
"performed": False,
|
||||
"reasons": exec_reasons,
|
||||
"audit_phase": audit_reconciliation_mode.current_phase(),
|
||||
}
|
||||
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
base = repo_api_url(h, o, r)
|
||||
@@ -3654,53 +3679,6 @@ def gitea_reconcile_merged_cleanups(
|
||||
return {"success": True, "performed": True, **report}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_authorize_reconciliation_cleanup_phase(
|
||||
operator_approved: bool = False,
|
||||
workflow_authorized: bool = False,
|
||||
delete_capability_proven: bool = False,
|
||||
safe_to_delete_remote: bool = False,
|
||||
safe_to_remove_worktree: bool = False,
|
||||
before_state: str = "",
|
||||
after_state: str = "",
|
||||
) -> dict:
|
||||
"""Authorize cleanup phase after audit-only reconciliation (#419).
|
||||
|
||||
Requires operator or workflow approval, exact delete_branch capability proof,
|
||||
branch/worktree safety proof, and before/after state snapshots. Audit phase
|
||||
forbids branch deletion, worktree removal, pushes, and issue/PR mutations.
|
||||
"""
|
||||
delete_gate = _profile_operation_gate("gitea.branch.delete")
|
||||
capability_ok = not bool(delete_gate)
|
||||
if delete_capability_proven and delete_gate:
|
||||
return {
|
||||
"authorized": False,
|
||||
"performed": False,
|
||||
"delete_capability_verified": False,
|
||||
"reasons": [
|
||||
"delete_capability_proven=true but active profile lacks "
|
||||
"gitea.branch.delete",
|
||||
] + delete_gate,
|
||||
"audit_phase": audit_reconciliation_mode.current_phase(),
|
||||
}
|
||||
result = audit_reconciliation_mode.authorize_cleanup_phase(
|
||||
operator_approved=operator_approved,
|
||||
workflow_authorized=workflow_authorized,
|
||||
delete_capability_proven=delete_capability_proven and capability_ok,
|
||||
safety_proof={
|
||||
"safe_to_delete_remote": safe_to_delete_remote,
|
||||
"safe_to_remove_worktree": safe_to_remove_worktree,
|
||||
},
|
||||
before_after_snapshot={
|
||||
"before": (before_state or "").strip(),
|
||||
"after": (after_state or "").strip(),
|
||||
},
|
||||
)
|
||||
result["performed"] = bool(result.get("authorized"))
|
||||
result["delete_capability_verified"] = capability_ok
|
||||
return result
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_assess_already_landed_reconciliation(
|
||||
pr_number: int,
|
||||
@@ -6075,6 +6053,14 @@ def gitea_mark_issue(
|
||||
)
|
||||
|
||||
if action == "start":
|
||||
_enforce_author_duplicate_work_gate(
|
||||
issue_number,
|
||||
"claim",
|
||||
h=h,
|
||||
o=o,
|
||||
r=r,
|
||||
auth=auth,
|
||||
)
|
||||
with _audited("label_issue", host=h, remote=remote, org=o, repo=r,
|
||||
issue_number=issue_number,
|
||||
request_metadata={"op": "add", "label": "status:in-progress"}):
|
||||
@@ -6156,6 +6142,65 @@ def gitea_post_heartbeat(
|
||||
)
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_assess_author_duplicate_work(
|
||||
issue_number: int,
|
||||
stage: str,
|
||||
remote: str = "dadeschools",
|
||||
host: str | None = None,
|
||||
org: str | None = None,
|
||||
repo: str | None = None,
|
||||
allow_stale_takeover: bool = False,
|
||||
exclude_branch_name: str | None = None,
|
||||
) -> dict:
|
||||
"""Read-only: assess duplicate-work risk before author mutations (#400).
|
||||
|
||||
Call at claim, lock, worktree, edit, commit, push, and create_pr stages.
|
||||
"""
|
||||
read_block = _profile_operation_gate("gitea.read")
|
||||
if read_block:
|
||||
return {
|
||||
"success": False,
|
||||
"reasons": read_block,
|
||||
"permission_report": _permission_block_report("gitea.read"),
|
||||
}
|
||||
h, o, r = _resolve(remote, host, org, repo)
|
||||
auth = _auth(h)
|
||||
ctx = _gather_author_duplicate_work_context(
|
||||
issue_number,
|
||||
h=h,
|
||||
o=o,
|
||||
r=r,
|
||||
auth=auth,
|
||||
exclude_branch_name=exclude_branch_name,
|
||||
)
|
||||
result = author_duplicate_work_gate.classify_and_assess(
|
||||
ctx["issue"],
|
||||
stage=stage,
|
||||
comments=ctx["comments"],
|
||||
open_prs=ctx["open_prs"],
|
||||
branch_names=ctx["branch_names"],
|
||||
allow_stale_takeover=allow_stale_takeover,
|
||||
)
|
||||
assessment = result.get("duplicate_work") or {}
|
||||
return {
|
||||
"success": True,
|
||||
"performed": False,
|
||||
"issue_number": issue_number,
|
||||
"stage": stage,
|
||||
"allowed": assessment.get("allowed"),
|
||||
"block": assessment.get("block"),
|
||||
"eligibility_class": assessment.get("eligibility_class"),
|
||||
"outcome": assessment.get("outcome"),
|
||||
"linked_open_pr": assessment.get("linked_open_pr"),
|
||||
"matching_branches": assessment.get("matching_branches"),
|
||||
"claim_status": assessment.get("claim_status"),
|
||||
"reasons": assessment.get("reasons"),
|
||||
"safe_next_action": assessment.get("safe_next_action"),
|
||||
"claim": result.get("claim"),
|
||||
}
|
||||
|
||||
|
||||
@mcp.tool()
|
||||
def gitea_reconcile_issue_claims(
|
||||
state: str = "open",
|
||||
@@ -6807,9 +6852,6 @@ def gitea_resolve_task_capability(
|
||||
if reason_msg:
|
||||
result["reason"] = reason_msg
|
||||
role_session_router.sync_route_from_capability(result)
|
||||
if audit_reconciliation_mode.check_audit_task_enters_phase(task):
|
||||
phase_record = audit_reconciliation_mode.enter_audit_phase(task)
|
||||
result["reconciliation_phase"] = phase_record.get("phase")
|
||||
was_terminal = capability_stop_terminal.is_active()
|
||||
terminal = capability_stop_terminal.sync_from_capability_result(result)
|
||||
if terminal:
|
||||
|
||||
+19
-10
@@ -3622,18 +3622,33 @@ def assess_work_issue_mode_isolation(report_text: str) -> dict:
|
||||
}
|
||||
|
||||
|
||||
def assess_work_issue_duplicate_prevention_report(report_text, **kwargs):
|
||||
"""#400: work-issue reports must classify duplicate-work prevention."""
|
||||
from author_duplicate_work_gate import (
|
||||
assess_work_issue_duplicate_prevention_report as _assess,
|
||||
)
|
||||
|
||||
return _assess(report_text, **kwargs)
|
||||
|
||||
|
||||
def assess_work_issue_final_report(report_text: str) -> dict:
|
||||
"""#139: composite verifier for work-issue final reports."""
|
||||
checks = {
|
||||
"workflow_source": assess_work_issue_workflow_source(report_text),
|
||||
"mode_isolation": assess_work_issue_mode_isolation(report_text),
|
||||
"duplicate_prevention": assess_work_issue_duplicate_prevention_report(
|
||||
report_text
|
||||
),
|
||||
}
|
||||
|
||||
reasons = []
|
||||
downgraded = False
|
||||
for name, result in checks.items():
|
||||
verdict = result.get("verdict")
|
||||
if verdict in ("missing", "incomplete"):
|
||||
if result.get("block"):
|
||||
downgraded = True
|
||||
reasons.extend(result.get("reasons") or [])
|
||||
elif verdict in ("missing", "incomplete"):
|
||||
downgraded = True
|
||||
reasons.extend(result.get("reasons") or [])
|
||||
elif result.get("downgraded") or not result.get("complete", True):
|
||||
@@ -3641,6 +3656,9 @@ def assess_work_issue_final_report(report_text: str) -> dict:
|
||||
reasons.extend(
|
||||
f"{name}: {r}" for r in (result.get("reasons") or [])
|
||||
)
|
||||
elif result.get("proven") is False:
|
||||
downgraded = True
|
||||
reasons.extend(result.get("reasons") or [])
|
||||
|
||||
grade = "A" if not downgraded else "downgraded"
|
||||
return {
|
||||
@@ -5112,15 +5130,6 @@ def assess_pr_queue_cleanup_report(report_text: str | None) -> dict:
|
||||
return _assess(report_text or "")
|
||||
|
||||
|
||||
def assess_audit_reconciliation_report(report_text: str | None) -> dict:
|
||||
"""#419: validate audit vs cleanup reconciliation report boundaries."""
|
||||
from audit_reconciliation_mode import (
|
||||
assess_audit_reconciliation_report as _assess,
|
||||
)
|
||||
|
||||
return _assess(report_text or "")
|
||||
|
||||
|
||||
_GATE_PASSED_VALUE = re.compile(r"\bpassed\b", re.I)
|
||||
|
||||
_NOT_APPLICABLE_VALUE = re.compile(
|
||||
|
||||
@@ -304,40 +304,6 @@ If any required mutation capability is missing:
|
||||
* include safe next action (profile switch, human close, or dedicated reconciler
|
||||
profile)
|
||||
|
||||
## 15A. Audit vs cleanup phase (#419)
|
||||
|
||||
Reconciliation audits are **read-only** unless a separate cleanup phase is
|
||||
explicitly authorized.
|
||||
|
||||
**Audit phase forbids** (``audit_reconciliation_mode.check_audit_mutation_allowed``
|
||||
fails closed):
|
||||
|
||||
* ``gitea_delete_branch``
|
||||
* ``git branch -D``
|
||||
* ``git worktree remove``
|
||||
* pushes
|
||||
* issue/PR mutations
|
||||
* file edits
|
||||
|
||||
Dry-run merged-cleanup reconciliation (``gitea_reconcile_merged_cleanups`` with
|
||||
``dry_run=True``) stays in audit phase. Execution requires:
|
||||
|
||||
1. Operator approval or workflow authorization
|
||||
2. Exact ``delete_branch`` capability proof (``gitea.branch.delete``)
|
||||
3. Proof branch/worktree is safe to remove
|
||||
4. Before/after state snapshot
|
||||
|
||||
Call ``gitea_authorize_reconciliation_cleanup_phase`` before any cleanup
|
||||
mutation. Final reports must not claim ``no mutations`` if cleanup occurred.
|
||||
Classify cleanup mutations as:
|
||||
|
||||
* remote branch deletion → **External-state mutations**
|
||||
* local branch deletion → **Git ref mutations**
|
||||
* worktree removal → **Cleanup mutations**
|
||||
|
||||
``audit_reconciliation_mode.assess_audit_reconciliation_report`` validates
|
||||
these boundaries in final reports.
|
||||
|
||||
## 16. Mutation classification
|
||||
|
||||
Use precise mutation categories in the final report:
|
||||
|
||||
@@ -277,6 +277,24 @@ Do not select an issue based only on memory from a previous session.
|
||||
|
||||
Before claiming or working on an issue, check whether there is already an open PR, branch, or active claim for that issue.
|
||||
|
||||
Run `gitea_assess_author_duplicate_work` at these stages and stop when `block` is true:
|
||||
|
||||
* `claim` — before `gitea_mark_issue`
|
||||
* `lock` — before `gitea_lock_issue`
|
||||
* `worktree` / `edit` — before creating a worktree or editing files
|
||||
* `commit` — immediately before `git commit`
|
||||
* `push` — immediately before `git push`
|
||||
* `create_pr` — immediately before `gitea_create_pr` (also enforced server-side)
|
||||
|
||||
`gitea_mark_issue`, `gitea_lock_issue`, and `gitea_create_pr` enforce the same gate server-side and fail closed.
|
||||
|
||||
If a concurrent open PR appears after work begins:
|
||||
|
||||
* before commit or push — stop and preserve local work without pushing
|
||||
* after push but before PR creation — produce a reconciliation handoff instead of opening a PR
|
||||
|
||||
Final reports must name the duplicate-work outcome (`duplicate PR prevented`, `duplicate branch prevented`, `duplicate commit prevented`, `duplicate push prevented`, `duplicate work not prevented`, or `reconciliation handoff`).
|
||||
|
||||
If an open PR already exists for the issue, do not implement duplicate work.
|
||||
|
||||
Classify the issue as:
|
||||
|
||||
@@ -104,10 +104,6 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
|
||||
"permission": "gitea.read",
|
||||
"role": "author",
|
||||
},
|
||||
"reconciliation_cleanup": {
|
||||
"permission": "gitea.branch.delete",
|
||||
"role": "author",
|
||||
},
|
||||
"work_issue": {
|
||||
"permission": "gitea.pr.create",
|
||||
"role": "author",
|
||||
|
||||
@@ -1,291 +0,0 @@
|
||||
"""Tests for audit vs cleanup reconciliation mode (#419)."""
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
from unittest.mock import patch
|
||||
|
||||
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
|
||||
|
||||
import audit_reconciliation_mode as arm
|
||||
import mcp_server
|
||||
from audit_reconciliation_mode import (
|
||||
AUDIT_FORBIDDEN_TASKS,
|
||||
PHASE_AUDIT,
|
||||
PHASE_CLEANUP,
|
||||
assess_audit_command_allowed,
|
||||
assess_audit_reconciliation_report,
|
||||
authorize_cleanup_phase,
|
||||
check_audit_mutation_allowed,
|
||||
check_cleanup_execution_allowed,
|
||||
classify_cleanup_mutation,
|
||||
clear_phase,
|
||||
enter_audit_phase,
|
||||
)
|
||||
from final_report_validator import assess_final_report_validator
|
||||
from review_proofs import assess_audit_reconciliation_report as proofs_assess
|
||||
from task_capability_map import required_permission, required_role
|
||||
|
||||
DELETE_PROFILE = {
|
||||
"profile_name": "prgs-author-delete",
|
||||
"allowed_operations": ["gitea.read", "gitea.branch.delete"],
|
||||
"forbidden_operations": [],
|
||||
"audit_label": "prgs-author-delete",
|
||||
}
|
||||
|
||||
READ_PROFILE = {
|
||||
"profile_name": "prgs-author",
|
||||
"allowed_operations": ["gitea.read", "gitea.issue.comment"],
|
||||
"forbidden_operations": ["gitea.branch.delete"],
|
||||
"audit_label": "prgs-author",
|
||||
}
|
||||
|
||||
READ_ENV = {
|
||||
"GITEA_MCP_CONFIG": os.path.join(
|
||||
os.path.dirname(__file__), "..", "profiles.json"
|
||||
),
|
||||
"GITEA_MCP_PROFILE": "prgs-author",
|
||||
}
|
||||
|
||||
|
||||
def _authorize_cleanup(**kwargs):
|
||||
defaults = {
|
||||
"operator_approved": True,
|
||||
"delete_capability_proven": True,
|
||||
"safety_proof": {"safe_to_delete_remote": True},
|
||||
"before_after_snapshot": {
|
||||
"before": "remote branch exists",
|
||||
"after": "remote branch absent",
|
||||
},
|
||||
}
|
||||
defaults.update(kwargs)
|
||||
return authorize_cleanup_phase(**defaults)
|
||||
|
||||
|
||||
def _cleanup_report(**overrides):
|
||||
base = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"Workflow source: workflows/reconcile-landed-pr.md\n"
|
||||
"Audit phase: read-only assessment\n"
|
||||
"Cleanup phase authorized: true\n"
|
||||
"Delete-branch capability proven: true\n"
|
||||
"Branch safe to remove: true\n"
|
||||
"Before/after state snapshot: remote branch feat/x present → absent\n"
|
||||
"External-state mutations: deleted remote branch feat/x\n"
|
||||
"Git ref mutations: none\n"
|
||||
"Cleanup mutations: removed worktree branches/feat-x\n"
|
||||
)
|
||||
for key, value in overrides.items():
|
||||
base = base.replace(key, value)
|
||||
return base
|
||||
|
||||
|
||||
class TestAuditPhaseGates(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile-landed-pr")
|
||||
|
||||
def tearDown(self):
|
||||
clear_phase()
|
||||
|
||||
def test_audit_blocks_delete_branch_task(self):
|
||||
allowed, reasons = check_audit_mutation_allowed("delete_branch")
|
||||
self.assertFalse(allowed)
|
||||
self.assertTrue(reasons)
|
||||
|
||||
def test_audit_blocks_worktree_shell_commands(self):
|
||||
allowed, reasons = assess_audit_command_allowed(
|
||||
"git worktree remove branches/feat-x"
|
||||
)
|
||||
self.assertFalse(allowed)
|
||||
self.assertTrue(reasons)
|
||||
|
||||
def test_audit_blocks_local_branch_delete_command(self):
|
||||
allowed, reasons = assess_audit_command_allowed("git branch -D feat/x")
|
||||
self.assertFalse(allowed)
|
||||
|
||||
def test_audit_allows_read_tasks(self):
|
||||
allowed, _ = check_audit_mutation_allowed("reconcile_landed_pr")
|
||||
self.assertTrue(allowed)
|
||||
|
||||
def test_forbidden_set_covers_issue_and_pr_mutations(self):
|
||||
for task in ("close_pr", "comment_issue", "commit_files", "push_branch"):
|
||||
self.assertIn(task, AUDIT_FORBIDDEN_TASKS)
|
||||
|
||||
|
||||
class TestCleanupAuthorization(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile_merged_cleanups")
|
||||
|
||||
def tearDown(self):
|
||||
clear_phase()
|
||||
|
||||
def test_cleanup_without_approval_blocked(self):
|
||||
result = authorize_cleanup_phase(
|
||||
delete_capability_proven=True,
|
||||
safety_proof={"safe_to_delete_remote": True},
|
||||
before_after_snapshot={"before": "a", "after": "b"},
|
||||
)
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_cleanup_without_capability_proof_blocked(self):
|
||||
result = _authorize_cleanup(delete_capability_proven=False)
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_cleanup_without_snapshot_blocked(self):
|
||||
result = _authorize_cleanup(before_after_snapshot={"before": "", "after": ""})
|
||||
self.assertFalse(result["authorized"])
|
||||
|
||||
def test_authorized_cleanup_switches_phase(self):
|
||||
result = _authorize_cleanup()
|
||||
self.assertTrue(result["authorized"])
|
||||
self.assertEqual(result["phase"], PHASE_CLEANUP)
|
||||
|
||||
def test_cleanup_execution_allowed_only_after_authorization(self):
|
||||
self.assertFalse(check_cleanup_execution_allowed()[0])
|
||||
_authorize_cleanup()
|
||||
self.assertTrue(check_cleanup_execution_allowed()[0])
|
||||
|
||||
|
||||
class TestReportVerifier(unittest.TestCase):
|
||||
def test_false_no_mutations_after_cleanup_blocked(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"No mutations performed.\n"
|
||||
"delete_remote_branch feat/dup\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertIn("no mutations", result["reasons"][0].lower())
|
||||
|
||||
def test_cleanup_without_authorization_fields_blocked(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"remove_local_worktree branches/feat-x\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_authorized_cleanup_report_passes(self):
|
||||
result = assess_audit_reconciliation_report(_cleanup_report())
|
||||
self.assertTrue(result["proven"])
|
||||
|
||||
def test_mutation_classification_enforced(self):
|
||||
report = (
|
||||
"Task mode: reconcile-landed-pr\n"
|
||||
"Cleanup phase authorized: true\n"
|
||||
"Delete-branch capability proven: true\n"
|
||||
"Branch safe to remove: true\n"
|
||||
"Before/after state snapshot: present\n"
|
||||
"remove_local_worktree branches/feat-x\n"
|
||||
)
|
||||
result = assess_audit_reconciliation_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
|
||||
def test_proofs_export_matches_module(self):
|
||||
report = "No mutations performed.\ndelete_remote_branch feat/dup"
|
||||
self.assertEqual(
|
||||
proofs_assess(report)["proven"],
|
||||
assess_audit_reconciliation_report(report)["proven"],
|
||||
)
|
||||
|
||||
def test_final_report_validator_includes_boundary_rule(self):
|
||||
report = "No mutations performed.\ndelete_remote_branch feat/dup"
|
||||
result = assess_final_report_validator(
|
||||
report_text=report,
|
||||
task_kind="reconcile_already_landed",
|
||||
)
|
||||
self.assertTrue(result["blocked"])
|
||||
rule_ids = [f["rule_id"] for f in result["findings"]]
|
||||
self.assertIn("reconcile.audit_cleanup_boundary", rule_ids)
|
||||
|
||||
|
||||
class TestMutationClassification(unittest.TestCase):
|
||||
def test_remote_delete_is_external_state(self):
|
||||
self.assertEqual(
|
||||
classify_cleanup_mutation("delete_remote_branch"),
|
||||
"external-state",
|
||||
)
|
||||
|
||||
def test_worktree_remove_is_cleanup(self):
|
||||
self.assertEqual(
|
||||
classify_cleanup_mutation("remove_local_worktree"),
|
||||
"cleanup",
|
||||
)
|
||||
|
||||
|
||||
class TestMcpGates(unittest.TestCase):
|
||||
def setUp(self):
|
||||
clear_phase()
|
||||
enter_audit_phase("reconcile_merged_cleanups")
|
||||
self.mock_api = patch("mcp_server.api_request").start()
|
||||
self.mock_auth = patch(
|
||||
"mcp_server.get_auth_header", return_value="token test"
|
||||
).start()
|
||||
|
||||
def tearDown(self):
|
||||
patch.stopall()
|
||||
clear_phase()
|
||||
mcp_server._IDENTITY_CACHE.clear()
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_delete_branch_blocked_in_audit_phase(self, _profile):
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
|
||||
self.assertFalse(result["success"])
|
||||
self.assertEqual(result["audit_phase"], PHASE_AUDIT)
|
||||
self.mock_api.assert_not_called()
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_reconcile_execute_blocked_without_cleanup_auth(self, _profile):
|
||||
with self.assertRaises(ValueError):
|
||||
mcp_server.gitea_reconcile_merged_cleanups(
|
||||
dry_run=False,
|
||||
execute_confirmed=False,
|
||||
remote="prgs",
|
||||
)
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=READ_PROFILE)
|
||||
def test_cleanup_auth_fails_without_delete_capability(self, _profile):
|
||||
result = mcp_server.gitea_authorize_reconciliation_cleanup_phase(
|
||||
operator_approved=True,
|
||||
delete_capability_proven=True,
|
||||
safe_to_delete_remote=True,
|
||||
before_state="exists",
|
||||
after_state="gone",
|
||||
)
|
||||
self.assertFalse(result["authorized"])
|
||||
self.assertFalse(result["delete_capability_verified"])
|
||||
|
||||
@patch.dict(os.environ, READ_ENV, clear=True)
|
||||
@patch("mcp_server.get_profile", return_value=DELETE_PROFILE)
|
||||
def test_delete_branch_allowed_after_cleanup_authorization(self, _profile):
|
||||
mcp_server.gitea_authorize_reconciliation_cleanup_phase(
|
||||
operator_approved=True,
|
||||
delete_capability_proven=True,
|
||||
safe_to_delete_remote=True,
|
||||
before_state="exists",
|
||||
after_state="gone",
|
||||
)
|
||||
mcp_server.record_preflight_check("whoami")
|
||||
mcp_server.record_preflight_check("capability", resolved_role="author")
|
||||
self.mock_api.return_value = {}
|
||||
result = mcp_server.gitea_delete_branch(branch="feat/dup", remote="prgs")
|
||||
self.assertTrue(result["success"])
|
||||
|
||||
|
||||
class TestTaskCapabilityMap(unittest.TestCase):
|
||||
def test_reconciliation_cleanup_maps_delete_permission(self):
|
||||
self.assertEqual(
|
||||
required_permission("reconciliation_cleanup"),
|
||||
"gitea.branch.delete",
|
||||
)
|
||||
self.assertEqual(required_role("reconciliation_cleanup"), "author")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,102 @@
|
||||
"""Tests for early author duplicate-work gate (#400)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from author_duplicate_work_gate import ( # noqa: E402
|
||||
ELIGIBILITY_OPEN_PR_EXISTS,
|
||||
assess_author_duplicate_work,
|
||||
assess_work_issue_duplicate_prevention_report,
|
||||
)
|
||||
|
||||
|
||||
def _open_pr(number: int = 397, issue: int = 395) -> dict:
|
||||
return {
|
||||
"number": number,
|
||||
"head": {"ref": f"feat/issue-{issue}-example"},
|
||||
"title": f"feat: example (Closes #{issue})",
|
||||
"body": f"Closes #{issue}",
|
||||
}
|
||||
|
||||
|
||||
class TestAuthorDuplicateWorkGate(unittest.TestCase):
|
||||
def test_clear_when_no_duplicates(self):
|
||||
result = assess_author_duplicate_work(
|
||||
400,
|
||||
stage="claim",
|
||||
open_prs=[],
|
||||
branch_names=["master", "feat/issue-399-other"],
|
||||
)
|
||||
self.assertTrue(result["allowed"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_open_pr_blocks_claim(self):
|
||||
result = assess_author_duplicate_work(
|
||||
395,
|
||||
stage="claim",
|
||||
open_prs=[_open_pr()],
|
||||
branch_names=[],
|
||||
)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(result["eligibility_class"], ELIGIBILITY_OPEN_PR_EXISTS)
|
||||
|
||||
def test_matching_branch_blocks_lock_not_create_pr(self):
|
||||
branches = ["feat/issue-400-early-duplicate-work-gate"]
|
||||
lock = assess_author_duplicate_work(
|
||||
400,
|
||||
stage="lock",
|
||||
open_prs=[],
|
||||
branch_names=branches,
|
||||
)
|
||||
self.assertFalse(lock["allowed"])
|
||||
|
||||
create_pr = assess_author_duplicate_work(
|
||||
400,
|
||||
stage="create_pr",
|
||||
open_prs=[],
|
||||
branch_names=branches,
|
||||
matching_branches=[],
|
||||
)
|
||||
self.assertTrue(create_pr["allowed"])
|
||||
|
||||
def test_open_pr_blocks_create_pr_stage(self):
|
||||
result = assess_author_duplicate_work(
|
||||
395,
|
||||
stage="create_pr",
|
||||
open_prs=[_open_pr()],
|
||||
branch_names=["feat/issue-395-proof-backed-review-handoff"],
|
||||
)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(result["outcome"], "duplicate_pr_prevented")
|
||||
|
||||
def test_push_stage_blocks_on_concurrent_pr(self):
|
||||
result = assess_author_duplicate_work(
|
||||
395,
|
||||
stage="push",
|
||||
open_prs=[_open_pr()],
|
||||
branch_names=[],
|
||||
)
|
||||
self.assertFalse(result["allowed"])
|
||||
self.assertEqual(result["outcome"], "duplicate_push_prevented")
|
||||
|
||||
def test_duplicate_prevention_report_requires_outcome(self):
|
||||
bad = assess_work_issue_duplicate_prevention_report(
|
||||
"Duplicate work detected for issue #395."
|
||||
)
|
||||
self.assertFalse(bad["proven"])
|
||||
|
||||
good = assess_work_issue_duplicate_prevention_report(
|
||||
"Duplicate PR prevented; reconciliation handoff produced."
|
||||
)
|
||||
self.assertTrue(good["proven"])
|
||||
|
||||
def test_exported_from_review_proofs(self):
|
||||
from review_proofs import assess_work_issue_duplicate_prevention_report as exported
|
||||
|
||||
self.assertTrue(callable(exported))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -87,14 +87,6 @@ def test_reconcile_landed_workflow_contract():
|
||||
assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text
|
||||
assert "RECOVERY_HANDOFF_ONLY" in text
|
||||
assert "resolve_partial_reconciliation_plan" in text
|
||||
assert "check_audit_mutation_allowed" in text
|
||||
assert "gitea_authorize_reconciliation_cleanup_phase" in text
|
||||
|
||||
|
||||
def test_audit_reconciliation_verifier_exported():
|
||||
from review_proofs import assess_audit_reconciliation_report
|
||||
|
||||
assert callable(assess_audit_reconciliation_report)
|
||||
|
||||
|
||||
def test_create_issue_workflow_contract():
|
||||
@@ -103,6 +95,12 @@ def test_create_issue_workflow_contract():
|
||||
assert "## 9. Duplicate search before mutation" in text
|
||||
|
||||
|
||||
def test_author_duplicate_work_gate_exported():
|
||||
from review_proofs import assess_work_issue_duplicate_prevention_report
|
||||
|
||||
assert callable(assess_work_issue_duplicate_prevention_report)
|
||||
|
||||
|
||||
def test_work_issue_workflow_contract():
|
||||
text = (SKILL_DIR / "workflows" / "work-issue.md").read_text(encoding="utf-8")
|
||||
assert "canonical: true" in text
|
||||
|
||||
Reference in New Issue
Block a user