diff --git a/final_report_validator.py b/final_report_validator.py index 28e62b1..d7c32ed 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -12,6 +12,7 @@ import re from typing import Any, Callable import issue_lock_provenance +from mcp_native_cleanup_proof import assess_mcp_native_cleanup_proof from post_merge_cleanup_proof import assess_post_merge_cleanup_proof from review_proofs import ( HANDOFF_HEADING, @@ -1158,12 +1159,30 @@ def _rule_reviewer_post_merge_cleanup_proof(report_text: str) -> list[dict[str, ) +def _rule_shared_mcp_native_cleanup_proof(report_text: str) -> list[dict[str, str]]: + result = assess_mcp_native_cleanup_proof(report_text) + if not result.get("block"): + return [] + return _findings_from_reasons( + "shared.mcp_native_cleanup_proof", + result.get("reasons") or [], + field="Cleanup mutations", + severity="block", + safe_next_action=result.get("safe_next_action") + or "use authorized reconciler MCP cleanup tools; never raw scripts", + ) + + _SHARED_ISSUE_LOCK_RULES = ( _rule_shared_issue_lock_external_state, _rule_shared_manual_lock_pr_override, _rule_shared_author_reviewer_same_run, ) +_SHARED_CLEANUP_PROOF_RULES = ( + _rule_shared_mcp_native_cleanup_proof, +) + _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { "review_pr": [ _rule_shared_controller_handoff, @@ -1189,12 +1208,14 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_mutation_ledger, _rule_reviewer_review_mutation, _rule_reviewer_post_merge_cleanup_proof, + *_SHARED_CLEANUP_PROOF_RULES, _rule_reviewer_stale_head_proof, ], "reconcile_already_landed": [ _rule_reconcile_controller_handoff, _rule_shared_email_disclosure, *_SHARED_ISSUE_LOCK_RULES, + *_SHARED_CLEANUP_PROOF_RULES, _rule_reconcile_stale_author_fields, _rule_reconcile_eligible_reviewed, _rule_reconcile_linked_issue_live, diff --git a/mcp_native_cleanup_proof.py b/mcp_native_cleanup_proof.py new file mode 100644 index 0000000..70f2a5e --- /dev/null +++ b/mcp_native_cleanup_proof.py @@ -0,0 +1,359 @@ +"""MCP-native post-merge cleanup proof verifier (#517). + +Post-merge cleanup of leases, comments, branches, and worktrees must be +proven through explicit MCP tools (or approved helpers), never raw scripts or +ad hoc git/API commands. Merger/reviewer sessions must hand cleanup to a +reconciler profile with the right capability proof. +""" + +from __future__ import annotations + +import re +from typing import Any + +AUTHORIZED_CLEANUP_TOOLS = frozenset({ + "gitea_cleanup_post_merge_moot_lease", + "gitea_reconcile_merged_cleanups", + "gitea_delete_branch", + "gitea_cleanup_merged_pr_branch", + "gitea_audit_worktree_cleanup", + "gitea_capture_branches_worktree_snapshot", + "gitea_assess_worktree_cleanup_integrity", + "gitea_authorize_reconciliation_cleanup_phase", +}) + +_RAW_BRANCH_DELETE_PATTERNS = ( + re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+branch\s+-[dD]\b[^\n\r]*", re.I), + re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s--delete\b[^\n\r]*", re.I), + re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+push\b[^\n\r]*\s:[^\s`]+", re.I), +) + +_RAW_COMMENT_DELETE_PATTERNS = ( + re.compile( + r"(?:curl|wget|httpie)\b[^\n\r]*\b(?:DELETE|delete)\b[^\n\r]*" + r"(?:/comments/|issues/\d+/comments)", + re.I, + ), + re.compile( + r"\bDELETE\b[^\n\r]*/repos/[^\n\r]*/issues/\d+/comments/\d+", + re.I, + ), + re.compile( + r"\b(?:delete_issue_comment|remove_issue_comment|purge_comments?)\b", + re.I, + ), + re.compile( + r"\b(?:raw|ad hoc|adhoc)\b[^\n\r]{0,40}\bcomment\b[^\n\r]{0,40}\bdelete", + re.I, + ), +) + +_CLEANUP_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*cleanup mutations\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_GIT_REF_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*git ref mutations\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_WORKTREE_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*worktree mutations\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_MERGE_MUTATIONS_RE = re.compile( + r"^\s*[-*]?\s*merge mutations\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_NONE_LEDGER_VALUES = frozenset({"", "none", "n/a"}) +_MUTATION_LEDGER_PATTERNS = ( + _CLEANUP_MUTATIONS_RE, + _GIT_REF_MUTATIONS_RE, + _WORKTREE_MUTATIONS_RE, +) +_RAW_WORKTREE_REMOVE_PATTERNS = ( + re.compile(r"\bgit(?:\s+-C\s+\S+)?\s+worktree\s+remove\b", re.I), +) +_AUTHORIZED_TOOL_RE = re.compile( + r"\b(" + "|".join(re.escape(t) for t in sorted(AUTHORIZED_CLEANUP_TOOLS)) + r")\b", + re.IGNORECASE, +) +_RECONCILER_CAPABILITY_RE = re.compile( + r"(?:reconciler|gitea\.branch\.delete|delete_branch|reconcile_merged_cleanups)", + re.IGNORECASE, +) +_MERGER_CLEANUP_ROLE_RE = re.compile( + r"(?:merger|reviewer).{0,80}(?:deleted|removed|cleaned).{0,80}" + r"(?:branch|worktree|comment|lease)", + re.IGNORECASE, +) +_HANDOFF_TO_RECONCILER_RE = re.compile( + r"(?:hand(?:ed)? off|defer(?:red)?|next actor).{0,60}reconciler", + re.IGNORECASE, +) + + +def _ledger_field_body(text: str, pattern: re.Pattern[str]) -> str | None: + match = pattern.search(text) + if not match: + return None + return (match.group(1) or "").strip() + + +def _is_none_ledger_value(value: str) -> bool: + return value.strip().lower() in _NONE_LEDGER_VALUES + + +def scoped_cleanup_mutation_ledger_text(text: str | None) -> str: + """Return mutation-ledger bodies scoped to cleanup enforcement (#517).""" + text = text or "" + parts: list[str] = [] + for pattern in _MUTATION_LEDGER_PATTERNS: + body = _ledger_field_body(text, pattern) + if body is not None and not _is_none_ledger_value(body): + parts.append(body) + return "\n".join(parts) + + +def _git_ref_cleanup_claimed(body: str) -> bool: + return bool(raw_branch_delete_commands(body)) + + +def _worktree_cleanup_claimed(body: str) -> bool: + for pattern in _RAW_WORKTREE_REMOVE_PATTERNS: + if pattern.search(body): + return True + return bool( + re.search( + r"\b(?:removed|deleted)\b[^\n]{0,40}\bworktree\b", + body, + re.IGNORECASE, + ) + ) + + +def _cleanup_claiming_ledger_fields(text: str) -> list[tuple[str, str]]: + claiming: list[tuple[str, str]] = [] + cleanup_body = _ledger_field_body(text, _CLEANUP_MUTATIONS_RE) + if cleanup_body is not None and not _is_none_ledger_value(cleanup_body): + claiming.append(("Cleanup mutations", cleanup_body)) + + git_ref_body = _ledger_field_body(text, _GIT_REF_MUTATIONS_RE) + if git_ref_body is not None and not _is_none_ledger_value(git_ref_body): + if _git_ref_cleanup_claimed(git_ref_body): + claiming.append(("Git ref mutations", git_ref_body)) + + worktree_body = _ledger_field_body(text, _WORKTREE_MUTATIONS_RE) + if worktree_body is not None and not _is_none_ledger_value(worktree_body): + if _worktree_cleanup_claimed(worktree_body): + claiming.append(("Worktree mutations", worktree_body)) + return claiming + + +def raw_branch_delete_commands(text: str | None) -> list[str]: + """Return raw git branch-delete commands cited in *text*.""" + if not text: + return [] + commands: list[str] = [] + for pattern in _RAW_BRANCH_DELETE_PATTERNS: + commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text)) + return list(dict.fromkeys(commands)) + + +def raw_comment_delete_commands(text: str | None) -> list[str]: + """Return raw comment-deletion commands/scripts cited in *text*.""" + if not text: + return [] + commands: list[str] = [] + for pattern in _RAW_COMMENT_DELETE_PATTERNS: + commands.extend(match.group(0).strip("` ") for match in pattern.finditer(text)) + return list(dict.fromkeys(commands)) + + +def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]: + """Fail closed when mutation ledgers cite raw git branch deletion.""" + commands = raw_branch_delete_commands(scoped_cleanup_mutation_ledger_text(text)) + reasons = [ + ( + "raw git branch deletion bypasses MCP branch.delete cleanup gates: " + f"{command}" + ) + for command in commands + ] + return { + "proven": not reasons, + "block": bool(reasons), + "commands": commands, + "reasons": reasons, + "safe_next_action": ( + "use gitea_delete_branch, gitea_reconcile_merged_cleanups, or another " + "approved MCP cleanup helper with explicit branch.delete capability" + if reasons + else "proceed" + ), + } + + +def assess_raw_comment_delete_report(text: str | None) -> dict[str, Any]: + """Fail closed when mutation ledgers cite raw comment deletion.""" + commands = raw_comment_delete_commands(scoped_cleanup_mutation_ledger_text(text)) + reasons = [ + ( + "raw comment deletion bypasses MCP lease cleanup gates: " + f"{command}" + ) + for command in commands + ] + return { + "proven": not reasons, + "block": bool(reasons), + "commands": commands, + "reasons": reasons, + "safe_next_action": ( + "use gitea_cleanup_post_merge_moot_lease (append-only release comment) " + "or hand cleanup to a reconciler session; never delete lease comments" + if reasons + else "proceed" + ), + } + + +def assess_merge_cleanup_mutation_separation(text: str | None) -> dict[str, Any]: + """Require merge mutations and cleanup mutations in separate ledger fields.""" + text = text or "" + merge_match = _MERGE_MUTATIONS_RE.search(text) + cleanup_match = _CLEANUP_MUTATIONS_RE.search(text) + reasons: list[str] = [] + + if cleanup_match and not merge_match: + combined = re.search( + r"merge.{0,40}cleanup mutations|cleanup.{0,40}merge mutations", + text, + re.IGNORECASE, + ) + if combined: + reasons.append( + "merge and cleanup mutations must use separate 'Merge mutations' " + "and 'Cleanup mutations' ledger fields" + ) + + if merge_match and cleanup_match: + merge_val = (merge_match.group(1) or "").strip().lower() + cleanup_val = (cleanup_match.group(1) or "").strip().lower() + if merge_val == cleanup_val and merge_val not in {"", "none"}: + reasons.append( + "merge mutations and cleanup mutations must not duplicate the " + "same ledger entry" + ) + + return { + "proven": not reasons, + "block": bool(reasons), + "reasons": reasons, + "safe_next_action": ( + "split merge mutations (gitea_merge_pr) from cleanup mutations " + "(reconciler MCP tools) in the controller handoff" + if reasons + else "proceed" + ), + } + + +def assess_authorized_reconciler_cleanup_path(text: str | None) -> dict[str, Any]: + """Validate cleanup claims cite authorized MCP tools and reconciler capability.""" + text = text or "" + claiming = _cleanup_claiming_ledger_fields(text) + if not claiming: + return { + "proven": True, + "block": False, + "reasons": [], + "safe_next_action": "proceed", + } + + reasons: list[str] = [] + for field_name, body in claiming: + if not _AUTHORIZED_TOOL_RE.search(body): + reasons.append( + f"{field_name} cleanup must name an authorized MCP cleanup tool " + f"({', '.join(sorted(AUTHORIZED_CLEANUP_TOOLS))})" + ) + if not _RECONCILER_CAPABILITY_RE.search(text): + reasons.append( + "post-merge cleanup requires reconciler capability proof " + "(reconciler profile, gitea.branch.delete, or reconcile_merged_cleanups)" + ) + + return { + "proven": not reasons, + "block": bool(reasons), + "reasons": reasons, + "safe_next_action": ( + "hand cleanup to a prgs-reconciler session and cite the exact MCP tool " + "plus delete_branch/reconcile_merged_cleanups capability proof" + if reasons + else "proceed" + ), + } + + +def assess_merger_cleanup_handoff_guidance(text: str | None) -> dict[str, Any]: + """Merger sessions that performed cleanup must hand off to reconciler.""" + text = text or "" + if not _MERGER_CLEANUP_ROLE_RE.search(text): + return { + "proven": True, + "block": False, + "reasons": [], + "safe_next_action": "proceed", + } + if _HANDOFF_TO_RECONCILER_RE.search(text): + return { + "proven": True, + "block": False, + "reasons": [], + "safe_next_action": "proceed", + } + return { + "proven": False, + "block": True, + "reasons": [ + "merger/reviewer session performed cleanup but did not hand off to " + "reconciler for MCP-native cleanup" + ], + "safe_next_action": ( + "merger sessions must end with cleanup handed to prgs-reconciler; " + "do not perform ad hoc branch/comment/worktree cleanup as merger" + ), + } + + +def assess_mcp_native_cleanup_proof(report_text: str | None) -> dict[str, Any]: + """Composite #517 verifier for MCP-native post-merge cleanup proof.""" + text = report_text or "" + checks = ( + assess_raw_branch_delete_report(text), + assess_raw_comment_delete_report(text), + assess_merge_cleanup_mutation_separation(text), + assess_authorized_reconciler_cleanup_path(text), + assess_merger_cleanup_handoff_guidance(text), + ) + reasons: list[str] = [] + safe_next = "proceed" + for result in checks: + reasons.extend(result.get("reasons") or []) + if result.get("block") and result.get("safe_next_action"): + safe_next = result["safe_next_action"] + + block = bool(reasons) + return { + "proven": not block, + "block": block, + "reasons": reasons, + "safe_next_action": safe_next, + "raw_branch_commands": raw_branch_delete_commands( + scoped_cleanup_mutation_ledger_text(text) + ), + "raw_comment_commands": raw_comment_delete_commands( + scoped_cleanup_mutation_ledger_text(text) + ), + } \ No newline at end of file diff --git a/skills/llm-project-workflow/templates/merge-pr.md b/skills/llm-project-workflow/templates/merge-pr.md index ae99544..a0ee00d 100644 --- a/skills/llm-project-workflow/templates/merge-pr.md +++ b/skills/llm-project-workflow/templates/merge-pr.md @@ -41,7 +41,15 @@ Steps: 8. Confirm remote master now contains the merge commit (or the expected changes if squash merged). *Note: Gitea PR "closed" state is NOT equivalent to "merged". Do not assume a closed PR succeeded without verifying the actual landed changes.* -Then run the cleanup template (worktree-cleanup.md): +Post-merge cleanup (#517): merger sessions must NOT perform ad hoc cleanup. +- Record merge mutations separately from cleanup mutations in the controller handoff. +- Hand cleanup to a `prgs-reconciler` session — never raw `git branch -d`, + `git push --delete`, curl/API comment deletion, or local scripts. +- Reconciler cleanup must cite authorized MCP tools + (`gitea_reconcile_merged_cleanups`, `gitea_cleanup_post_merge_moot_lease`, + `gitea_delete_branch`, etc.) plus `gitea.branch.delete` capability proof. + +Then run the cleanup template (worktree-cleanup.md) in a reconciler session: - Verify expected file/commit presence on master (post-merge file-presence verification): - Run: git fetch --prune; git checkout master; git pull master --ff-only - Verify that the expected files added/modified in the PR are present on master (or absent if deleted). diff --git a/skills/llm-project-workflow/templates/post-merge-cleanup-handoff.md b/skills/llm-project-workflow/templates/post-merge-cleanup-handoff.md new file mode 100644 index 0000000..37a0ba6 --- /dev/null +++ b/skills/llm-project-workflow/templates/post-merge-cleanup-handoff.md @@ -0,0 +1,27 @@ +# Template: post-merge cleanup (reconciler only) + +Copy, fill the `<...>` fields, and paste as the task prompt. Run only after merge +is confirmed on remote master. Merger sessions must hand off here — never perform +this cleanup inline (#517). + +```text +Task: MCP-native post-merge cleanup for PR # / issue #. + +Rules: +- Active profile must be reconciler (`prgs-reconciler`) with `gitea.branch.delete`. +- Use MCP tools only — no raw git branch delete, no API comment deletion scripts. +- Record cleanup mutations separately from merge mutations in the handoff. + +Steps: +1. `gitea_resolve_task_capability(task="reconcile_merged_cleanups", remote=prgs)` +2. Confirm PR # merged on /master. +3. `gitea_cleanup_post_merge_moot_lease` if a reviewer lease remains (append-only). +4. `gitea_reconcile_merged_cleanups` for branch/worktree cleanup with dry-run first. +5. Report authorized cleanup tools used and reconciler capability proof. + +Handoff ledger (required fields): +- Merge mutations: (none — merger already recorded gitea_merge_pr) +- Cleanup mutations: list exact MCP tools invoked +- Reconciler capability: profile + gitea.branch.delete proof +- Next actor: controller acceptance or none +``` \ No newline at end of file diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index 4b4abb0..44b9f12 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -978,6 +978,27 @@ If any gate fails, report: Skipped cleanup with an exact blocker passes validation. Performed-cleanup claims without the checklist fail validation. +## 28B. MCP-native cleanup only (#517) + +Post-merge cleanup of leases, comments, branches, and worktrees must go through +explicit MCP tools — never raw git, curl/API scripts, or ad hoc helper scripts. + +Merger and reviewer sessions must **not** perform cleanup inline. Record: + +* **Merge mutations** — only `gitea_merge_pr` (or review mutations for review-only runs) +* **Cleanup mutations** — only authorized reconciler MCP tools, cited by exact tool name + +Authorized cleanup tools include `gitea_reconcile_merged_cleanups`, +`gitea_cleanup_post_merge_moot_lease`, `gitea_delete_branch`, and +`gitea_cleanup_merged_pr_branch` (when available). Lease cleanup uses +append-only release comments — never delete another session's lease comment. + +Hand cleanup to a `prgs-reconciler` session with `gitea.branch.delete` capability +proof. Raw `git branch -d`, `git push --delete`, and comment-deletion API calls +are blocked in final-report validation. + +Template: `templates/post-merge-cleanup-handoff.md`. + ## 29. Recovery handoff rules If blocked, produce a recovery handoff with: diff --git a/tests/test_mcp_native_cleanup_proof.py b/tests/test_mcp_native_cleanup_proof.py new file mode 100644 index 0000000..d965b0d --- /dev/null +++ b/tests/test_mcp_native_cleanup_proof.py @@ -0,0 +1,251 @@ +"""Tests for MCP-native post-merge cleanup proof (#517).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from final_report_validator import assess_final_report_validator # noqa: E402 +from mcp_native_cleanup_proof import ( # noqa: E402 + assess_authorized_reconciler_cleanup_path, + assess_mcp_native_cleanup_proof, + assess_merger_cleanup_handoff_guidance, + assess_raw_branch_delete_report, + assess_raw_comment_delete_report, +) + + +def _authorized_cleanup_report(**overrides): + fields = { + "Merge mutations": "gitea_merge_pr on PR #100", + "Cleanup mutations": ( + "gitea_reconcile_merged_cleanups deleted remote branch; " + "gitea_cleanup_post_merge_moot_lease released lease" + ), + "Reconciler capability": "prgs-reconciler / gitea.branch.delete resolved", + "Next actor": "reconciler session complete", + } + fields.update(overrides) + lines = ["## Controller Handoff", ""] + lines.extend(f"- {key}: {value}" for key, value in fields.items()) + return "\n".join(lines) + + +class TestRawBranchDeleteBlocked(unittest.TestCase): + def test_git_branch_d_blocked(self): + report = "Cleanup mutations: git branch -d feat/issue-1-x" + result = assess_raw_branch_delete_report(report) + self.assertTrue(result["block"]) + self.assertIn("git branch -d", result["commands"][0]) + + def test_git_push_delete_in_ledger_blocked(self): + report = "- Git ref mutations: git push origin --delete feat/x" + result = assess_raw_branch_delete_report(report) + self.assertTrue(result["block"]) + + def test_narrative_git_push_delete_not_blocked(self): + report = "\n".join([ + "## Validation", + "Workflow §28B forbids raw git push --delete for cleanup.", + "- Cleanup mutations: none", + "- Git ref mutations: none", + "- Worktree mutations: none", + ]) + result = assess_raw_branch_delete_report(report) + self.assertFalse(result["block"]) + + def test_authorized_mcp_path_passes(self): + report = _authorized_cleanup_report() + result = assess_raw_branch_delete_report(report) + self.assertFalse(result["block"]) + + +class TestRawCommentDeleteBlocked(unittest.TestCase): + def test_curl_delete_comment_in_ledger_blocked(self): + report = ( + "- Cleanup mutations: curl -X DELETE https://gitea.example/api/v1/repos/o/r/" + "issues/9/comments/42" + ) + result = assess_raw_comment_delete_report(report) + self.assertTrue(result["block"]) + + def test_narrative_comment_delete_not_blocked(self): + report = "\n".join([ + "Files reviewed: post_merge_cleanup_proof.py (raw comment delete patterns)", + "Validation note: never use delete_issue_comment for lease cleanup.", + "- Cleanup mutations: none", + "- Git ref mutations: none", + "- Worktree mutations: none", + ]) + result = assess_raw_comment_delete_report(report) + self.assertFalse(result["block"]) + + def test_delete_issue_comment_helper_in_ledger_blocked(self): + report = "- Cleanup mutations: delete_issue_comment purged lease comments" + result = assess_raw_comment_delete_report(report) + self.assertTrue(result["block"]) + + def test_moot_lease_append_only_passes(self): + report = _authorized_cleanup_report() + result = assess_raw_comment_delete_report(report) + self.assertFalse(result["block"]) + + +class TestAuthorizedReconcilerPath(unittest.TestCase): + def test_cleanup_without_mcp_tool_blocked(self): + report = "\n".join([ + "## Controller Handoff", + "- Cleanup mutations: removed branch manually after merge", + "- Reconciler capability: prgs-reconciler", + ]) + result = assess_authorized_reconciler_cleanup_path(report) + self.assertTrue(result["block"]) + self.assertTrue(any("authorized MCP" in r for r in result["reasons"])) + + def test_authorized_cleanup_succeeds(self): + result = assess_authorized_reconciler_cleanup_path(_authorized_cleanup_report()) + self.assertFalse(result["block"]) + + def test_git_ref_cleanup_without_mcp_tool_blocked(self): + report = "\n".join([ + "## Controller Handoff", + "- Cleanup mutations: none", + "- Git ref mutations: git branch -D feat/issue-517-test", + "- Reconciler capability: prgs-reconciler", + ]) + result = assess_authorized_reconciler_cleanup_path(report) + self.assertTrue(result["block"]) + self.assertTrue( + any("Git ref mutations cleanup must name" in r for r in result["reasons"]) + ) + + def test_worktree_cleanup_without_mcp_tool_blocked(self): + report = "\n".join([ + "## Controller Handoff", + "- Cleanup mutations: none", + "- Worktree mutations: git worktree remove branches/review-pr542", + "- Reconciler capability: prgs-reconciler", + ]) + result = assess_authorized_reconciler_cleanup_path(report) + self.assertTrue(result["block"]) + self.assertTrue( + any("Worktree mutations cleanup must name" in r for r in result["reasons"]) + ) + + def test_git_fetch_in_git_ref_mutations_not_cleanup_claim(self): + report = "\n".join([ + "## Controller Handoff", + "- Cleanup mutations: none", + "- Git ref mutations: git fetch prgs master", + ]) + result = assess_authorized_reconciler_cleanup_path(report) + self.assertFalse(result["block"]) + + +class TestMergerHandoffGuidance(unittest.TestCase): + def test_merger_cleanup_without_handoff_blocked(self): + report = "Merger deleted remote branch and removed worktree after merge" + result = assess_merger_cleanup_handoff_guidance(report) + self.assertTrue(result["block"]) + + def test_merger_cleanup_with_reconciler_handoff_passes(self): + report = ( + "Merger deleted remote branch; next actor: reconciler for worktree audit" + ) + result = assess_merger_cleanup_handoff_guidance(report) + self.assertFalse(result["block"]) + + +class TestCompositeVerifier(unittest.TestCase): + def test_full_authorized_report_passes(self): + result = assess_mcp_native_cleanup_proof(_authorized_cleanup_report()) + self.assertFalse(result["block"], result["reasons"]) + + def test_raw_bypasses_fail_composite(self): + report = "\n".join([ + "## Controller Handoff", + "- Merge mutations: gitea_merge_pr", + "- Cleanup mutations: git branch -D feat/x; curl -X DELETE .../comments/1", + ]) + result = assess_mcp_native_cleanup_proof(report) + self.assertTrue(result["block"]) + self.assertGreaterEqual(len(result["reasons"]), 2) + + def test_narrative_cleanup_mentions_pass_when_ledgers_none(self): + report = "\n".join([ + "## Review summary", + "Validated workflow §28B MCP-native cleanup guidance.", + "Files reviewed describe raw git branch -d and delete_issue_comment blocks.", + "", + "## Controller Handoff", + "- Cleanup mutations: none", + "- Git ref mutations: none", + "- Worktree mutations: none", + ]) + result = assess_mcp_native_cleanup_proof(report) + self.assertFalse(result["block"], result["reasons"]) + + def test_git_ref_cleanup_bypass_blocked_in_composite(self): + report = "\n".join([ + "## Controller Handoff", + "- Cleanup mutations: none", + "- Git ref mutations: git branch -D feat/issue-517-test", + "- Reconciler capability: prgs-reconciler", + ]) + result = assess_mcp_native_cleanup_proof(report) + self.assertTrue(result["block"]) + self.assertTrue( + any("Git ref mutations cleanup must name" in r for r in result["reasons"]) + ) + + +class TestFinalReportValidatorIntegration(unittest.TestCase): + def test_validator_blocks_raw_branch_delete_in_review_report(self): + report = "\n".join([ + "## Controller Handoff", + "- Cleanup mutations: git branch -d feat/issue-517-test", + ]) + result = assess_final_report_validator(report, "review_pr") + rules = {f["rule_id"] for f in result["findings"]} + self.assertIn("shared.mcp_native_cleanup_proof", rules) + + def test_validator_passes_authorized_cleanup_report(self): + result = assess_final_report_validator(_authorized_cleanup_report(), "review_pr") + blocked = [f for f in result["findings"] if f.get("severity") == "block"] + mcp_blocks = [ + f for f in blocked if f.get("rule_id") == "shared.mcp_native_cleanup_proof" + ] + self.assertEqual(mcp_blocks, []) + + def test_validator_passes_narrative_cleanup_mentions_with_none_ledgers(self): + report = "\n".join([ + "## Review summary", + "Workflow §28B documents raw git push --delete and delete_issue_comment blocks.", + "", + "## Controller Handoff", + "- Cleanup mutations: none", + "- Git ref mutations: none", + "- Worktree mutations: none", + ]) + result = assess_final_report_validator(report, "review_pr") + mcp_blocks = [ + f for f in result["findings"] + if f.get("rule_id") == "shared.mcp_native_cleanup_proof" + and f.get("severity") == "block" + ] + self.assertEqual(mcp_blocks, []) + + def test_validator_blocks_git_ref_cleanup_bypass(self): + report = "\n".join([ + "## Controller Handoff", + "- Cleanup mutations: none", + "- Git ref mutations: git branch -D feat/issue-517-test", + "- Reconciler capability: prgs-reconciler", + ]) + result = assess_final_report_validator(report, "reconcile_already_landed") + rules = {f["rule_id"] for f in result["findings"] if f.get("severity") == "block"} + self.assertIn("shared.mcp_native_cleanup_proof", rules) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file