feat: gate reconciliation audit mode from unauthorized cleanup (Closes #419)

Add audit vs cleanup phase tracking so reconciliation audits stay read-only
unless cleanup is explicitly authorized with delete capability proof, safety
proof, and before/after snapshots. Block gitea_delete_branch and merged-cleanup
execution during audit phase, validate final reports for false no-mutations
claims, and document the boundary in the reconcile-landed-pr workflow.
This commit is contained in:
2026-07-07 12:03:08 -04:00
parent 1a1e679246
commit 126a983b69
8 changed files with 782 additions and 0 deletions
+75
View File
@@ -506,6 +506,7 @@ import issue_claim_heartbeat # 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
@@ -3478,6 +3479,18 @@ 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)
@@ -3547,6 +3560,18 @@ 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)
@@ -3629,6 +3654,53 @@ 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,
@@ -6735,6 +6807,9 @@ 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: