"""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": ""}. ' "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)