From 749ca0438859e48e6760a7b459d78684dc811fd5 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Wed, 8 Jul 2026 13:43:39 -0400 Subject: [PATCH] fix: scope MCP cleanup proof to mutation ledgers (#517) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Raw branch/comment delete detection now scans only Cleanup mutations, Git ref mutations, and Worktree mutations ledger fields so narrative §28B citations and validation notes no longer false-positive. Authorized cleanup tool proof also applies when cleanup is claimed under Git ref or Worktree mutation ledgers while Cleanup mutations is none. Co-Authored-By: Claude Opus 4.8 (1M context) --- mcp_native_cleanup_proof.py | 114 +++++++++++++++++----- tests/test_mcp_native_cleanup_proof.py | 125 +++++++++++++++++++++++-- 2 files changed, 211 insertions(+), 28 deletions(-) diff --git a/mcp_native_cleanup_proof.py b/mcp_native_cleanup_proof.py index 8ee79aa..70f2a5e 100644 --- a/mcp_native_cleanup_proof.py +++ b/mcp_native_cleanup_proof.py @@ -52,10 +52,27 @@ _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, @@ -75,6 +92,63 @@ _HANDOFF_TO_RECONCILER_RE = re.compile( ) +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: @@ -96,8 +170,8 @@ def raw_comment_delete_commands(text: str | None) -> list[str]: def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]: - """Fail closed when a report uses raw git branch deletion as cleanup proof.""" - commands = raw_branch_delete_commands(text) + """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: " @@ -120,8 +194,8 @@ def assess_raw_branch_delete_report(text: str | None) -> dict[str, Any]: def assess_raw_comment_delete_report(text: str | None) -> dict[str, Any]: - """Fail closed when a report uses raw comment deletion as cleanup proof.""" - commands = raw_comment_delete_commands(text) + """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: " @@ -187,17 +261,8 @@ def assess_merge_cleanup_mutation_separation(text: str | None) -> dict[str, Any] 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 "" - cleanup_match = _CLEANUP_MUTATIONS_RE.search(text) - if not cleanup_match: - return { - "proven": True, - "block": False, - "reasons": [], - "safe_next_action": "proceed", - } - - cleanup_body = cleanup_match.group(1) or "" - if cleanup_body.strip().lower() in {"", "none", "n/a"}: + claiming = _cleanup_claiming_ledger_fields(text) + if not claiming: return { "proven": True, "block": False, @@ -206,11 +271,12 @@ def assess_authorized_reconciler_cleanup_path(text: str | None) -> dict[str, Any } reasons: list[str] = [] - if not _AUTHORIZED_TOOL_RE.search(cleanup_body): - reasons.append( - "cleanup mutations must name an authorized MCP cleanup tool " - f"({', '.join(sorted(AUTHORIZED_CLEANUP_TOOLS))})" - ) + 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 " @@ -284,6 +350,10 @@ def assess_mcp_native_cleanup_proof(report_text: str | None) -> dict[str, Any]: "block": block, "reasons": reasons, "safe_next_action": safe_next, - "raw_branch_commands": raw_branch_delete_commands(text), - "raw_comment_commands": raw_comment_delete_commands(text), + "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/tests/test_mcp_native_cleanup_proof.py b/tests/test_mcp_native_cleanup_proof.py index 686509b..d965b0d 100644 --- a/tests/test_mcp_native_cleanup_proof.py +++ b/tests/test_mcp_native_cleanup_proof.py @@ -38,11 +38,22 @@ class TestRawBranchDeleteBlocked(unittest.TestCase): self.assertTrue(result["block"]) self.assertIn("git branch -d", result["commands"][0]) - def test_git_push_delete_blocked(self): - report = "Ran git push origin --delete feat/x for cleanup" + 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) @@ -50,16 +61,27 @@ class TestRawBranchDeleteBlocked(unittest.TestCase): class TestRawCommentDeleteBlocked(unittest.TestCase): - def test_curl_delete_comment_blocked(self): + def test_curl_delete_comment_in_ledger_blocked(self): report = ( - "Cleanup: curl -X DELETE https://gitea.example/api/v1/repos/o/r/" + "- 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_delete_issue_comment_helper_blocked(self): - report = "Used delete_issue_comment script to purge lease comments" + 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"]) @@ -84,6 +106,41 @@ class TestAuthorizedReconcilerPath(unittest.TestCase): 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): @@ -114,6 +171,33 @@ class TestCompositeVerifier(unittest.TestCase): 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): @@ -133,6 +217,35 @@ class TestFinalReportValidatorIntegration(unittest.TestCase): ] 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