feat: add validation ledger and final-report verifier (Closes #271)
Introduce validation_ledger with command ledger format, full-suite claim checks, and structured final-report verification so sessions cannot overclaim validation or misreport review/merge state. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -0,0 +1,155 @@
|
||||
"""Tests for validation ledger and final-report verifier (#271)."""
|
||||
import sys
|
||||
import unittest
|
||||
from pathlib import Path
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
||||
|
||||
from validation_ledger import ( # noqa: E402
|
||||
assess_full_suite_claim,
|
||||
assess_ledger_entry,
|
||||
new_ledger_entry,
|
||||
verify_final_report,
|
||||
)
|
||||
|
||||
|
||||
class TestLedgerEntry(unittest.TestCase):
|
||||
def test_valid_entry_is_claimable(self):
|
||||
entry = new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/feat-issue-271",
|
||||
exit_code=0,
|
||||
output_summary="994 passed, 6 skipped",
|
||||
)
|
||||
result = assess_ledger_entry(entry)
|
||||
self.assertTrue(result["claimable"])
|
||||
self.assertEqual(result["verdict"], "strong")
|
||||
|
||||
def test_missing_command_output_is_invalid(self):
|
||||
entry = new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/task",
|
||||
exit_code=0,
|
||||
output_summary="",
|
||||
)
|
||||
result = assess_ledger_entry(entry)
|
||||
self.assertFalse(result["claimable"])
|
||||
self.assertEqual(result["verdict"], "invalid")
|
||||
|
||||
|
||||
class TestFullSuiteClaim(unittest.TestCase):
|
||||
def _entry(self, **kwargs):
|
||||
base = new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/task",
|
||||
exit_code=0,
|
||||
output_summary="10 passed",
|
||||
)
|
||||
base.update(kwargs)
|
||||
return base
|
||||
|
||||
def test_full_suite_overclaim_with_ignored_paths_fails(self):
|
||||
entry = self._entry(
|
||||
ignored_paths=[{"path": "tests/integration/", "justification": ""}],
|
||||
)
|
||||
result = assess_full_suite_claim(
|
||||
[entry],
|
||||
report_text="Validation: full suite passed.",
|
||||
)
|
||||
self.assertFalse(result["claimable"])
|
||||
self.assertTrue(
|
||||
any("full suite" in r.lower() for r in result["reasons"])
|
||||
)
|
||||
|
||||
def test_full_suite_except_note_allows_ignored_paths(self):
|
||||
entry = self._entry(
|
||||
ignored_paths=[{
|
||||
"path": "tests/integration/",
|
||||
"justification": "network tests require GITEA_INTEGRATION=1",
|
||||
}],
|
||||
)
|
||||
result = assess_full_suite_claim(
|
||||
[entry],
|
||||
report_text="Validation: full suite except integration tests.",
|
||||
)
|
||||
self.assertTrue(result["claimable"])
|
||||
|
||||
|
||||
class TestFinalReportVerifier(unittest.TestCase):
|
||||
def _base_report(self, **overrides):
|
||||
report = {
|
||||
"identity": "jcwalker3",
|
||||
"profile": "prgs-author",
|
||||
"capability_resolved": True,
|
||||
"worktree_path": "/repo/branches/feat-issue-271",
|
||||
"changed_files": ["validation_ledger.py"],
|
||||
"validation_ledger": [
|
||||
new_ledger_entry(
|
||||
command="pytest tests/test_validation_ledger.py -q",
|
||||
cwd="/repo/branches/feat-issue-271",
|
||||
exit_code=0,
|
||||
output_summary="12 passed",
|
||||
)
|
||||
],
|
||||
"validation_claim": "passed",
|
||||
"review_submitted": False,
|
||||
"review_visible_approved": False,
|
||||
"review_pending": False,
|
||||
"merge_performed": False,
|
||||
"merge_blocker": "not a reviewer task",
|
||||
"workspace_clean": True,
|
||||
}
|
||||
report.update(overrides)
|
||||
return report
|
||||
|
||||
def test_complete_report_is_proven(self):
|
||||
result = verify_final_report(self._base_report())
|
||||
self.assertTrue(result["proven"])
|
||||
self.assertFalse(result["block"])
|
||||
|
||||
def test_validation_overclaim_fails(self):
|
||||
report = self._base_report(
|
||||
validation_claim="passed",
|
||||
validation_ledger=[
|
||||
new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/task",
|
||||
exit_code=1,
|
||||
output_summary="3 failed",
|
||||
)
|
||||
],
|
||||
)
|
||||
result = verify_final_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["contradictions"])
|
||||
|
||||
def test_pending_approval_misreported_as_approved_fails(self):
|
||||
report = self._base_report(
|
||||
claims_approved_review=True,
|
||||
review_visible_approved=False,
|
||||
review_pending=True,
|
||||
)
|
||||
result = verify_final_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(
|
||||
any("approved" in c.lower() for c in result["contradictions"])
|
||||
)
|
||||
|
||||
def test_missing_command_output_in_ledger_fails(self):
|
||||
report = self._base_report(
|
||||
validation_ledger=[
|
||||
new_ledger_entry(
|
||||
command="pytest tests/ -q",
|
||||
cwd="/repo/branches/task",
|
||||
exit_code=0,
|
||||
output_summary="",
|
||||
)
|
||||
],
|
||||
)
|
||||
result = verify_final_report(report)
|
||||
self.assertFalse(result["proven"])
|
||||
self.assertTrue(result["reasons"])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
@@ -0,0 +1,239 @@
|
||||
"""Validation command ledger and final-report verifier (#271).
|
||||
|
||||
Sessions may only claim validation passed when the exact command, exit code,
|
||||
and output summary are recorded. Final reports are checked against the
|
||||
ledger and workflow proof fields before submission.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from datetime import datetime, timezone
|
||||
from typing import Any
|
||||
|
||||
LEDGER_REQUIRED_FIELDS = (
|
||||
"command",
|
||||
"cwd",
|
||||
"exit_code",
|
||||
"output_summary",
|
||||
"timestamp",
|
||||
)
|
||||
|
||||
OPTIONAL_LEDGER_FIELDS = (
|
||||
"skipped_tests",
|
||||
"ignored_paths",
|
||||
)
|
||||
|
||||
|
||||
def _clean(value: Any) -> str:
|
||||
return (value or "").strip() if isinstance(value, str) else str(value or "").strip()
|
||||
|
||||
|
||||
def normalize_ledger_entry(entry: dict | None) -> dict:
|
||||
"""Return a normalized ledger entry with required keys present."""
|
||||
data = dict(entry or {})
|
||||
normalized = {
|
||||
"command": _clean(data.get("command")),
|
||||
"cwd": _clean(data.get("cwd")),
|
||||
"exit_code": data.get("exit_code"),
|
||||
"output_summary": _clean(data.get("output_summary")),
|
||||
"timestamp": _clean(data.get("timestamp")),
|
||||
"skipped_tests": data.get("skipped_tests") or [],
|
||||
"ignored_paths": data.get("ignored_paths") or [],
|
||||
}
|
||||
return normalized
|
||||
|
||||
|
||||
def assess_ledger_entry(entry: dict | None) -> dict:
|
||||
"""Fail closed when a ledger entry cannot support a validation claim."""
|
||||
normalized = normalize_ledger_entry(entry)
|
||||
reasons: list[str] = []
|
||||
|
||||
for field in LEDGER_REQUIRED_FIELDS:
|
||||
value = normalized.get(field)
|
||||
if field == "exit_code":
|
||||
if value is None or not isinstance(value, int):
|
||||
reasons.append("exit_code missing or not an integer")
|
||||
continue
|
||||
if not _clean(value):
|
||||
reasons.append(f"ledger field '{field}' missing or empty")
|
||||
|
||||
exit_code = normalized.get("exit_code")
|
||||
if isinstance(exit_code, int) and exit_code != 0:
|
||||
reasons.append(f"exit_code {exit_code} indicates failure")
|
||||
|
||||
if not _clean(normalized.get("output_summary")):
|
||||
reasons.append("output_summary missing; command output was not summarized")
|
||||
|
||||
claimable = not reasons and isinstance(exit_code, int) and exit_code == 0
|
||||
return {
|
||||
"claimable": claimable,
|
||||
"entry": normalized,
|
||||
"reasons": reasons,
|
||||
"verdict": "strong" if claimable else "invalid",
|
||||
}
|
||||
|
||||
|
||||
def assess_full_suite_claim(
|
||||
ledger_entries: list[dict] | None,
|
||||
*,
|
||||
report_text: str = "",
|
||||
) -> dict:
|
||||
"""'Full suite passed' requires zero ignored tests unless explicitly noted."""
|
||||
entries = [normalize_ledger_entry(e) for e in (ledger_entries or [])]
|
||||
reasons: list[str] = []
|
||||
lower = (report_text or "").lower()
|
||||
|
||||
if not entries:
|
||||
return {
|
||||
"claimable": False,
|
||||
"verdict": "invalid",
|
||||
"reasons": ["no validation ledger entries recorded"],
|
||||
}
|
||||
|
||||
entry_assessments = [assess_ledger_entry(e) for e in entries]
|
||||
invalid = [a for a in entry_assessments if not a["claimable"]]
|
||||
if invalid:
|
||||
reasons.extend(invalid[0]["reasons"])
|
||||
|
||||
ignored_total = 0
|
||||
for entry in entries:
|
||||
ignored = entry.get("ignored_paths") or []
|
||||
ignored_total += len(ignored)
|
||||
for item in ignored:
|
||||
path = (item or {}).get("path") if isinstance(item, dict) else item
|
||||
justification = (item or {}).get("justification", "") if isinstance(item, dict) else ""
|
||||
if path and not _clean(justification):
|
||||
reasons.append(
|
||||
f"ignored path '{path}' lacks justification for full-suite claim"
|
||||
)
|
||||
|
||||
claims_full_suite = "full suite passed" in lower or "full test suite passed" in lower
|
||||
if claims_full_suite and ignored_total:
|
||||
except_phrases = (
|
||||
"full suite except",
|
||||
"except ignored",
|
||||
"ignored paths:",
|
||||
"ignored test",
|
||||
)
|
||||
if not any(phrase in lower for phrase in except_phrases):
|
||||
reasons.append(
|
||||
"report claims full suite passed but ledger records ignored "
|
||||
"tests/paths without an explicit 'full suite except …' note"
|
||||
)
|
||||
|
||||
claimable = not reasons and not invalid
|
||||
return {
|
||||
"claimable": claimable,
|
||||
"verdict": "strong" if claimable else ("invalid" if invalid else "weak"),
|
||||
"reasons": reasons,
|
||||
"entry_assessments": entry_assessments,
|
||||
"ignored_path_count": ignored_total,
|
||||
}
|
||||
|
||||
|
||||
def new_ledger_entry(
|
||||
*,
|
||||
command: str,
|
||||
cwd: str,
|
||||
exit_code: int,
|
||||
output_summary: str,
|
||||
skipped_tests: list[str] | None = None,
|
||||
ignored_paths: list[dict] | None = None,
|
||||
timestamp: str | None = None,
|
||||
) -> dict:
|
||||
"""Build a ledger entry with an ISO-8601 UTC timestamp."""
|
||||
ts = timestamp or datetime.now(timezone.utc).replace(microsecond=0).isoformat()
|
||||
return normalize_ledger_entry({
|
||||
"command": command,
|
||||
"cwd": cwd,
|
||||
"exit_code": exit_code,
|
||||
"output_summary": output_summary,
|
||||
"timestamp": ts,
|
||||
"skipped_tests": skipped_tests or [],
|
||||
"ignored_paths": ignored_paths or [],
|
||||
})
|
||||
|
||||
|
||||
def verify_final_report(report: dict | None) -> dict:
|
||||
"""Verify a structured final report against ledger and workflow proofs."""
|
||||
data = dict(report or {})
|
||||
reasons: list[str] = []
|
||||
contradictions: list[str] = []
|
||||
|
||||
identity = _clean(data.get("identity"))
|
||||
profile = _clean(data.get("profile"))
|
||||
capability_resolved = data.get("capability_resolved")
|
||||
worktree_path = _clean(data.get("worktree_path"))
|
||||
changed_files = data.get("changed_files") or []
|
||||
ledger_entries = data.get("validation_ledger") or data.get("ledger_entries") or []
|
||||
|
||||
if not identity:
|
||||
reasons.append("identity missing from final report")
|
||||
if not profile:
|
||||
reasons.append("profile missing from final report")
|
||||
if capability_resolved is not True:
|
||||
reasons.append("capability_resolved must be true with exact resolver evidence")
|
||||
if not worktree_path:
|
||||
reasons.append("worktree_path missing from final report")
|
||||
if not isinstance(changed_files, list):
|
||||
reasons.append("changed_files must be a list (use [] when none)")
|
||||
|
||||
ledger_verdict = assess_full_suite_claim(
|
||||
ledger_entries,
|
||||
report_text=_clean(data.get("report_text")),
|
||||
)
|
||||
if not ledger_verdict["claimable"]:
|
||||
reasons.extend(ledger_verdict["reasons"])
|
||||
|
||||
validation_claim = _clean(data.get("validation_claim")).lower()
|
||||
if validation_claim in ("passed", "pass", "full suite passed"):
|
||||
if ledger_verdict["verdict"] == "invalid":
|
||||
contradictions.append(
|
||||
"validation_claim says passed but ledger entries are invalid"
|
||||
)
|
||||
|
||||
review_submitted = data.get("review_submitted")
|
||||
review_visible_approved = data.get("review_visible_approved")
|
||||
review_pending = data.get("review_pending")
|
||||
if review_visible_approved is True and review_pending is True:
|
||||
contradictions.append(
|
||||
"review_visible_approved and review_pending cannot both be true"
|
||||
)
|
||||
if data.get("claims_approved_review") is True:
|
||||
if review_visible_approved is not True:
|
||||
contradictions.append(
|
||||
"report claims APPROVED review but review_visible_approved is not true"
|
||||
)
|
||||
if review_pending is True:
|
||||
contradictions.append(
|
||||
"report claims APPROVED review while review_pending is true"
|
||||
)
|
||||
|
||||
merge_performed = data.get("merge_performed")
|
||||
merge_blocker = _clean(data.get("merge_blocker"))
|
||||
if merge_performed is True and merge_blocker:
|
||||
contradictions.append("merge_performed true but merge_blocker also set")
|
||||
if merge_performed is not True and not merge_blocker and data.get("claims_merged") is True:
|
||||
contradictions.append("report claims merge but merge_performed is not true")
|
||||
|
||||
workspace_clean = data.get("workspace_clean")
|
||||
workspace_classification = _clean(data.get("workspace_classification"))
|
||||
if workspace_clean is False and not workspace_classification:
|
||||
reasons.append(
|
||||
"workspace is not clean but workspace_classification is missing"
|
||||
)
|
||||
|
||||
if review_submitted is False and data.get("claims_review_submitted") is True:
|
||||
contradictions.append(
|
||||
"claims_review_submitted true but review_submitted is false"
|
||||
)
|
||||
|
||||
all_reasons = reasons + contradictions
|
||||
proven = not all_reasons
|
||||
return {
|
||||
"proven": proven,
|
||||
"block": not proven,
|
||||
"reasons": all_reasons,
|
||||
"contradictions": contradictions,
|
||||
"ledger_verdict": ledger_verdict,
|
||||
}
|
||||
Reference in New Issue
Block a user