diff --git a/review_proofs.py b/review_proofs.py index d6acb01..ef74ad4 100644 --- a/review_proofs.py +++ b/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) diff --git a/reviewer_validation_integrity.py b/reviewer_validation_integrity.py new file mode 100644 index 0000000..038c457 --- /dev/null +++ b/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" + ), + } diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index f8d3b04..6f13a84 100644 --- a/tests/test_llm_workflow_split.py +++ b/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) \ No newline at end of file + 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) \ No newline at end of file diff --git a/tests/test_reviewer_validation_integrity.py b/tests/test_reviewer_validation_integrity.py new file mode 100644 index 0000000..5137057 --- /dev/null +++ b/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()