6krrt/verification.py

"""Pure verification of model responses.

Free, exact checks on what a model just returned, before the answer is
accepted. Like ``scoring.py`` and ``routing.py`` this module does no I/O and
makes no network calls, so it is testable in milliseconds and costs nothing
to run on every request.

**This module never executes model output.** ``eval_proficiency.py`` does run
generated code, but there the prompts are ones this project authored, so what
comes back is bounded. Here the code is whatever the user asked a model to
write — it could delete files, make requests, anything — and running it as a
side effect of *routing* would be indefensible. So the checks are structural:
parse it, don't run it. ``ast.parse`` builds a tree without evaluating,
``json.loads`` and ``yaml.safe_load`` never construct arbitrary objects, and
shell is checked with ``bash -n`` (parse-only) by the caller.

What this catches is the failure mode that has actually bitten this project
repeatedly: truncated and malformed output. A response cut off at the token
limit looks like a normal answer to a client and is silently broken.

Economics, measured on real traffic: a wasted cloud completion costs about 52
local checks at 1,500 tokens and 139 at 4,000, so a check that costs nothing
at all is worth running unconditionally.
"""

from __future__ import annotations

import ast
import json
import re
from dataclasses import dataclass, field
from typing import Literal, Optional

Verdict = Literal["ok", "truncated", "malformed", "unverifiable"]

# Languages whose syntax can be checked in-process without executing anything.
PYTHON_LANGS = frozenset({"python", "py", "python3"})
JSON_LANGS = frozenset({"json", "jsonc"})
YAML_LANGS = frozenset({"yaml", "yml"})
SHELL_LANGS = frozenset({"bash", "sh", "shell", "zsh"})

FENCE_RE = re.compile(
    r"```[ \t]*([A-Za-z0-9_+-]*)[ \t]*\r?\n(.*?)(?:```|\Z)", re.DOTALL
)


@dataclass
class Block:
    """One fenced block from a response."""

    lang: str
    code: str
    closed: bool


@dataclass
class Check:
    """The outcome of validating one block."""

    lang: str
    verdict: Verdict
    detail: str = ""


@dataclass
class Verification:
    """What we can say about a whole response."""

    verdict: Verdict
    checks: list[Check] = field(default_factory=list)
    detail: str = ""

    @property
    def failed(self) -> bool:
        return self.verdict in ("truncated", "malformed")


def extract_blocks(text: str) -> list[Block]:
    """Pull fenced blocks out of a response.

    An unterminated fence is kept and flagged rather than dropped — a response
    that runs out of tokens mid-block leaves exactly that, and it is the
    signal worth catching.
    """
    blocks = []
    for match in FENCE_RE.finditer(text or ""):
        lang = (match.group(1) or "").lower()
        body = match.group(2)
        closed = match.group(0).rstrip().endswith("```")
        blocks.append(Block(lang=lang, code=body, closed=closed))
    return blocks


def check_python(code: str) -> Check:
    """Parse Python without running it.

    ``ast.parse`` builds a syntax tree and evaluates nothing, so this is safe
    on arbitrary output. It catches truncation, unbalanced brackets, and the
    stray prose that models sometimes leave inside a fence.
    """
    try:
        ast.parse(code)
    except SyntaxError as e:
        return Check("python", "malformed", f"line {e.lineno}: {e.msg}")
    except (ValueError, MemoryError, RecursionError) as e:
        return Check("python", "malformed", f"{type(e).__name__}: {e}")
    return Check("python", "ok")


def check_json(code: str) -> Check:
    try:
        json.loads(code)
    except json.JSONDecodeError as e:
        return Check("json", "malformed", f"line {e.lineno}: {e.msg}")
    return Check("json", "ok")


def check_yaml(code: str) -> Check:
    # Imported lazily so the module stays dependency-free for callers that
    # only need the Python and JSON checks.
    try:
        import yaml
    except ImportError:  # pragma: no cover
        return Check("yaml", "unverifiable", "pyyaml not installed")
    try:
        yaml.safe_load(code)
    except yaml.YAMLError as e:
        return Check("yaml", "malformed", str(e).splitlines()[0][:120])
    return Check("yaml", "ok")


