Merge pull request 'feat: classify merge simulation as worktree/index mutations (Closes #317)' (#356) from feat/issue-317-merge-simulation-mutations into master

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

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"
+        ),
+    }

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
 

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()