feat: classify merge simulation as worktree/index mutations (Closes #317)
Add assess_merge_simulation_report to reject local merge simulations listed as read-only diagnostics and require Worktree/index mutations documentation with worktree path, pre/merge/abort/post-clean proof when simulation commands appear in the session log. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -3215,3 +3215,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
|
||||
"ref_mutating_commands": ref_commands,
|
||||
"fetch_targets": fetch_targets,
|
||||
}
|
||||
|
||||
|
||||
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)
|
||||
|
||||
@@ -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"
|
||||
),
|
||||
}
|
||||
@@ -121,4 +121,10 @@ def test_start_issue_template_references_work_issue_workflow():
|
||||
def test_create_issue_final_report_verifier_exported():
|
||||
from review_proofs import assess_create_issue_final_report
|
||||
|
||||
assert callable(assess_create_issue_final_report)
|
||||
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)
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user