def check_block(block: Block) -> Check:
    """Validate one block, or report that nothing structural applies.

    An unclosed fence is reported as truncated regardless of language: the
    content may happen to parse, but the response was cut off mid-thought and
    the client is about to act on a fragment.
    """
    if not block.closed:
        return Check(block.lang or "?", "truncated", "unterminated code fence")
    if not block.code.strip():
        return Check(block.lang or "?", "unverifiable", "empty block")

    if block.lang in PYTHON_LANGS:
        return check_python(block.code)
    if block.lang in JSON_LANGS:
        return check_json(block.code)
    if block.lang in YAML_LANGS:
        return check_yaml(block.code)
    if block.lang in SHELL_LANGS:
        # Shell needs `bash -n`, which is a subprocess and therefore the
        # caller's job; this module stays I/O-free.
        return Check(block.lang, "unverifiable", "shell needs an external parse")
    return Check(block.lang or "?", "unverifiable", "no structural check for this language")


def verify_response(
    text: str,
    finish_reason: Optional[str] = None,
    has_tool_calls: bool = False,
) -> Verification:
    """Structurally verify a completion.

    ``has_tool_calls`` short-circuits everything below it: an agent turn that
    calls a tool is not a prose answer and cannot be judged as one. It outranks
    even truncation, because a turn routinely stops mid-sentence at the call
    boundary and that is correct behaviour, not a budget overrun.

    ``finish_reason == 'length'`` is otherwise decisive on its own: the model
    ran out of budget mid-answer, so whatever came back is a fragment even if
    it happens to parse. That is checked before the content rules because a
    truncated response that parses is the most dangerous case — it looks fine.

    A response with nothing checkable returns ``unverifiable``, which is NOT a
    failure. Most prose answers land here, and treating "we could not check
    this" as "this is wrong" would penalize models for the checker's limits —
    the same mistake as scoring a judge malfunction against a model.
    """
    # A turn that calls a tool is not a prose answer and must not be judged as
    # one. Its text is routinely empty, or a half-sentence before the call, and
    # both are correct behaviour.
    #
    # This was not theoretical. On a real agent session, 29 of 63 completions
    # were recorded as "malformed: empty response" purely for making tool
    # calls, and folding those in would have penalized the models that shipped
    # a working feature with 349 passing tests, clean mypy and clean ruff.
    if has_tool_calls:
        return Verification(
            "unverifiable", detail="tool-call turn; no prose answer to check"
        )

    if finish_reason == "length":
        return Verification("truncated", detail="finish_reason=length")

    # Decided in code, not by a model. An empty answer is unambiguously a
    # failure, and asking a 9.7B model about it produced a verdict of "ok"
    # with the reason "Answer is too short" — self-contradictory. Never ask a
    # model what an if-statement can settle.
    if not (text or "").strip():
        return Verification("malformed", detail="empty response")

    blocks = extract_blocks(text)
    if not blocks:
        return Verification("unverifiable", detail="no fenced blocks in response")

    checks = [check_block(b) for b in blocks]

    for verdict in ("truncated", "malformed"):
        bad = [c for c in checks if c.verdict == verdict]
        if bad:
            return Verification(
                verdict,
                checks=checks,
                detail=f"{len(bad)}/{len(checks)} blocks: {bad[0].lang}: {bad[0].detail}",
            )

    if any(c.verdict == "ok" for c in checks):
        return Verification("ok", checks=checks, detail=f"{len(checks)} block(s) parsed")
    return Verification(
        "unverifiable", checks=checks, detail="no block had a structural check"
    )


# --- local LLM verification (for responses nothing structural can check) ---

VERDICT_JSON_RE = re.compile(r"\{.*\}", re.DOTALL)

