diff --git a/review_proofs.py b/review_proofs.py index ee81e5e..43dfbd3 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -4453,6 +4453,13 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None): } +def assess_merge_simulation_report(report_text, **kwargs): + """#317: classify local merge simulation as worktree/index mutations.""" + from reviewer_merge_simulation import assess_merge_simulation_report as _assess + + return _assess(report_text, **kwargs) + + def assess_prior_blocker_skip_proof(report_text, **kwargs): """#318: require live blocker proof before skipping earlier open PRs.""" from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess diff --git a/reviewer_merge_simulation.py b/reviewer_merge_simulation.py new file mode 100644 index 0000000..5397c4f --- /dev/null +++ b/reviewer_merge_simulation.py @@ -0,0 +1,148 @@ +"""Merge-simulation worktree mutation verifier for reviewer reports (#317).""" + +from __future__ import annotations + +import re +from typing import Any + +_WORKTREE_INDEX_FIELD_RE = re.compile( + r"^\s*worktree/index mutations\s*:\s*(.+?)\s*$", + re.IGNORECASE | re.MULTILINE, +) +_READONLY_DIAG_RE = re.compile( + r"read[- ]only diagnostics\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_WORKTREE_PATH_RE = re.compile( + r"(?:worktree path|diagnostic worktree|simulation worktree)\s*:", + re.IGNORECASE, +) +_PRE_CLEAN_RE = re.compile( + r"(?:pre[- ]simulation|before simulation|clean before).*(?:clean|status)", + re.IGNORECASE, +) +_MERGE_RESULT_RE = re.compile( + r"(?:merge result|simulation result|conflict status|merge conflict)", + re.IGNORECASE, +) +_ABORT_RE = re.compile( + r"(?:merge --abort|abort/reset command|abort command|git merge --abort)", + re.IGNORECASE, +) +_POST_CLEAN_RE = re.compile( + r"(?:post[- ]abort|after abort|clean after).*(?:clean|status)", + re.IGNORECASE, +) + + +def _command_text(entry: Any) -> str: + if isinstance(entry, dict): + return str(entry.get("command") or "").strip() + return str(entry or "").strip() + + +def _git_subcommand_line(command: str) -> str | None: + tokens = command.split() + if not tokens or tokens[0] != "git": + return None + return " ".join(tokens[1:]) + + +def merge_simulation_commands(command_log: list | None) -> list[str]: + """Return logged commands that perform local merge simulation (#317).""" + out: list[str] = [] + for entry in command_log or []: + command = _command_text(entry) + rest = _git_subcommand_line(command) + if rest is None: + continue + lower = command.lower() + if rest.startswith("merge"): + if "--no-commit" in lower or ( + "--no-ff" in lower and "merge" in lower and "--commit" not in lower + ): + out.append(command) + elif "--abort" in lower: + out.append(command) + return out + + +def assess_merge_simulation_report( + report_text: str, + *, + command_log: list | None = None, +) -> dict[str, Any]: + """Reject merge simulation classified as read-only; require worktree proof (#317).""" + text = report_text or "" + lower = text.lower() + reasons: list[str] = [] + simulation_commands = merge_simulation_commands(command_log) + has_abort = any("--abort" in c.lower() for c in simulation_commands) + has_simulation = any( + "--no-commit" in c.lower() or ( + "--no-ff" in c.lower() and "--abort" not in c.lower() + ) + for c in simulation_commands + ) + + if not simulation_commands: + return { + "proven": True, + "block": False, + "reasons": [], + "simulation_commands": [], + "safe_next_action": "proceed", + } + + field = _WORKTREE_INDEX_FIELD_RE.search(text) + value = (field.group(1).strip().lower() if field else "") or "" + if not value or value in {"none", "n/a", "no"}: + reasons.append( + "merge simulation commands ran but report lacks " + "'Worktree/index mutations' entry" + ) + elif "merge" not in value and "simulation" not in value: + if not any(cmd.lower() in lower for cmd in simulation_commands): + reasons.append( + "Worktree/index mutations must document merge simulation commands" + ) + + readonly = _READONLY_DIAG_RE.search(text) + if readonly: + readonly_value = readonly.group(1) + for command in simulation_commands: + if command.lower() in readonly_value.lower(): + reasons.append( + "merge simulation may not be listed under read-only diagnostics" + ) + if has_simulation and "merge simulation" in readonly_value.lower(): + reasons.append( + "merge simulation may not be classified as read-only diagnostics" + ) + + if has_simulation: + if not _WORKTREE_PATH_RE.search(text): + reasons.append("merge simulation report missing worktree path") + if not _PRE_CLEAN_RE.search(text): + reasons.append("merge simulation report missing pre-simulation clean status") + if not _MERGE_RESULT_RE.search(text): + reasons.append("merge simulation report missing merge result/conflict status") + if has_abort or has_simulation: + if has_abort and not _ABORT_RE.search(text): + reasons.append("merge simulation report missing abort/reset command proof") + if has_abort and not _POST_CLEAN_RE.search(text): + reasons.append("merge simulation report missing post-abort clean status") + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": reasons, + "simulation_commands": simulation_commands, + "safe_next_action": ( + "classify merge simulation under Worktree/index mutations with full " + "pre/merge/abort/post-clean proof; never list as read-only diagnostics" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 42ce8b9..0c2b89a 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -136,6 +136,12 @@ def test_create_issue_final_report_verifier_exported(): assert callable(assess_create_issue_final_report) +def test_merge_simulation_verifier_exported(): + from review_proofs import assess_merge_simulation_report + + assert callable(assess_merge_simulation_report) + + def test_prior_blocker_skip_verifier_exported(): from review_proofs import assess_prior_blocker_skip_proof diff --git a/tests/test_reviewer_merge_simulation.py b/tests/test_reviewer_merge_simulation.py new file mode 100644 index 0000000..e3101f8 --- /dev/null +++ b/tests/test_reviewer_merge_simulation.py @@ -0,0 +1,135 @@ +"""Tests for merge-simulation worktree mutation verifier (#317).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_merge_simulation import ( # noqa: E402 + assess_merge_simulation_report, + merge_simulation_commands, +) + + +def _good_simulation_report(**overrides) -> str: + lines = [ + "Worktree/index mutations: merge simulation in branches/review-pr281", + "Worktree path: branches/review-pr281", + "Pre-simulation clean status: clean (git status --porcelain empty)", + "Merge result: conflicts in review_proofs.py", + "Abort command: git merge --abort", + "Post-abort clean status: clean after abort (git status --porcelain empty)", + ] + text = "\n".join(lines) + for key, value in overrides.items(): + text = text.replace(f"{key}:", f"{key}: {value}") + return text + + +class TestMergeSimulationDetection(unittest.TestCase): + def test_detects_no_commit_merge(self): + commands = merge_simulation_commands([ + "git merge prgs/master --no-commit --no-ff", + "git status", + ]) + self.assertEqual(commands, ["git merge prgs/master --no-commit --no-ff"]) + + def test_detects_merge_abort(self): + commands = merge_simulation_commands([ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ]) + self.assertIn("git merge --abort", commands) + + +class TestMergeSimulationReport(unittest.TestCase): + def test_no_simulation_passes(self): + result = assess_merge_simulation_report( + "Review complete. Worktree/index mutations: none", + command_log=["git status --porcelain", "git diff prgs/master...HEAD"], + ) + self.assertTrue(result["proven"]) + + def test_documented_simulation_passes(self): + report = _good_simulation_report() + result = assess_merge_simulation_report( + report, + command_log=[ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ], + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_readonly_classification_blocks(self): + report = "\n".join([ + "Read-only diagnostics: git merge prgs/master --no-commit --no-ff, git status", + "Worktree/index mutations: none", + ]) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("read-only" in r.lower() for r in result["reasons"])) + + def test_missing_worktree_field_blocks(self): + report = "Merge simulation ran but only noted in prose." + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("worktree/index mutations" in r.lower() for r in result["reasons"])) + + def test_conflict_simulation_requires_merge_result(self): + report = _good_simulation_report().replace( + "Merge result: conflicts in review_proofs.py", + "", + ) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("merge result" in r.lower() for r in result["reasons"])) + + def test_aborted_simulation_requires_post_clean_proof(self): + report = _good_simulation_report().replace( + "Post-abort clean status: clean after abort (git status --porcelain empty)", + "", + ) + result = assess_merge_simulation_report( + report, + command_log=[ + "git merge prgs/master --no-commit --no-ff", + "git merge --abort", + ], + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("post-abort" in r.lower() for r in result["reasons"])) + + def test_successful_simulation_without_abort_omits_abort_fields(self): + report = "\n".join([ + "Worktree/index mutations: merge simulation completed cleanly", + "Worktree path: branches/review-pr281", + "Pre-simulation clean status: clean before simulation", + "Merge result: clean fast-forward simulation", + ]) + result = assess_merge_simulation_report( + report, + command_log=["git merge prgs/master --no-commit --no-ff"], + ) + self.assertTrue(result["proven"], result["reasons"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_merge_simulation_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file