review_proofs.py

@@ -4579,6 +4579,13 @@ def assess_validation_worktree_edit_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_mutation_categories_report(report_text, **kwargs):
+    """#319: require precise mutation categories in reviewer controller handoffs."""
+    from reviewer_mutation_categories import assess_mutation_categories_report as _assess
+
+    return _assess(report_text, **kwargs)
+
+
 def assess_worktree_ownership_report(report_text, **kwargs):
     """#312: prove session-owned or safe-reuse worktree before reset/validation."""
     from reviewer_worktree_ownership import assess_worktree_ownership_report as _assess

reviewer_mutation_categories.py

@@ -0,0 +1,130 @@
+"""Precise mutation category verifier for reviewer controller handoffs (#319)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_CONTROLLER_HANDOFF_RE = re.compile(r"controller handoff", re.IGNORECASE)
+_WORKSPACE_MUTATIONS_RE = re.compile(
+    r"^\s*[-*]?\s*workspace\s+mutations\s*:",
+    re.IGNORECASE | re.MULTILINE,
+)
+_VAGUE_MUTATIONS_NONE_RE = re.compile(
+    r"^\s*[-*]?\s*mutations\s*:\s*none\s*$",
+    re.IGNORECASE | re.MULTILINE,
+)
+
+_REQUIRED_CATEGORIES = (
+    "file edits by reviewer",
+    "worktree/index mutations",
+    "git ref mutations",
+)
+_WORKTREE_FIELD_ALIASES = (
+    "worktree/index mutations",
+    "worktree mutations",
+)
+
+
+def _has_category(text: str, category: str) -> bool:
+    pattern = re.compile(
+        rf"^\s*[-*]?\s*{re.escape(category)}\s*:",
+        re.IGNORECASE | re.MULTILINE,
+    )
+    return bool(pattern.search(text))
+
+
+def _has_worktree_category(text: str) -> bool:
+    return any(_has_category(text, alias) for alias in _WORKTREE_FIELD_ALIASES)
+
+
+def _normalize_for_consistency_check(text: str) -> str:
+    """Map #319 precise labels to #313 field names for consistency delegation."""
+    normalized = text
+    if _has_category(text, "worktree/index mutations") and not _has_category(
+        text, "worktree mutations"
+    ):
+        normalized = re.sub(
+            r"(worktree/index mutations\s*:)",
+            "Worktree mutations:",
+            normalized,
+            flags=re.IGNORECASE,
+        )
+    return normalized
+
+
+def assess_mutation_categories_report(
+    report_text: str,
+    *,
+    handoff_session: dict | None = None,
+    observed_commands: list | None = None,
+) -> dict[str, Any]:
+    """Require precise mutation categories in reviewer controller handoffs (#319)."""
+    text = report_text or ""
+    session = dict(handoff_session or {})
+    reasons: list[str] = []
+    lower = text.lower()
+
+    is_controller = bool(
+        session.get("controller_handoff")
+        or _CONTROLLER_HANDOFF_RE.search(text)
+    )
+    if not is_controller and not session.get("require_categories"):
+        return {
+            "proven": True,
+            "block": False,
+            "reasons": [],
+            "controller_handoff": False,
+            "safe_next_action": "proceed",
+        }
+
+    if _WORKSPACE_MUTATIONS_RE.search(text):
+        reasons.append(
+            "controller handoffs must not use legacy 'Workspace mutations'; "
+            "use precise mutation category fields"
+        )
+
+    if _VAGUE_MUTATIONS_NONE_RE.search(text):
+        reasons.append(
+            "controller handoffs must not use vague 'Mutations: none'; "
+            "report each precise category separately"
+        )
+
+    missing = [
+        category
+        for category in _REQUIRED_CATEGORIES
+        if category != "worktree/index mutations" and not _has_category(text, category)
+    ]
+    if not _has_worktree_category(text):
+        missing.append("worktree/index mutations")
+
+    if missing:
+        reasons.append(
+            "controller handoff missing precise mutation categories: "
+            + ", ".join(missing)
+        )
+
+    commands = observed_commands or session.get("observed_commands")
+    if commands:
+        from review_proofs import assess_workspace_mutation_consistency
+
+        consistency = assess_workspace_mutation_consistency(
+            _normalize_for_consistency_check(text),
+            commands,
+        )
+        if not consistency.get("complete"):
+            reasons.extend(consistency.get("reasons") or [])
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "controller_handoff": True,
+        "safe_next_action": (
+            "replace Workspace mutations with file edits, worktree/index, git ref, "
+            "and other precise mutation category fields"
+            if reasons
+            else "proceed"
+        ),
+    }

tests/test_llm_workflow_split.py

@@ -160,6 +160,12 @@ def test_validation_worktree_edit_verifier_exported():
     assert callable(assess_validation_worktree_edit_report)
 
 
+def test_mutation_categories_verifier_exported():
+    from review_proofs import assess_mutation_categories_report
+
+    assert callable(assess_mutation_categories_report)
+
+
 def test_worktree_ownership_verifier_exported():
     from review_proofs import assess_worktree_ownership_report
 

tests/test_reviewer_mutation_categories.py

@@ -0,0 +1,88 @@
+"""Tests for precise mutation category verifier (#319)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_mutation_categories import assess_mutation_categories_report  # noqa: E402
+
+
+def _precise_handoff() -> str:
+    return "\n".join([
+        "Controller Handoff",
+        "File edits by reviewer: none",
+        "Worktree/index mutations: git worktree add branches/review-pr1",
+        "Git ref mutations: git fetch prgs master",
+        "MCP/Gitea mutations: gitea_view_pr",
+        "Review mutations: gitea_review_pr request_changes",
+        "Read-only diagnostics: git status, git diff",
+    ])
+
+
+class TestMutationCategories(unittest.TestCase):
+    def test_precise_categories_pass(self):
+        result = assess_mutation_categories_report(_precise_handoff())
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_legacy_workspace_mutations_blocks(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "Workspace mutations: none (no local file changes)",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: none",
+            "Git ref mutations: none",
+        ])
+        result = assess_mutation_categories_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("workspace mutations" in r.lower() for r in result["reasons"]))
+
+    def test_workspace_none_with_worktree_command_blocks(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "Workspace mutations: none (no local file changes)",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: none",
+            "Git ref mutations: none",
+        ])
+        result = assess_mutation_categories_report(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD"],
+        )
+        self.assertFalse(result["proven"])
+
+    def test_file_edits_none_with_worktree_mutation_passes(self):
+        report = "\n".join([
+            "Controller Handoff",
+            "File edits by reviewer: none",
+            "Worktree/index mutations: git reset --hard FETCH_HEAD",
+            "Git ref mutations: git fetch prgs",
+        ])
+        result = assess_mutation_categories_report(
+            report,
+            observed_commands=["git reset --hard FETCH_HEAD", "git fetch prgs"],
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_missing_required_categories_blocks(self):
+        report = "Controller Handoff\nFile edits by reviewer: none\n"
+        result = assess_mutation_categories_report(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("missing" in r.lower() for r in result["reasons"]))
+
+    def test_non_controller_handoff_skips(self):
+        result = assess_mutation_categories_report(
+            "Workspace mutations: none\n",
+        )
+        self.assertTrue(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_mutation_categories_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()