Gitea-Tools/validation_ledger.py

"""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,
    }