From ddaf380db2c674f4dadaeaed1fadbe4488bc31f9 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 12:03:08 -0400 Subject: [PATCH 1/2] 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. --- audit_reconciliation_mode.py | 344 ++++++++++++++++++ final_report_validator.py | 17 + gitea_mcp_server.py | 75 ++++ review_proofs.py | 9 + .../workflows/reconcile-landed-pr.md | 34 ++ task_capability_map.py | 4 + tests/test_audit_reconciliation_mode.py | 291 +++++++++++++++ tests/test_llm_workflow_split.py | 8 + 8 files changed, 782 insertions(+) create mode 100644 audit_reconciliation_mode.py create mode 100644 tests/test_audit_reconciliation_mode.py diff --git a/audit_reconciliation_mode.py b/audit_reconciliation_mode.py new file mode 100644 index 0000000..52d0244 --- /dev/null +++ b/audit_reconciliation_mode.py @@ -0,0 +1,344 @@ +"""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, [] \ No newline at end of file diff --git a/final_report_validator.py b/final_report_validator.py index bbc4b6c..0dae1a2 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -919,6 +919,22 @@ def _rule_shared_author_reviewer_same_run(report_text: str) -> list[dict[str, st ) +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, *, @@ -977,6 +993,7 @@ _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, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 17938e3..2fd613c 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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: diff --git a/review_proofs.py b/review_proofs.py index b74ebb8..9a37277 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -5115,6 +5115,15 @@ 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( diff --git a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md index 10ba60d..a9f4cd1 100644 --- a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md +++ b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md @@ -304,6 +304,40 @@ 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: diff --git a/task_capability_map.py b/task_capability_map.py index 1d40a78..9e7d0ba 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -104,6 +104,10 @@ 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", diff --git a/tests/test_audit_reconciliation_mode.py b/tests/test_audit_reconciliation_mode.py new file mode 100644 index 0000000..a117fab --- /dev/null +++ b/tests/test_audit_reconciliation_mode.py @@ -0,0 +1,291 @@ +"""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() \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 44d80ce..d5f1fcb 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -87,6 +87,14 @@ 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(): -- 2.43.7 From 551ee7d70c7968d3c43015084faa13b0b56a46e2 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 17:20:53 -0400 Subject: [PATCH 2/2] fix: resolve conflicts for PR #421 -- 2.43.7