Gitea-Tools/validation_status_vocabulary.py

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