feat: add precise validation status vocabulary for reviewer reports (Closes #406)
Introduce validation_status_vocabulary with proof-path binding for baseline- equivalent, merge-simulation-resolved, and transient-pass labels. Wire the assessor into final_report_validator and document the taxonomy in the review-merge workflow. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
@@ -20,6 +20,7 @@ from review_proofs import (
|
||||
assess_review_mutation_final_report,
|
||||
assess_validation_report,
|
||||
)
|
||||
from validation_status_vocabulary import assess_validation_status_vocabulary
|
||||
|
||||
FINAL_REPORT_TASK_KINDS = frozenset({
|
||||
"review_pr",
|
||||
@@ -598,6 +599,34 @@ def _rule_reviewer_main_checkout_baseline(report_text: str) -> list[dict[str, st
|
||||
]
|
||||
|
||||
|
||||
def _rule_reviewer_validation_status_vocabulary(
|
||||
report_text: str,
|
||||
*,
|
||||
action_log: list[dict] | None = None,
|
||||
) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if not re.search(
|
||||
r"validation status|pr-head validation status|official validation status",
|
||||
text,
|
||||
re.IGNORECASE,
|
||||
):
|
||||
return []
|
||||
result = assess_validation_status_vocabulary(
|
||||
text,
|
||||
command_log=action_log,
|
||||
)
|
||||
if not result.get("block"):
|
||||
return []
|
||||
return _findings_from_reasons(
|
||||
"reviewer.validation_status_vocabulary",
|
||||
result.get("reasons") or [],
|
||||
field="Validation status",
|
||||
severity="block",
|
||||
safe_next_action=result.get("safe_next_action")
|
||||
or "use a validation status that matches the proof path executed",
|
||||
)
|
||||
|
||||
|
||||
def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]:
|
||||
text = report_text or ""
|
||||
if "baseline worktree path" not in text.lower():
|
||||
@@ -902,6 +931,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
|
||||
_rule_reviewer_validation_structured,
|
||||
_rule_reviewer_linked_issue,
|
||||
_rule_reviewer_baseline_on_failure,
|
||||
_rule_reviewer_validation_status_vocabulary,
|
||||
_rule_reviewer_main_checkout_baseline,
|
||||
_rule_reviewer_main_checkout_path,
|
||||
_rule_reviewer_already_landed_eligible,
|
||||
|
||||
@@ -568,6 +568,28 @@ If the cause is unknown, do not erase the earlier failure with plain
|
||||
`gitea_validate_review_final_report` rejects reports that omit known earlier
|
||||
validation failures when `validation_session.observed_failures` is supplied.
|
||||
|
||||
## 21B. Validation status taxonomy (#406)
|
||||
|
||||
When the final report summarizes how validation concluded, use one of these
|
||||
**validation status** labels (distinct from per-command pass/fail entries):
|
||||
|
||||
* `passed` — raw PR-head validation passed on the unmodified head.
|
||||
* `failed` — raw PR-head validation failed and no allowed resolution path
|
||||
was proven.
|
||||
* `baseline-equivalent failure accepted` — only when a clean baseline
|
||||
worktree under `branches/` proves matching failure signatures on the target
|
||||
branch (baseline path, target SHA, exact commands, failure lists, and
|
||||
`failure signatures match: true`).
|
||||
* `raw-head failure resolved by merge simulation` — raw PR-head validation
|
||||
failed, but merge simulation into the current target passed cleanly; report
|
||||
merge simulation under `Worktree/index mutations` with full #317 proof.
|
||||
* `passed after transient failure investigation` — a later run passed after an
|
||||
earlier failure in the same session; document the failure history (#396).
|
||||
|
||||
Do not use `baseline-equivalent failure accepted` when only merge simulation
|
||||
resolved the failure. Do not use bare `passed` when raw PR-head validation
|
||||
failed unless one of the resolution statuses above applies.
|
||||
|
||||
## 22. Baseline validation rule
|
||||
|
||||
Do not run tests in the main checkout.
|
||||
|
||||
@@ -0,0 +1,198 @@
|
||||
"""Tests for validation status vocabulary (#406)."""
|
||||
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 validation_status_vocabulary import ( # noqa: E402
|
||||
STATUS_BASELINE_EQUIVALENT,
|
||||
STATUS_FAILED,
|
||||
STATUS_MERGE_SIM_RESOLVED,
|
||||
STATUS_PASSED,
|
||||
STATUS_TRANSIENT_PASS,
|
||||
assess_validation_status_vocabulary,
|
||||
)
|
||||
|
||||
|
||||
def _handoff(**extra):
|
||||
fields = {
|
||||
"Task": "review PR #386",
|
||||
"Validation status": STATUS_PASSED,
|
||||
"Raw PR-head validation result": "passed",
|
||||
"Merge simulation result": "not run",
|
||||
"Baseline worktree used": "none",
|
||||
}
|
||||
fields.update(extra)
|
||||
lines = ["## Controller Handoff", ""]
|
||||
lines.extend(f"- {key}: {value}" for key, value in fields.items())
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
class TestValidationStatusVocabulary(unittest.TestCase):
|
||||
def test_raw_head_pass_status(self):
|
||||
report = _handoff()
|
||||
result = assess_validation_status_vocabulary(report)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertEqual(result["status_claimed"], STATUS_PASSED)
|
||||
|
||||
def test_raw_head_failure_with_baseline_match(self):
|
||||
report = _handoff(
|
||||
**{
|
||||
"Validation status": STATUS_BASELINE_EQUIVALENT,
|
||||
"Raw PR-head validation result": "failed",
|
||||
"Baseline worktree used": "branches/baseline-master-pr386",
|
||||
"Baseline target SHA": "a" * 40,
|
||||
"Baseline failures": "test_foo failed",
|
||||
"PR failures": "test_foo failed",
|
||||
"Failure signatures match": "true",
|
||||
}
|
||||
)
|
||||
result = assess_validation_status_vocabulary(report)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["baseline_proof_complete"])
|
||||
|
||||
def test_baseline_equivalent_without_baseline_proof_blocked(self):
|
||||
report = _handoff(
|
||||
**{
|
||||
"Validation status": STATUS_BASELINE_EQUIVALENT,
|
||||
"Raw PR-head validation result": "failed",
|
||||
}
|
||||
)
|
||||
result = assess_validation_status_vocabulary(report)
|
||||
self.assertTrue(result["block"])
|
||||
self.assertIn("baseline-equivalent", result["reasons"][0])
|
||||
|
||||
def test_merge_simulation_resolution_passes(self):
|
||||
report = "\n".join([
|
||||
_handoff(
|
||||
**{
|
||||
"Validation status": STATUS_MERGE_SIM_RESOLVED,
|
||||
"Raw PR-head validation result": "failed",
|
||||
"Merge simulation result": "passed",
|
||||
}
|
||||
),
|
||||
"Worktree/index mutations: merge simulation in branches/review-pr386",
|
||||
"Worktree path: branches/review-pr386",
|
||||
"Pre-simulation clean status: clean",
|
||||
"Merge result: clean merge",
|
||||
"Abort command: git merge --abort",
|
||||
"Post-abort clean status: clean",
|
||||
])
|
||||
command_log = [
|
||||
{"command": "git merge --no-commit prgs/master"},
|
||||
{"command": "git merge --abort"},
|
||||
]
|
||||
result = assess_validation_status_vocabulary(
|
||||
report, command_log=command_log
|
||||
)
|
||||
self.assertFalse(result["block"])
|
||||
self.assertTrue(result["merge_simulation_passed"])
|
||||
|
||||
def test_merge_simulation_failure_stays_failed(self):
|
||||
report = _handoff(
|
||||
**{
|
||||
"Validation status": STATUS_FAILED,
|
||||
"Raw PR-head validation result": "failed",
|
||||
"Merge simulation result": "failed",
|
||||
}
|
||||
)
|
||||
result = assess_validation_status_vocabulary(report)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_failed_status_with_passing_merge_sim_blocked(self):
|
||||
report = "\n".join([
|
||||
_handoff(
|
||||
**{
|
||||
"Validation status": STATUS_FAILED,
|
||||
"Raw PR-head validation result": "failed",
|
||||
"Merge simulation result": "passed",
|
||||
}
|
||||
),
|
||||
"Worktree/index mutations: merge simulation",
|
||||
"Worktree path: branches/review-pr386",
|
||||
"Pre-simulation clean status: clean",
|
||||
"Merge result: clean",
|
||||
"Abort command: git merge --abort",
|
||||
"Post-abort clean status: clean",
|
||||
])
|
||||
result = assess_validation_status_vocabulary(
|
||||
report,
|
||||
command_log=[{"command": "git merge --no-commit prgs/master"}],
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_transient_failure_then_pass(self):
|
||||
report = _handoff(
|
||||
**{
|
||||
"Validation status": STATUS_TRANSIENT_PASS,
|
||||
"Raw PR-head validation result": "passed",
|
||||
"Transient validation failure history": (
|
||||
"first run failed with infra flake; rerun passed"
|
||||
),
|
||||
}
|
||||
)
|
||||
result = assess_validation_status_vocabulary(report)
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_transient_pass_without_history_blocked(self):
|
||||
report = _handoff(
|
||||
**{
|
||||
"Validation status": STATUS_TRANSIENT_PASS,
|
||||
"Raw PR-head validation result": "passed",
|
||||
}
|
||||
)
|
||||
result = assess_validation_status_vocabulary(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_bare_passed_after_raw_failure_blocked(self):
|
||||
report = _handoff(
|
||||
**{
|
||||
"Validation status": STATUS_PASSED,
|
||||
"Raw PR-head validation result": "failed",
|
||||
}
|
||||
)
|
||||
result = assess_validation_status_vocabulary(report)
|
||||
self.assertTrue(result["block"])
|
||||
|
||||
def test_wrong_baseline_label_when_merge_sim_used_blocked(self):
|
||||
report = "\n".join([
|
||||
_handoff(
|
||||
**{
|
||||
"Validation status": STATUS_BASELINE_EQUIVALENT,
|
||||
"Raw PR-head validation result": "failed",
|
||||
"Merge simulation result": "passed",
|
||||
}
|
||||
),
|
||||
"Worktree/index mutations: merge simulation",
|
||||
"Worktree path: branches/review-pr386",
|
||||
"Pre-simulation clean status: clean",
|
||||
"Merge result: clean",
|
||||
"Abort command: git merge --abort",
|
||||
"Post-abort clean status: clean",
|
||||
])
|
||||
result = assess_validation_status_vocabulary(
|
||||
report,
|
||||
command_log=[{"command": "git merge --no-commit prgs/master"}],
|
||||
)
|
||||
self.assertTrue(result["block"])
|
||||
joined = " ".join(result["reasons"]).lower()
|
||||
self.assertTrue(
|
||||
"misleading" in joined or "baseline-equivalent" in joined
|
||||
)
|
||||
|
||||
def test_final_report_validator_integration_blocks_misleading_label(self):
|
||||
report = _handoff(
|
||||
**{
|
||||
"Validation status": STATUS_BASELINE_EQUIVALENT,
|
||||
"Raw PR-head validation result": "failed",
|
||||
}
|
||||
)
|
||||
result = assess_final_report_validator(report, "review_pr")
|
||||
blocked_ids = {f["rule_id"] for f in result["findings"]}
|
||||
self.assertIn("reviewer.validation_status_vocabulary", blocked_ids)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,205 @@
|
||||
"""Precise validation-status vocabulary for reviewer final reports (#406)."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from typing import Any
|
||||
|
||||
from reviewer_merge_simulation import assess_merge_simulation_report
|
||||
|
||||
STATUS_PASSED = "passed"
|
||||
STATUS_FAILED = "failed"
|
||||
STATUS_BASELINE_EQUIVALENT = "baseline-equivalent failure accepted"
|
||||
STATUS_MERGE_SIM_RESOLVED = "raw-head failure resolved by merge simulation"
|
||||
STATUS_TRANSIENT_PASS = "passed after transient failure investigation"
|
||||
|
||||
ALLOWED_VALIDATION_STATUSES = frozenset({
|
||||
STATUS_PASSED,
|
||||
STATUS_FAILED,
|
||||
STATUS_BASELINE_EQUIVALENT,
|
||||
STATUS_MERGE_SIM_RESOLVED,
|
||||
STATUS_TRANSIENT_PASS,
|
||||
})
|
||||
|
||||
_STATUS_FIELD_RE = re.compile(
|
||||
r"^\s*[-*]?\s*(?:validation status|pr-head validation status|"
|
||||
r"official validation status)\s*:\s*(.+?)\s*$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_RAW_HEAD_RESULT_RE = re.compile(
|
||||
r"^\s*[-*]?\s*raw pr-head validation result\s*:\s*(.+?)\s*$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_MERGE_SIM_RESULT_RE = re.compile(
|
||||
r"^\s*[-*]?\s*merge simulation result\s*:\s*(.+?)\s*$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_BASELINE_WORKTREE_USED_RE = re.compile(
|
||||
r"^\s*[-*]?\s*baseline (?:validation )?worktree(?: used)?\s*:\s*(.+?)\s*$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_BASELINE_TARGET_SHA_RE = re.compile(
|
||||
r"^\s*[-*]?\s*baseline target sha\s*:\s*([0-9a-f]{7,40})\s*$",
|
||||
re.IGNORECASE | re.MULTILINE,
|
||||
)
|
||||
_FAILURE_SIGNATURE_RE = re.compile(
|
||||
r"failure signatures match\s*:\s*(true|yes)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_BASELINE_FAILURES_RE = re.compile(
|
||||
r"baseline failures\s*:",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_TRANSIENT_HISTORY_RE = re.compile(
|
||||
r"(?:transient validation failure|earlier validation failure|"
|
||||
r"prior failure|failure history|failed then passed)",
|
||||
re.IGNORECASE,
|
||||
)
|
||||
_FULL_SHA = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
|
||||
|
||||
|
||||
def _first_match(pattern: re.Pattern[str], text: str) -> str:
|
||||
match = pattern.search(text or "")
|
||||
return (match.group(1).strip() if match else "")
|
||||
|
||||
|
||||
def _normalize_status_label(raw: str) -> str:
|
||||
text = (raw or "").strip().lower()
|
||||
for status in ALLOWED_VALIDATION_STATUSES:
|
||||
if text == status.lower():
|
||||
return status
|
||||
return raw.strip()
|
||||
|
||||
|
||||
def _baseline_proof_complete(text: str, baseline_proof: dict | None) -> bool:
|
||||
proof = baseline_proof or {}
|
||||
worktree = (
|
||||
(proof.get("worktree_path") or "").strip()
|
||||
or _first_match(_BASELINE_WORKTREE_USED_RE, text)
|
||||
).lower()
|
||||
if not worktree or worktree in {"none", "n/a", "not used", "not applicable"}:
|
||||
return False
|
||||
if "branches/" not in worktree and not worktree.startswith("branches/"):
|
||||
return False
|
||||
target_sha = (proof.get("baseline_target_sha") or "").strip()
|
||||
if not target_sha:
|
||||
target_sha = _first_match(_BASELINE_TARGET_SHA_RE, text)
|
||||
if not _FULL_SHA.match(target_sha or ""):
|
||||
return False
|
||||
if proof.get("failure_signatures_match") is True:
|
||||
return True
|
||||
if _FAILURE_SIGNATURE_RE.search(text) and _BASELINE_FAILURES_RE.search(text):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _merge_simulation_passed(text: str, command_log: list | None) -> bool:
|
||||
merge_result = _first_match(_MERGE_SIM_RESULT_RE, text).lower()
|
||||
if merge_result in {"passed", "pass", "clean", "succeeded", "success"}:
|
||||
sim = assess_merge_simulation_report(text, command_log=command_log)
|
||||
return sim.get("proven") and not sim.get("block")
|
||||
if "pass" in merge_result and "fail" not in merge_result:
|
||||
sim = assess_merge_simulation_report(text, command_log=command_log)
|
||||
return sim.get("proven") and not sim.get("block")
|
||||
return False
|
||||
|
||||
|
||||
def _raw_head_failed(text: str) -> bool:
|
||||
raw = _first_match(_RAW_HEAD_RESULT_RE, text).lower()
|
||||
if raw in {"failed", "fail", "failure"}:
|
||||
return True
|
||||
if "fail" in raw and "pass" not in raw:
|
||||
return True
|
||||
return bool(re.search(r"\bfailed\b.*pr-head validation", text, re.IGNORECASE))
|
||||
|
||||
|
||||
def _raw_head_passed(text: str) -> bool:
|
||||
raw = _first_match(_RAW_HEAD_RESULT_RE, text).lower()
|
||||
return raw in {"passed", "pass", "success"}
|
||||
|
||||
|
||||
def assess_validation_status_vocabulary(
|
||||
report_text: str,
|
||||
*,
|
||||
command_log: list | None = None,
|
||||
baseline_proof: dict | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""Bind validation-status labels to the proof path that actually ran (#406)."""
|
||||
text = report_text or ""
|
||||
reasons: list[str] = []
|
||||
status_raw = _first_match(_STATUS_FIELD_RE, text)
|
||||
status = _normalize_status_label(status_raw) if status_raw else ""
|
||||
|
||||
if status_raw and status not in ALLOWED_VALIDATION_STATUSES:
|
||||
reasons.append(
|
||||
f"unknown validation status {status_raw!r}; use one of "
|
||||
f"{sorted(ALLOWED_VALIDATION_STATUSES)}"
|
||||
)
|
||||
|
||||
if status == STATUS_BASELINE_EQUIVALENT:
|
||||
if not _baseline_proof_complete(text, baseline_proof):
|
||||
reasons.append(
|
||||
"baseline-equivalent failure accepted requires baseline "
|
||||
"worktree path, baseline target SHA, and matching failure "
|
||||
"signatures (#406)"
|
||||
)
|
||||
|
||||
if status == STATUS_MERGE_SIM_RESOLVED:
|
||||
if not _raw_head_failed(text):
|
||||
reasons.append(
|
||||
"raw-head failure resolved by merge simulation requires "
|
||||
"raw PR-head validation result: failed (#406)"
|
||||
)
|
||||
if not _merge_simulation_passed(text, command_log):
|
||||
reasons.append(
|
||||
"raw-head failure resolved by merge simulation requires "
|
||||
"passing merge simulation with worktree/index mutation proof "
|
||||
"(#317/#406)"
|
||||
)
|
||||
|
||||
if status == STATUS_TRANSIENT_PASS:
|
||||
if not _TRANSIENT_HISTORY_RE.search(text):
|
||||
reasons.append(
|
||||
"passed after transient failure investigation requires "
|
||||
"documented earlier validation failure history (#396/#406)"
|
||||
)
|
||||
|
||||
if status == STATUS_PASSED and _raw_head_failed(text):
|
||||
reasons.append(
|
||||
"validation status passed contradicts raw PR-head validation "
|
||||
"failure; use a precise status (#406)"
|
||||
)
|
||||
|
||||
if status == STATUS_BASELINE_EQUIVALENT and _merge_simulation_passed(
|
||||
text, command_log
|
||||
) and not _baseline_proof_complete(text, baseline_proof):
|
||||
reasons.append(
|
||||
"baseline-equivalent failure accepted is misleading when only "
|
||||
"merge simulation resolved the failure; use "
|
||||
"'raw-head failure resolved by merge simulation' (#406)"
|
||||
)
|
||||
|
||||
if status == STATUS_FAILED and _merge_simulation_passed(text, command_log):
|
||||
reasons.append(
|
||||
"validation status failed contradicts passing merge simulation; "
|
||||
"report the precise resolution status (#406)"
|
||||
)
|
||||
|
||||
block = bool(reasons)
|
||||
return {
|
||||
"block": block,
|
||||
"proven": not block,
|
||||
"status_claimed": status or None,
|
||||
"raw_status_label": status_raw or None,
|
||||
"raw_head_failed": _raw_head_failed(text),
|
||||
"raw_head_passed": _raw_head_passed(text),
|
||||
"merge_simulation_passed": _merge_simulation_passed(text, command_log),
|
||||
"baseline_proof_complete": _baseline_proof_complete(text, baseline_proof),
|
||||
"reasons": reasons,
|
||||
"safe_next_action": (
|
||||
"use a validation status that matches the proof path executed "
|
||||
"(baseline worktree, merge simulation, or transient history)"
|
||||
if reasons
|
||||
else "proceed"
|
||||
),
|
||||
}
|
||||
Reference in New Issue
Block a user