Gitea-Tools/thread_state_ledger_validator.py

"""Two-comment workflow validation for Controller Handoff + Thread State Ledger (#507).

Validates the paired reporting pattern without replacing the CTH umbrella (#505)
or the broader lifecycle ledger (#494/#495).
"""

from __future__ import annotations

import re
from typing import Any

CONTROLLER_HANDOFF_TAG = "[CONTROLLER HANDOFF]"
THREAD_STATE_LEDGER_TAG = "[THREAD STATE LEDGER]"

BLOCKER_CLASSIFICATIONS = frozenset({
    "code blocker",
    "test blocker",
    "merge conflict",
    "stale head",
    "permission/capability blocker",
    "environment/tooling blocker",
    "process/rule blocker",
    "queue/lease blocker",
    "duplicate/canonicalization blocker",
    "no blocker",
})

_PRECISE_APPROVE_PHRASES = (
    "approve verdict prepared locally",
    "approved review posted to gitea",
    "request_changes posted to gitea",
)

_PRECISE_MERGE_PHRASES = (
    "merge performed",
    "merge not performed",
)

_PRECISE_SERVER_PHRASES = (
    "server-side state changed",
    "no server-side state changed",
)

_AMBIGUOUS_STANDALONE_TERMS = (
    "approved",
    "ready",
    "queued",
    "blocked",
    "done",
    "merged",
    "submitted",
    "validated",
)

_HANDOFF_TAG_RE = re.compile(r"\[CONTROLLER\s+HANDOFF\]", re.IGNORECASE)
_LEDGER_TAG_RE = re.compile(r"\[THREAD\s+STATE\s+LEDGER\]", re.IGNORECASE)
_LEGACY_HANDOFF_RE = re.compile(r"^##\s*Controller Handoff\s*$", re.I | re.M)

_MUTATION_LEDGER_RE = re.compile(
    r"server-side mutation ledger\s*:",
    re.IGNORECASE,
)
_BLOCKER_CLASS_RE = re.compile(
    r"blocker\s+classification\s*:\s*(.+)",
    re.IGNORECASE,
)

_LEDGER_REQUIRED_SECTIONS = (
    "what is true now",
    "what changed",
    "what is blocked",
    "who/what acts next",
)

_LEDGER_REQUIRED_FIELDS = (
    "server-side decision state",
    "local verdict/state",
    "next actor",
    "required action",
)


def _lower(text: str) -> str:
    return (text or "").lower()


def has_controller_handoff_marker(text: str) -> bool:
    return bool(_HANDOFF_TAG_RE.search(text or ""))


def has_thread_state_ledger_marker(text: str) -> bool:
    return bool(_LEDGER_TAG_RE.search(text or ""))


def _has_tagged_handoff(text: str) -> bool:
    return has_controller_handoff_marker(text)


def _has_tagged_ledger(text: str) -> bool:
    return has_thread_state_ledger_marker(text)


def _has_legacy_handoff(text: str) -> bool:
    return bool(_LEGACY_HANDOFF_RE.search(text or ""))


def _section_after_tag(text: str, tag_pattern: re.Pattern[str]) -> str | None:
    text = text or ""
    match = tag_pattern.search(text)
    if not match:
        return None
    start = match.start()
    other_tags = (
        _HANDOFF_TAG_RE,
        _LEDGER_TAG_RE,
    )
    end = len(text)
    for other in other_tags:
        if other.pattern == tag_pattern.pattern:
            continue
        later = other.search(text, match.end())
        if later:
            end = min(end, later.start())
    return text[start:end]


def _extract_mutation_ledger(text: str) -> str:
    lines = (text or "").splitlines()
    capture = False
    chunks: list[str] = []
    for line in lines:
        stripped = line.strip().lower()
        if "server-side mutation ledger" in stripped:
            capture = True
            continue
        if capture:
            if stripped.startswith("local-only") or stripped.startswith("blockers:"):
                break
            chunks.append(line)
    return "\n".join(chunks)


