From 4ea618e8b442e6bbd7d5f93c4b790e4241bc5c0d Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:19:22 -0400 Subject: [PATCH 1/2] 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) --- review_proofs.py | 7 + reviewer_validation_integrity.py | 148 ++++++++++++++++++++ tests/test_llm_workflow_split.py | 8 +- tests/test_reviewer_validation_integrity.py | 130 +++++++++++++++++ 4 files changed, 292 insertions(+), 1 deletion(-) create mode 100644 reviewer_validation_integrity.py create mode 100644 tests/test_reviewer_validation_integrity.py 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() From b71d3aeab38b1c05ff74f1c23e0d34f77e2a5ca4 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 05:31:24 -0400 Subject: [PATCH 2/2] Add precise mutation category verifier for controller handoffs (#319) Reject legacy Workspace mutations wording and require explicit file, worktree/index, and git ref mutation fields in reviewer controller handoffs. Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 7 ++ reviewer_mutation_categories.py | 130 +++++++++++++++++++++ tests/test_llm_workflow_split.py | 6 + tests/test_reviewer_mutation_categories.py | 88 ++++++++++++++ 4 files changed, 231 insertions(+) create mode 100644 reviewer_mutation_categories.py create mode 100644 tests/test_reviewer_mutation_categories.py diff --git a/review_proofs.py b/review_proofs.py index e7c94e2..add5ddf 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3839,3 +3839,10 @@ def assess_non_mergeable_skip_proof(report_text, **kwargs): from reviewer_mergeability_skip import assess_non_mergeable_skip_proof as _assess 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) diff --git a/reviewer_mutation_categories.py b/reviewer_mutation_categories.py new file mode 100644 index 0000000..a09e73b --- /dev/null +++ b/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" + ), + } \ No newline at end of file diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 9a96d17..ba4df08 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -134,3 +134,9 @@ def test_non_mergeable_skip_verifier_exported(): from review_proofs import assess_non_mergeable_skip_proof assert callable(assess_non_mergeable_skip_proof) + + +def test_mutation_categories_verifier_exported(): + from review_proofs import assess_mutation_categories_report + + assert callable(assess_mutation_categories_report) diff --git a/tests/test_reviewer_mutation_categories.py b/tests/test_reviewer_mutation_categories.py new file mode 100644 index 0000000..055b174 --- /dev/null +++ b/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() \ No newline at end of file