fix: resolve conflicts for PR #369 against latest master

review_proofs.py

@@ -4736,6 +4736,13 @@ def assess_merge_simulation_report(report_text, **kwargs):
     return _assess(report_text, **kwargs)
 
 
+def assess_validation_integrity_report(report_text, **kwargs):
+    """#316: separate official PR-head validation from diagnostic experiments."""
+    from reviewer_validation_integrity import assess_validation_integrity_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
@@ -4766,6 +4773,13 @@ def assess_inventory_worktree_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"
+        ),
+    }

reviewer_validation_integrity.py

@@ -0,0 +1,148 @@
+"""PR-head vs diagnostic validation integrity verifier (#316)."""
+
+from __future__ import annotations
+
+import re
+from typing import Any
+
+_DIAGNOSTIC_LABEL_RE = re.compile(
+    r"diagnostic local experiment|not pr-head validation|diagnostic-only validation",
+    re.IGNORECASE,
+)
+_OFFICIAL_VALIDATION_RE = re.compile(
+    r"(?:official validation|pr-head validation|validation integrity)",
+    re.IGNORECASE,
+)
+_OFFICIAL_COMMAND_RE = re.compile(
+    r"(?:official validation command|pr-head validation command|validation command)",
+    re.IGNORECASE,
+)
+_OFFICIAL_RESULT_RE = re.compile(
+    r"(?:official validation result|pr-head validation result|validation result|validation integrity status)",
+    re.IGNORECASE,
+)
+_DIRTY_AFTER_RE = re.compile(
+    r"(?:worktree dirty after (?:official )?validation|dirty (?:state )?after (?:official )?validation|"
+    r"validation worktree dirty after)",
+    re.IGNORECASE,
+)
+_DIAGNOSTIC_EDIT_RE = re.compile(
+    r"(?:diagnostic edit(?:ed)? path|file edits by reviewer|diagnostic local edit)",
+    re.IGNORECASE,
+)
+_DIAGNOSTIC_COMMAND_RE = re.compile(
+    r"(?:diagnostic (?:validation )?command|diagnostic test run|diagnostic result)",
+    re.IGNORECASE,
+)
+_SUGGESTED_FIX_RE = re.compile(
+    r"(?:diagnostic results? (?:were )?used only for suggested fix|suggested fix only|"
+    r"not used as official validation)",
+    re.IGNORECASE,
+)
+_INTEGRITY_STATUS_RE = re.compile(
+    r"validation integrity status\s*:\s*(passed|failed|not run|contaminated)",
+    re.IGNORECASE,
+)
+_POST_EDIT_AS_OFFICIAL_RE = re.compile(
+    r"(?:official validation(?:\s+result)?\s*:\s*pass|validation passed after (?:local )?edit|"
+    r"full suite passed after diagnostic edit)",
+    re.IGNORECASE,
+)
+
+
+def _performed_edits(action_log: list[dict] | None) -> list[str]:
+    paths: list[str] = []
+    for entry in action_log or []:
+        if entry.get("gated_rejected") or entry.get("performed") is False:
+            continue
+        path = entry.get("path")
+        if path and entry.get("kind", "file_edit") in {"file_edit", "edit", "write"}:
+            paths.append(str(path))
+    return paths
+
+
+def assess_validation_integrity_report(
+    report_text: str,
+    *,
+    validation_session: dict | None = None,
+    action_log: list[dict] | None = None,
+) -> dict[str, Any]:
+    """Separate official PR-head validation from diagnostic edited-worktree tests (#316)."""
+    text = report_text or ""
+    session = dict(validation_session or {})
+    reasons: list[str] = []
+
+    diagnostic_edits = list(session.get("diagnostic_edits") or [])
+    if not diagnostic_edits:
+        diagnostic_edits = _performed_edits(action_log)
+    diagnostic_ran = bool(
+        session.get("diagnostic_validation_ran")
+        or session.get("diagnostic_runs")
+    )
+    official_ran = bool(session.get("official_validation_ran", True))
+    official_result = (session.get("official_result") or "").strip().lower()
+    diagnostic_result = (session.get("diagnostic_result") or "").strip().lower()
+    dirty_after = session.get("worktree_dirty_after_official")
+    contaminated = bool(session.get("contaminated"))
+
+    if official_ran:
+        if not _OFFICIAL_VALIDATION_RE.search(text) and not _OFFICIAL_COMMAND_RE.search(text):
+            reasons.append("report missing official PR-head validation section")
+        if not _OFFICIAL_COMMAND_RE.search(text) and "validation:" not in text.lower():
+            reasons.append("report missing official validation command")
+        if not _OFFICIAL_RESULT_RE.search(text):
+            reasons.append("report missing official validation result or integrity status")
+        if dirty_after is not None and not _DIRTY_AFTER_RE.search(text):
+            reasons.append(
+                "report missing worktree dirty state after official validation"
+            )
+        elif dirty_after is None and official_ran and not _DIRTY_AFTER_RE.search(text):
+            reasons.append(
+                "report missing worktree dirty state after official validation"
+            )
+
+    if diagnostic_edits or diagnostic_ran:
+        if not _DIAGNOSTIC_LABEL_RE.search(text):
+            reasons.append(
+                "post-edit diagnostic runs must be labeled "
+                "'Diagnostic local experiment — not PR-head validation'"
+            )
+        if diagnostic_edits and not _DIAGNOSTIC_EDIT_RE.search(text):
+            reasons.append("report missing diagnostic edit path")
+        if diagnostic_ran and not _DIAGNOSTIC_COMMAND_RE.search(text):
+            reasons.append("report missing diagnostic command/result")
+        if not _SUGGESTED_FIX_RE.search(text):
+            reasons.append(
+                "report must state diagnostic results were used only for suggested fix"
+            )
+        if _POST_EDIT_AS_OFFICIAL_RE.search(text):
+            reasons.append(
+                "post-edit diagnostic results cannot be reported as official PR-head validation"
+            )
+        if diagnostic_result == "pass" and official_result == "fail":
+            if "request_changes" not in text.lower() and "failed" not in text.lower():
+                reasons.append(
+                    "request-changes must cite unmodified PR-head failure, not diagnostic pass"
+                )
+
+    if contaminated:
+        status = _INTEGRITY_STATUS_RE.search(text)
+        if not status or "contaminated" not in status.group(1).lower():
+            reasons.append(
+                "contaminated validation must report integrity status "
+                "'contaminated — recovery required'"
+            )
+
+    proven = not reasons
+    return {
+        "proven": proven,
+        "block": not proven,
+        "reasons": reasons,
+        "diagnostic_edits": diagnostic_edits,
+        "safe_next_action": (
+            "separate official PR-head validation from diagnostic experiments; "
+            "report dirty-after-validation state"
+            if reasons
+            else "proceed"
+        ),
+    }