def _ambiguous_term_violations(text: str, *, scoped: bool) -> list[str]:
    if not scoped:
        return []
    reasons: list[str] = []
    lower = _lower(text)
    for term in _AMBIGUOUS_STANDALONE_TERMS:
        if not re.search(rf"\b{re.escape(term)}\b", lower):
            continue
        if term == "approved":
            if any(p in lower for p in _PRECISE_APPROVE_PHRASES):
                continue
            if "review decision:" in lower and "approve" in lower:
                continue
            if "approval_at_current_head" in lower:
                continue
        elif term == "merged":
            if any(p in lower for p in _PRECISE_MERGE_PHRASES):
                continue
            if "merge result:" in lower:
                continue
        elif term == "blocked":
            if any(c in lower for c in BLOCKER_CLASSIFICATIONS):
                continue
            if "blocker classification:" in lower:
                continue
        reasons.append(
            f"ambiguous standalone term '{term}' without precise state language"
        )
    return reasons


def _mutation_contradiction_reasons(handoff_text: str) -> list[str]:
    reasons: list[str] = []
    lower = _lower(handoff_text)
    ledger = _lower(_extract_mutation_ledger(handoff_text))
    no_server = (
        "no server-side state changed" in ledger
        or "none — no server-side" in ledger
        or re.search(r"\bnone\b", ledger)
    )
    claims_posted = "approved review posted to gitea" in lower
    claims_merge = "merge performed" in lower or (
        re.search(r"\bmerged\b", lower) and "merge not performed" not in lower
    )
    if claims_posted and no_server:
        reasons.append(
            "narrative claims APPROVED review posted but mutation ledger "
            "shows no server-side state changed"
        )
    if claims_merge and no_server:
        reasons.append(
            "narrative claims merge but mutation ledger lacks merge proof"
        )
    if (
        "approved review posted to gitea" in lower
        and "no server-side state changed" in ledger
    ):
        reasons.append(
            "mutation ledger contradicts APPROVED review posted claim"
        )
    return reasons


def _blocked_without_gate_reasons(handoff_text: str, ledger_text: str) -> list[str]:
    reasons: list[str] = []
    combined = _lower(handoff_text) + "\n" + _lower(ledger_text)
    if not re.search(r"\bblocked\b", combined):
        return reasons
    has_classification = bool(_BLOCKER_CLASS_RE.search(ledger_text or ""))
    has_gate = any(
        marker in combined
        for marker in (
            "exact gate",
            "failing gate",
            "blocker classification:",
            "process/rule blocker",
            "environment/tooling blocker",
            "permission/capability blocker",
        )
    )
    if not has_classification and not has_gate:
        reasons.append(
            "handoff or ledger says blocked but does not identify exact "
            "failing gate or blocker classification"
        )
    return reasons


def _local_server_separation_reasons(ledger_text: str) -> list[str]:
    reasons: list[str] = []
    lower = _lower(ledger_text)
    server_line = ""
    local_line = ""
    for line in (ledger_text or "").splitlines():
        stripped = line.strip().lower()
        if stripped.startswith("- server-side decision state:"):
            server_line = stripped
        if stripped.startswith("- local verdict/state:"):
            local_line = stripped
    if not server_line or not local_line:
        return reasons
    if "approve verdict prepared locally" in server_line:
        reasons.append(
            "local-only verdict incorrectly listed under server-side "
            "decision state"
        )
    if "approved review posted to gitea" in local_line:
        reasons.append(
            "server-side decision incorrectly listed under local verdict/state"
        )
    return reasons


def _ledger_field_reasons(ledger_text: str) -> list[str]:
    reasons: list[str] = []
    lower = _lower(ledger_text)
    for section in _LEDGER_REQUIRED_SECTIONS:
        if section not in lower:
            reasons.append(f"thread state ledger missing section '{section}'")
    for field in _LEDGER_REQUIRED_FIELDS:
        if field not in lower:
            reasons.append(f"thread state ledger missing field '{field}'")
    match = _BLOCKER_CLASS_RE.search(ledger_text or "")
    if not match:
        reasons.append("thread state ledger missing blocker classification")
    else:
        value = match.group(1).strip().lower().rstrip(".")
        if value not in BLOCKER_CLASSIFICATIONS:
            reasons.append(
                f"blocker classification '{value}' is not a recognized value"
            )
    if "do not do:" not in lower:
        reasons.append(
            "thread state ledger missing explicit 'Do not do' guidance"
        )
    return reasons


