Add validation integrity verifier for PR-head vs diagnostic tests (#316)

Separate official PR-head validation from post-edit diagnostic runs in
reviewer reports. Block treating diagnostic passes as official validation
and require contaminated sessions to report recovery-required status.

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

review_proofs.py

@@ -3413,3 +3413,10 @@ def assess_git_ref_mutation_report(report_text, *, command_log=None):
         "ref_mutating_commands": ref_commands,
         "fetch_targets": fetch_targets,
     }
+
+
+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)

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

@@ -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_validation_integrity_verifier_exported():
+    from review_proofs import assess_validation_integrity_report
+
+    assert callable(assess_validation_integrity_report)

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