tests/test_llm_workflow_split.py

@@ -142,6 +142,12 @@ def test_merge_simulation_verifier_exported():
     assert callable(assess_merge_simulation_report)
 
 
+def test_validation_integrity_verifier_exported():
+    from review_proofs import assess_validation_integrity_report
+
+    assert callable(assess_validation_integrity_report)
+
+
 def test_prior_blocker_skip_verifier_exported():
     from review_proofs import assess_prior_blocker_skip_proof
 
@@ -166,6 +172,12 @@ def test_inventory_worktree_verifier_exported():
     assert callable(assess_inventory_worktree_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()

tests/test_reviewer_validation_integrity.py

@@ -0,0 +1,130 @@
+"""Tests for PR-head vs diagnostic validation integrity verifier (#316)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+from reviewer_validation_integrity import assess_validation_integrity_report  # noqa: E402
+
+
+def _official_only_report() -> str:
+    return "\n".join([
+        "Official PR-head validation on unmodified worktree.",
+        "Official validation command: pytest tests/ -q",
+        "Official validation result: failed (3 failed)",
+        "Validation integrity status: failed",
+        "Worktree dirty after official validation: clean",
+        "Review decision: request_changes",
+    ])
+
+
+def _diagnostic_report() -> str:
+    return "\n".join([
+        _official_only_report(),
+        "Diagnostic local experiment — not PR-head validation",
+        "Diagnostic edit path: tests/test_example.py",
+        "File edits by reviewer: tests/test_example.py",
+        "Diagnostic command: pytest tests/test_example.py -q",
+        "Diagnostic result: passed after local fix",
+        "Diagnostic results used only for suggested fix; blocker remains PR-head failure.",
+    ])
+
+
+class TestValidationIntegrity(unittest.TestCase):
+    def test_official_only_passes(self):
+        result = assess_validation_integrity_report(
+            _official_only_report(),
+            validation_session={
+                "official_validation_ran": True,
+                "official_result": "fail",
+                "worktree_dirty_after_official": False,
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_failing_pr_head_with_diagnostic_pass_passes(self):
+        result = assess_validation_integrity_report(
+            _diagnostic_report(),
+            validation_session={
+                "official_validation_ran": True,
+                "official_result": "fail",
+                "diagnostic_validation_ran": True,
+                "diagnostic_result": "pass",
+                "diagnostic_edits": ["tests/test_example.py"],
+                "worktree_dirty_after_official": False,
+            },
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_post_edit_tests_as_official_blocks(self):
+        report = "\n".join([
+            "Official validation result: passed after diagnostic edit",
+            "pytest passed after local fix in validation worktree",
+        ])
+        result = assess_validation_integrity_report(
+            report,
+            validation_session={
+                "official_validation_ran": True,
+                "diagnostic_edits": ["tests/test_example.py"],
+                "diagnostic_validation_ran": True,
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(result["block"])
+
+    def test_missing_dirty_after_validation_blocks(self):
+        report = _official_only_report().replace(
+            "Worktree dirty after official validation: clean",
+            "",
+        )
+        result = assess_validation_integrity_report(
+            report,
+            validation_session={"official_validation_ran": True, "official_result": "fail"},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("dirty" in r.lower() for r in result["reasons"]))
+
+    def test_diagnostic_without_label_blocks(self):
+        report = _diagnostic_report().replace(
+            "Diagnostic local experiment — not PR-head validation",
+            "",
+        )
+        result = assess_validation_integrity_report(
+            report,
+            validation_session={
+                "official_validation_ran": True,
+                "diagnostic_edits": ["tests/test_example.py"],
+                "diagnostic_validation_ran": True,
+            },
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("diagnostic" in r.lower() for r in result["reasons"]))
+
+    def test_contaminated_requires_integrity_status(self):
+        report = "Validation worktree was edited before official validation completed."
+        result = assess_validation_integrity_report(
+            report,
+            validation_session={"contaminated": True, "official_validation_ran": True},
+        )
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("contaminated" in r.lower() for r in result["reasons"]))
+
+    def test_action_log_edits_trigger_diagnostic_rules(self):
+        result = assess_validation_integrity_report(
+            _official_only_report(),
+            validation_session={"official_validation_ran": True, "official_result": "fail"},
+            action_log=[{"path": "tests/test_example.py", "kind": "file_edit", "performed": True}],
+        )
+        self.assertFalse(result["proven"])
+
+
+class TestExport(unittest.TestCase):
+    def test_review_proofs_reexport(self):
+        from review_proofs import assess_validation_integrity_report as exported
+
+        self.assertTrue(callable(exported))
+
+
+if __name__ == "__main__":
+    unittest.main()