def _assessment(
    reasons: list[str],
    *,
    performed: bool = True,
) -> dict[str, Any]:
    return {
        "proven": not reasons,
        "block": bool(reasons),
        "reasons": reasons,
        "performed": performed,
    }


def assess_controller_handoff_comment(body: str) -> dict[str, Any]:
    """Validate a standalone ``[CONTROLLER HANDOFF]`` comment body."""
    text = body or ""
    if not _has_tagged_handoff(text):
        return _assessment([], performed=False)
    reasons: list[str] = []
    if not _MUTATION_LEDGER_RE.search(text):
        reasons.append(
            "controller handoff missing 'Server-side mutation ledger' section"
        )
    reasons.extend(_ambiguous_term_violations(text, scoped=True))
    reasons.extend(_mutation_contradiction_reasons(text))
    return _assessment(reasons)


def assess_thread_state_ledger_comment(body: str) -> dict[str, Any]:
    """Validate a standalone ``[THREAD STATE LEDGER]`` comment body."""
    text = body or ""
    if not _has_tagged_ledger(text):
        return _assessment([], performed=False)
    reasons = _ledger_field_reasons(text)
    reasons.extend(_ambiguous_term_violations(text, scoped=True))
    reasons.extend(_local_server_separation_reasons(text))
    return _assessment(reasons)


def assess_two_comment_pair(
    handoff_body: str,
    ledger_body: str,
) -> dict[str, Any]:
    """Validate paired Gitea comments (handoff then ledger)."""
    reasons: list[str] = []
    handoff = handoff_body or ""
    ledger = ledger_body or ""
    if _has_tagged_handoff(handoff) and not _has_tagged_ledger(ledger):
        reasons.append(
            "controller handoff posted without follow-up THREAD STATE LEDGER"
        )
    handoff_result = assess_controller_handoff_comment(handoff)
    ledger_result = assess_thread_state_ledger_comment(ledger)
    reasons.extend(handoff_result.get("reasons") or [])
    reasons.extend(ledger_result.get("reasons") or [])
    reasons.extend(_blocked_without_gate_reasons(handoff, ledger))
    if handoff_result.get("performed") or ledger_result.get("performed"):
        performed = True
    else:
        performed = False
    return _assessment(reasons, performed=performed)


def assess_two_comment_workflow(report_text: str) -> dict[str, Any]:
    """Validate combined final-report text containing both comment types."""
    text = report_text or ""
    if not (_has_tagged_handoff(text) or _has_tagged_ledger(text)):
        if _has_legacy_handoff(text):
            return _assessment([], performed=False)
        return _assessment([], performed=False)

    reasons: list[str] = []
    if _has_tagged_handoff(text) and not _has_tagged_ledger(text):
        reasons.append(
            "tagged controller handoff present without THREAD STATE LEDGER"
        )

    handoff_section = _section_after_tag(text, _HANDOFF_TAG_RE) or text
    ledger_section = _section_after_tag(text, _LEDGER_TAG_RE) or ""

    reasons.extend(
        (assess_controller_handoff_comment(handoff_section).get("reasons") or [])
    )
    if ledger_section:
        reasons.extend(
            (assess_thread_state_ledger_comment(ledger_section).get("reasons") or [])
        )
    reasons.extend(_blocked_without_gate_reasons(handoff_section, ledger_section))
    reasons.extend(_mutation_contradiction_reasons(handoff_section))
    reasons.extend(_local_server_separation_reasons(ledger_section))

    return _assessment(reasons)


def findings_for_final_report(report_text: str) -> list[dict[str, str]]:
    """Return final_report_validator findings for the two-comment workflow."""
    from final_report_validator import validator_finding

    if not (_has_tagged_handoff(report_text) or _has_tagged_ledger(report_text)):
        return []

    result = assess_two_comment_workflow(report_text)
    if not result.get("reasons"):
        return []

    findings: list[dict[str, str]] = []
    for reason in result.get("reasons") or []:
        severity = "block" if result.get("block") else "downgrade"
        findings.append(
            validator_finding(
                "shared.two_comment_workflow",
                severity,
                "Thread State Ledger",
                reason,
                "add a [THREAD STATE LEDGER] after [CONTROLLER HANDOFF] "
                "with precise server-side vs local state language",
            )
        )
    return findings