LOCAL_VERIFY_SYSTEM = (
    "You check whether an assistant's answer actually addresses the user's "
    "request. Reply with ONLY a JSON object: "
    '{"ok": true|false, "reason": "<at most 12 words>"}. '
    "Answer false ONLY for a clear failure: the answer is empty, refuses "
    "without cause, ENDS mid-sentence, contradicts itself, or responds to a "
    "different question. Style, brevity and debatable choices are NOT "
    "failures. When unsure, answer true.\n"
    "IMPORTANT: a marker reading '[... N characters elided from the middle "
    "...]' is this harness shortening a long answer so you can read it. It is "
    "NOT a defect. Ignore it and judge only the beginning and the ending you "
    "were given."
)


ELISION = "\n\n[... {n} characters elided from the middle ...]\n\n"


def excerpt(text: str, limit: int) -> str:
    """Shorten text for the checker WITHOUT making it look truncated.

    Naive head-truncation is not safe here. A complete 6,637-character answer
    cut to 4,000 reaches the checker ending mid-sentence, and it duly reports
    "cut off mid-thought" — a false failure caused entirely by the harness.
    That happened on the first live run, and had it been wired into
    proficiency it would have recorded every long answer as a failure.

    So the middle is dropped instead of the tail, and the cut is labelled, so
    the ending the checker judges is the real ending.
    """
    if not text or len(text) <= limit:
        return text
    head = limit // 2
    tail = limit - head
    dropped = len(text) - limit
    return text[:head] + ELISION.format(n=dropped) + text[-tail:]


def build_local_verify_prompt(request_text: str, answer: str, limit: int = 4000) -> str:
    """Frame one answer for the local checker.

    Both sides are shortened — the local model is small and slow, and a check
    that reads 100k tokens costs more than the answer it guards — but via
    ``excerpt``, so an elision is never mistaken for a truncated answer.
    """
    return (
        f"USER'S REQUEST:\n{excerpt(request_text, limit)}\n\n"
        f"ASSISTANT'S ANSWER:\n{excerpt(answer, limit)}"
    )


def parse_verdict_json(raw: str) -> Optional[dict]:
    """Pull the verdict object out of a local model's reply.

    Same defence as the eval judge: local models are reasoning models and leak
    their thinking into the content despite response_format, so the object
    usually arrives wrapped in prose. Returns None when nothing usable came
    back, so the caller records no sample rather than a false verdict.
    """
    if not raw:
        return None
    candidates = [raw]
    match = VERDICT_JSON_RE.search(raw)
    if match:
        candidates.append(match.group(0))
    for candidate in candidates:
        try:
            parsed = json.loads(candidate)
        except json.JSONDecodeError:
            continue
        if isinstance(parsed, dict) and "ok" in parsed:
            return parsed
    return None


def interpret_local_verdict(parsed: Optional[dict]) -> Optional[Check]:
    """Turn a parsed local verdict into a Check, or None if unusable.

    A malfunctioning checker must never produce a failure verdict: that would
    charge the model for the checker's problem, which is the mistake this
    project already made once with the eval judge.
    """
    if parsed is None:
        return None
    ok = parsed.get("ok")
    if not isinstance(ok, bool):
        return None
    reason = str(parsed.get("reason", ""))[:120]
    return Check("local_llm", "ok" if ok else "malformed", reason)


def worth_local_check(
    verdict: Verdict,
    completion_tokens: Optional[int],
    min_tokens: int,
    has_tool_calls: bool = False,
) -> bool:
    """Whether a local LLM check earns its cost on this response.

    Only for responses nothing structural could judge — if the code already
    parsed, or failed to, that verdict is exact and free and a fuzzy second
    opinion adds nothing.

    The size gate is economics, measured on real traffic: a local check costs
    about 15% of a median 193-token answer, so it pays only if such answers
    fail more than ~15% of the time. On a 1,500-token answer the break-even
    failure rate drops to ~1.9%, which is plausible. Small answers are simply
    not worth checking.
    """
    if verdict != "unverifiable":
        return False
    if has_tool_calls:
        # Same reason the structural check declines: an agent turn ending in a
        # tool call reads as "cuts off mid-sentence" to a checker expecting a
        # finished answer. Observed producing 8 false failures in 9 checks on
        # real agent traffic.
        return False
    return bool(completion_tokens and completion_tokens >= min_tokens)