From 71ccbca10f5e135bc3bcafb0d0aedbded4d11168 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 11:21:28 -0400 Subject: [PATCH] feat: require transient validation failure history in reviewer reports (Closes #396) Add assess_validation_failure_history_report so reviewer final reports must document every validation failure observed during a session, not only the final passing result. Wire the verifier into final_report_validator, gitea_validate_review_final_report, and the review-merge workflow template. Co-Authored-By: Claude Opus 4.8 (1M context) --- final_report_validator.py | 37 ++++ gitea_mcp_server.py | 4 +- review_final_report_schema.py | 2 + review_proofs.py | 9 + reviewer_validation_failure_history.py | 198 ++++++++++++++++++ .../workflows/review-merge-pr.md | 31 +++ tests/test_llm_workflow_split.py | 6 + ...est_reviewer_validation_failure_history.py | 147 +++++++++++++ 8 files changed, 433 insertions(+), 1 deletion(-) create mode 100644 reviewer_validation_failure_history.py create mode 100644 tests/test_reviewer_validation_failure_history.py diff --git a/final_report_validator.py b/final_report_validator.py index 1bb9cd2..3fbac0f 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -456,6 +456,40 @@ def _rule_reviewer_git_fetch_readonly( return [] +def _rule_reviewer_validation_failure_history( + report_text: str, + *, + validation_session: dict | None = None, +) -> list[dict[str, str]]: + from reviewer_validation_failure_history import ( + assess_validation_failure_history_report, + ) + + session = validation_session or {} + observed = session.get("observed_failures") or [] + if not observed: + return [] + + result = assess_validation_failure_history_report( + report_text, + validation_session=session, + ) + if result.get("proven"): + return [] + severity = "block" if result.get("violations") else "downgrade" + return [ + validator_finding( + "reviewer.validation_failure_history", + severity, + "Validation failure history", + reason, + result.get("safe_next_action") + or "document every observed validation failure before claiming pass", + ) + for reason in (result.get("violations") or result.get("reasons") or ["incomplete"]) + ] + + def _rule_reviewer_validation_command(report_text: str) -> list[dict[str, str]]: text = report_text or "" if not _BARE_PYTEST_RE.search(text): @@ -864,6 +898,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reviewer_mutation_categories, _rule_reviewer_git_fetch_readonly, _rule_reviewer_validation_command, + _rule_reviewer_validation_failure_history, _rule_reviewer_validation_structured, _rule_reviewer_linked_issue, _rule_reviewer_baseline_on_failure, @@ -970,6 +1005,7 @@ def assess_final_report_validator( local_edits: bool = False, issue_filing_lock: dict | None = None, session_pr_opened: bool = False, + validation_session: dict | None = None, ) -> dict[str, Any]: """Validate final-report text against task-specific proof rules (#327). @@ -1025,6 +1061,7 @@ def assess_final_report_validator( "mutations_observed": mutations_observed, "local_edits": local_edits, "session_pr_opened": session_pr_opened, + "validation_session": validation_session, } for rule in _RULES_BY_TASK.get(normalized_kind, ()): diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 0d4ed0f..8a50e59 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -1244,7 +1244,7 @@ def gitea_create_pr( ) if blocked: return blocked - verify_preflight_purity(remote) + verify_preflight_purity(remote, worktree_path=worktree_path) h, o, r = _resolve(remote, host, org, repo) # ── Issue Lock Validation (Issue #194 / #196) ── @@ -5504,6 +5504,7 @@ def gitea_validate_review_final_report( action_log: list[dict] | None = None, mutations_observed: bool = False, local_edits: bool = False, + validation_session: dict | None = None, ) -> dict: """Read-only: machine-validate reviewer final report schema before output (#391). @@ -5521,6 +5522,7 @@ def gitea_validate_review_final_report( action_log=action_log, mutations_observed=mutations_observed, local_edits=local_edits, + validation_session=validation_session, ) diff --git a/review_final_report_schema.py b/review_final_report_schema.py index 01e2e34..e428015 100644 --- a/review_final_report_schema.py +++ b/review_final_report_schema.py @@ -301,6 +301,7 @@ def assess_review_final_report_schema( action_log: list[dict] | None = None, mutations_observed: bool = False, local_edits: bool = False, + validation_session: dict | None = None, ) -> dict[str, Any]: """Validate reviewer final report text before session completion (#391).""" base = assess_final_report_validator( @@ -312,6 +313,7 @@ def assess_review_final_report_schema( action_log=action_log, mutations_observed=mutations_observed, local_edits=local_edits, + validation_session=validation_session, ) extra: list[dict[str, str]] = [] for rule in _SCHEMA_RULES: diff --git a/review_proofs.py b/review_proofs.py index ac479d9..733bb8b 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -5365,6 +5365,15 @@ def assess_validation_integrity_report(report_text, **kwargs): return _assess(report_text, **kwargs) +def assess_validation_failure_history_report(report_text, **kwargs): + """#396: final reports must account for transient validation failures.""" + from reviewer_validation_failure_history import ( + assess_validation_failure_history_report as _assess, + ) + + return _assess(report_text, **kwargs) + + def assess_already_landed_classification_report(report_text, **kwargs): """#295: already-landed PRs are reconciliation-only, not review eligible.""" from reviewer_already_landed_classification import ( diff --git a/reviewer_validation_failure_history.py b/reviewer_validation_failure_history.py new file mode 100644 index 0000000..80da5c3 --- /dev/null +++ b/reviewer_validation_failure_history.py @@ -0,0 +1,198 @@ +"""Transient validation failure history verifier (#396). + +Reviewer sessions may observe validation failures that later pass on rerun. +Final reports must document every failure observed during the session, not +only the last passing result. +""" + +from __future__ import annotations + +import re +from typing import Any + +_SECTION_RE = re.compile( + r"validation failure history", + re.IGNORECASE, +) +_FAILURE_ENTRY_RE = re.compile( + r"(?:failure\s*(?:#|entry)?\s*\d+|validation failure)\s*:", + re.IGNORECASE, +) +_COMMAND_RE = re.compile( + r"(?:command|validation command)\s*:\s*(.+)", + re.IGNORECASE, +) +_FAILING_TEST_RE = re.compile( + r"(?:failing test|failure|error)\s*:\s*(.+)", + re.IGNORECASE, +) +_CAUSE_RE = re.compile( + r"(?:suspected cause|cause)\s*:\s*(.+)", + re.IGNORECASE, +) +_REPRODUCED_RE = re.compile( + r"(?:reproduced|reproduces)\s*:\s*(yes|no|unknown)", + re.IGNORECASE, +) +_BASELINE_RE = re.compile( + r"(?:on baseline master|baseline master|exists on baseline)\s*:\s*(yes|no|unknown|not checked)", + re.IGNORECASE, +) +_PR_CAUSED_RE = re.compile( + r"(?:pr[- ]caused|pr caused)\s*:\s*(yes|no|unknown|not proven)", + re.IGNORECASE, +) +_TRANSIENT_STATUS_RE = re.compile( + r"(?:passed after transient failure investigation|transient failure investigation)", + re.IGNORECASE, +) +_PLAIN_PASS_RE = re.compile( + r"validation\s*:\s*(?:pass|passed|strong|ok|green)\b", + re.IGNORECASE, +) +_ENV_CLEANUP_RE = re.compile( + r"(?:environmental cleanup|state cleaned|what changed between runs)\s*:", + re.IGNORECASE, +) +_UNKNOWN_CAUSE_RE = re.compile( + r"(?:suspected cause|cause)\s*:\s*(?:unknown|unexplained|not determined)", + re.IGNORECASE, +) + + +def _failure_documented_in_text(text: str, failure: dict[str, Any]) -> bool: + """Return True when *failure* appears documented in free-form report text.""" + command = (failure.get("command") or "").strip() + failing = (failure.get("failing_test") or failure.get("error") or "").strip() + if command and command not in text: + return False + if failing and failing not in text: + return False + return bool(command or failing) + + +def _section_has_structured_fields(text: str) -> bool: + if not _SECTION_RE.search(text): + return False + section_start = _SECTION_RE.search(text).start() + section = text[section_start:] + has_command = bool(_COMMAND_RE.search(section)) + has_failure = bool(_FAILING_TEST_RE.search(section)) + return has_command and has_failure + + +def assess_validation_failure_history_report( + report_text: str, + *, + validation_session: dict | None = None, +) -> dict[str, Any]: + """Require final reports to account for transient validation failures (#396).""" + text = report_text or "" + session = dict(validation_session or {}) + reasons: list[str] = [] + violations: list[str] = [] + + observed = list(session.get("observed_failures") or []) + final_status = (session.get("final_validation_status") or "").strip().lower() + cause_unknown = any( + (f.get("suspected_cause") or "").strip().lower() in { + "unknown", "unexplained", "not determined", "" + } + and not (f.get("pr_caused") or "").strip().lower() in {"yes", "no"} + for f in observed + ) or bool(session.get("cause_unknown")) + + if not observed: + return { + "proven": True, + "block": False, + "reasons": [], + "violations": [], + "observed_failure_count": 0, + "safe_next_action": "proceed", + } + + documented = _section_has_structured_fields(text) + if not documented: + for failure in observed: + if _failure_documented_in_text(text, failure): + documented = True + break + + if not documented: + violations.append( + "session observed validation failure(s) but final report omits " + "Validation failure history" + ) + reasons.append( + "every validation failure observed during the session must appear " + "in a Validation failure history section" + ) + + if documented and not _SECTION_RE.search(text): + reasons.append( + "failure details must appear under an explicit " + "'Validation failure history' heading" + ) + + for idx, failure in enumerate(observed, start=1): + prefix = f"failure #{idx}" + if not _failure_documented_in_text(text, failure) and documented: + reasons.append( + f"{prefix}: command and failing test/error not documented in report" + ) + command = (failure.get("command") or "").strip() + failing = (failure.get("failing_test") or failure.get("error") or "").strip() + if not command: + reasons.append(f"{prefix}: missing command in session failure record") + if not failing: + reasons.append( + f"{prefix}: missing failing test or error in session failure record" + ) + + plain_pass = bool(_PLAIN_PASS_RE.search(text)) + transient_wording = bool(_TRANSIENT_STATUS_RE.search(text)) + final_passed = final_status in { + "passed", + "pass", + "passed_after_transient_failure_investigation", + } + + if (final_passed or plain_pass) and observed: + if cause_unknown and not transient_wording: + violations.append( + "unknown transient failure cause cannot be erased as plain 'passed'" + ) + reasons.append( + "when cause is unknown, use status " + "'passed after transient failure investigation'" + ) + elif not transient_wording and final_status != "passed_after_transient_failure_investigation": + if not _ENV_CLEANUP_RE.search(text): + reasons.append( + "later passing rerun must document what changed between runs " + "or environmental cleanup performed" + ) + + if session.get("environmental_contamination") and not _ENV_CLEANUP_RE.search(text): + reasons.append( + "environmental /tmp state contamination must document cleanup or " + "what changed before the passing rerun" + ) + + proven = not reasons and not violations + return { + "proven": proven, + "block": bool(violations) or not proven, + "reasons": reasons, + "violations": violations, + "observed_failure_count": len(observed), + "documented": documented, + "safe_next_action": ( + "add Validation failure history with command, failing test, cause, " + "reproduction, baseline comparison, and PR-caused evidence; use " + "'passed after transient failure investigation' when appropriate" + if not proven + else "proceed" + ), + } \ No newline at end of file diff --git a/skills/llm-project-workflow/workflows/review-merge-pr.md b/skills/llm-project-workflow/workflows/review-merge-pr.md index 86eb359..5b0e10a 100644 --- a/skills/llm-project-workflow/workflows/review-merge-pr.md +++ b/skills/llm-project-workflow/workflows/review-merge-pr.md @@ -538,6 +538,36 @@ Official validation integrity status must be one of: Do not invent a softer status. +## 21A. Validation failure history rule + +Every validation failure observed during the review session must appear in the +final report, even if a later rerun passes. + +Before claiming `Validation: passed`, list every failed validation command from +the session under `Validation failure history`. + +For each failure, report: + +* command +* failing test or error +* suspected cause +* whether it reproduced +* whether it exists on baseline master +* evidence for whether it is or is not PR-caused + +If a later rerun passes, also report: + +* what changed between runs +* whether environmental state was cleaned (for example `/tmp` lock files) +* why the pass is trustworthy + +If the cause is unknown, do not erase the earlier failure with plain +`Validation: passed`. Use a status such as +`passed after transient failure investigation`. + +`gitea_validate_review_final_report` rejects reports that omit known earlier +validation failures when `validation_session.observed_failures` is supplied. + ## 22. Baseline validation rule Do not run tests in the main checkout. @@ -1096,6 +1126,7 @@ Controller Handoff: * Baseline worktree path: * Files reviewed: * Validation: +* Validation failure history: * Official validation integrity status: * Terminal review mutation: * Review decision: diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 72c5515..267cb23 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -199,6 +199,12 @@ def test_validation_integrity_verifier_exported(): assert callable(assess_validation_integrity_report) +def test_validation_failure_history_verifier_exported(): + from review_proofs import assess_validation_failure_history_report + + assert callable(assess_validation_failure_history_report) + + def test_prior_blocker_skip_verifier_exported(): from review_proofs import assess_prior_blocker_skip_proof diff --git a/tests/test_reviewer_validation_failure_history.py b/tests/test_reviewer_validation_failure_history.py new file mode 100644 index 0000000..957fd52 --- /dev/null +++ b/tests/test_reviewer_validation_failure_history.py @@ -0,0 +1,147 @@ +"""Tests for transient validation failure history verifier (#396).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from final_report_validator import assess_final_report_validator # noqa: E402 +from reviewer_validation_failure_history import ( # noqa: E402 + assess_validation_failure_history_report, +) + + +def _full_history_report() -> str: + return "\n".join([ + "Validation failure history", + "Failure #1:", + "Command: venv/bin/python -m pytest tests/test_worktrees.py -q", + "Failing test: tests/test_worktrees.py::TestWorktreeStart::test_rejects_untraceable_branches", + "Suspected cause: stale /tmp/gitea_issue_lock.json from prior session", + "Reproduced: no", + "On baseline master: no", + "PR-caused: no", + "What changed between runs: removed /tmp/gitea_issue_lock.json", + "Environmental cleanup: lock file removed before rerun", + "Validation: passed after transient failure investigation", + "Final validation command: venv/bin/python -m pytest tests/ -q", + "Final result: 1497 passed, 6 skipped", + ]) + + +class TestValidationFailureHistory(unittest.TestCase): + def test_no_failures_observed_passes(self): + result = assess_validation_failure_history_report( + "Validation: passed", + validation_session={"observed_failures": []}, + ) + self.assertTrue(result["proven"]) + + def test_omitted_failure_blocks(self): + result = assess_validation_failure_history_report( + "Validation: passed\n1497 passed, 6 skipped", + validation_session={ + "observed_failures": [{ + "command": "pytest tests/test_worktrees.py -q", + "failing_test": ( + "tests/test_worktrees.py::TestWorktreeStart::" + "test_rejects_untraceable_branches" + ), + }], + "final_validation_status": "passed", + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_full_history_with_transient_investigation_passes(self): + result = assess_validation_failure_history_report( + _full_history_report(), + validation_session={ + "observed_failures": [{ + "command": ( + "venv/bin/python -m pytest tests/test_worktrees.py -q" + ), + "failing_test": ( + "tests/test_worktrees.py::TestWorktreeStart::" + "test_rejects_untraceable_branches" + ), + "suspected_cause": "stale /tmp/gitea_issue_lock.json", + "reproduced": "no", + "on_baseline_master": "no", + "pr_caused": "no", + }], + "final_validation_status": "passed_after_transient_failure_investigation", + "environmental_contamination": True, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_baseline_equivalent_failure_documented(self): + report = "\n".join([ + "Validation failure history", + "Command: pytest tests/test_example.py -q", + "Failing test: tests/test_example.py::test_flaky", + "Suspected cause: pre-existing master failure", + "On baseline master: yes", + "PR-caused: no", + "What changed between runs: none; failure matches baseline", + "Validation: passed after transient failure investigation", + ]) + result = assess_validation_failure_history_report( + report, + validation_session={ + "observed_failures": [{ + "command": "pytest tests/test_example.py -q", + "failing_test": "tests/test_example.py::test_flaky", + "on_baseline_master": "yes", + "pr_caused": "no", + }], + "final_validation_status": "passed_after_transient_failure_investigation", + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_unknown_cause_plain_pass_blocks(self): + result = assess_validation_failure_history_report( + "Validation: passed", + validation_session={ + "observed_failures": [{ + "command": "pytest tests/ -q", + "failing_test": "tests/test_foo.py::test_bar", + "suspected_cause": "unknown", + }], + "final_validation_status": "passed", + "cause_unknown": True, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("transient" in r.lower() for r in result["reasons"])) + + def test_final_report_validator_rejects_omitted_failure(self): + result = assess_final_report_validator( + "Validation: passed", + "review_pr", + validation_session={ + "observed_failures": [{ + "command": "pytest tests/ -q", + "failing_test": "tests/test_foo.py::test_bar", + }], + }, + ) + self.assertTrue(result["blocked"] or result["downgraded"]) + self.assertTrue( + any( + f.get("rule_id") == "reviewer.validation_failure_history" + for f in result.get("findings") or [] + ) + ) + + def test_exported_from_review_proofs(self): + from review_proofs import assess_validation_failure_history_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file -- 2.43.7