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:
@@ -546,6 +546,7 @@ import issue_work_duplicate_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
|
||||
|
||||
@@ -3680,6 +3681,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)
|
||||
@@ -3749,6 +3762,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)
|
||||
@@ -3831,6 +3856,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,
|
||||
@@ -6937,6 +7009,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:
|
||||
|
||||
Reference in New Issue
Block a user