feat(benchmark): verified-retry orchestrator — production-legal best-of-N with a layered verifier

benchmark/verified_retry/README.md

@@ -0,0 +1,46 @@
+# Verified-retry orchestrator
+
+A production-legal best-of-N harness: run an executor agent up to N times in
+fresh workspaces, verify each attempt WITHOUT any benchmark answer key, stop
+at the first verified success, and fall back to a comparative LLM pick when
+nothing verifies.
+
+## Why this shape (measured, TB-2.1 Kimi-K3-Max study, 2026-08)
+
+- Retries are the dominant accuracy lever: single-attempt 70.5% vs oracle
+  pass@4 83.1% (K3 baseline, 4 replicate runs). No advisor/prompt config
+  moved the single-attempt mean.
+- An LLM judge alone is NOT verifier-strength: 82.5% verdict accuracy with a
+  35.6% false-pass rate on truly-failing attempts. Every false-pass came
+  from trusting narrated (not executed) evidence.
+- Judge confidence separates truth: true-pass mean 0.83 vs false-pass 0.69;
+  gating acceptance at >=0.85 removed 14/16 false-passes.
+- Comparative picking is much stronger than absolute judging: 87% correct
+  picks (vs 62.5% random), implying ~78.7%/89 for judge-selected best-of-4 —
+  ≈ a frontier model's single-attempt score, with no oracle.
+
+## The three verification layers
+
+1. `checks.py` — an LLM derives an executable check script from the task
+   statement ONLY, BEFORE any attempt exists (prevents inheriting a
+   solution's misreading), and the script is executed in the workspace.
+   Execution outranks opinion; where the statement is machine-checkable this
+   layer is benchmark-verifier-strength by construction.
+2. `judge.py::judge_attempt` — confidence-gated LLM verdict on demonstrated
+   evidence; veto layer over checks, sole verifier when no checks exist.
+   Unparseable verdicts fail closed.
+3. `judge.py::judge_compare` — when no attempt verifies within budget, a
+   comparative pick over archived candidates, explicitly flagged unverified.
+
+## Boundaries
+
+- This is NOT a Switchyard route/strategy: the gateway never touches the
+  workspace, and executing checks requires the workspace. It wraps the
+  executor where the sandbox lives.
+- Executor and LLM are injected callables (`spec.ExecutorFn`, `spec.LlmFn`) —
+  no provider SDK, no docker required for the core. Container/TB-image
+  integration is a follow-up on a docker-capable host.
+- Benchmark scores produced with this harness are pass@N-with-own-verifier;
+  report them as such, never as single-attempt accuracy.
+
+Run tests: `uv run pytest benchmark/verified_retry/tests -q`

benchmark/verified_retry/checks.py

@@ -0,0 +1,68 @@
+"""Layer 1: executable checks derived from the task statement.
+
+The check script is derived ONCE, from the instruction only, BEFORE any
+attempt runs. That ordering is load-bearing: a checker written after seeing
+a solution inherits the solution's misreading of the task (the observed
+failure mode on ambiguous tasks), while one written blind encodes only the
+stated requirements.
+"""
+from __future__ import annotations
+
+import re
+import subprocess
+from pathlib import Path
+
+from .spec import CheckResult, LlmFn
+
+DERIVE_PROMPT = """You are writing an automated acceptance check for a task, \
+BEFORE any attempt at the task exists. You see only the task statement.
+
+Write a single bash script that exits 0 if and only if a workspace satisfies \
+every requirement the task statement makes machine-checkable: required files \
+existing at their exact stated paths, outputs matching stated values or \
+tolerances, commands the statement says must succeed, services it says must \
+respond. Execute real commands against the workspace (the script runs with \
+the workspace as its working directory); print a line per check so failures \
+are diagnosable. Do NOT attempt to solve the task inside the script, and do \
+NOT invent requirements the statement does not make.
+
+If the statement contains NOTHING machine-checkable, output exactly the \
+single line NO_CHECKS instead of a script.
+
+Task statement:
+---
+{instruction}
+---
+
+Reply with only the bash script (or NO_CHECKS), no commentary."""
+
+
+def derive_check_script(llm: LlmFn, instruction: str) -> str | None:
+    """Ask the LLM for a check script; None when nothing is checkable."""
+    reply = llm(DERIVE_PROMPT.format(instruction=instruction)).strip()
+    if not reply or "NO_CHECKS" in reply.splitlines()[0]:
+        return None
+    # Tolerate a fenced code block.
+    m = re.search(r"```(?:bash|sh)?\n(.*?)```", reply, re.DOTALL)
+    script = m.group(1) if m else reply
+    return script.strip() or None
+
+
+def run_check_script(
+    script: str | None, workspace: Path, timeout_seconds: float = 300.0
+) -> CheckResult:
+    """Execute the derived checks inside the workspace."""
+    if script is None:
+        return CheckResult(available=False, passed=False, output="no executable checks")
+    try:
+        proc = subprocess.run(
+            ["bash", "-c", script],
+            cwd=workspace,
+            capture_output=True,
+            text=True,
+            timeout=timeout_seconds,
+        )
+    except subprocess.TimeoutExpired:
+        return CheckResult(available=True, passed=False, output="check script timed out")
+    output = (proc.stdout + proc.stderr)[-8000:]
+    return CheckResult(available=True, passed=proc.returncode == 0, output=output)

benchmark/verified_retry/judge.py

@@ -0,0 +1,105 @@
+"""Layers 2-3: confidence-gated LLM judge and comparative candidate pick.
+
+Measured behaviour this module is designed around (TB-2.1 study, 2026-08):
+judges are weak at absolute verdicts (35.6% false-pass from narrated
+evidence) but strong comparatively (87% correct picks), and their stated
+confidence separates true from false passes (0.83 vs 0.69 mean) - so the
+absolute verdict is gated on confidence and used as a veto layer on top of
+executed checks, never as the sole verifier.
+"""
+from __future__ import annotations
+
+import json
+import re
+
+from .spec import LlmFn, Verdict
+
+JUDGE_PROMPT = """You are verifying whether ONE attempt at a task actually \
+satisfied the task's stated requirements. Judge only from demonstrated \
+evidence: commands the attempt ran and their observed output, executed check \
+results, produced artifacts. Confident prose is not evidence. Re-read the \
+exact stated deliverables (paths, formats, tolerances) and confirm they were \
+literally met.
+
+Task statement:
+---
+{instruction}
+---
+Executed acceptance checks (ground evidence, weigh heavily):
+---
+{check_output}
+---
+Attempt evidence (final portion of the executor transcript):
+---
+{evidence}
+---
+
+Reply with a single JSON object: {{"pass": true|false, "confidence": 0.0-1.0, \
+"reason": "<one sentence>"}}. confidence 0.5 means coin-flip, 1.0 certain. \
+If the evidence does not DEMONSTRATE a stated requirement, do not assume it."""
+
+COMPARE_PROMPT = """Multiple independent attempts at the same task are below. \
+None passed automated verification, so pick the attempt MOST LIKELY to \
+satisfy the task's stated requirements, comparing demonstrated evidence only.
+
+Task statement:
+---
+{instruction}
+---
+{candidates}
+
+Reply with a single JSON object: {{"pick": "<attempt id>", "reason": "<one \
+sentence>"}}."""
+
+
+def _extract_json(text: str) -> dict:
+    m = re.search(r"\{.*\}", text, re.DOTALL)
+    if not m:
+        raise ValueError(f"judge reply had no JSON object: {text[:200]!r}")
+    return json.loads(m.group(0))
+
+
+def judge_attempt(
+    llm: LlmFn, instruction: str, evidence: str, check_output: str
+) -> Verdict:
+    reply = llm(
+        JUDGE_PROMPT.format(
+            instruction=instruction,
+            check_output=check_output or "(no executable checks were available)",
+            evidence=evidence,
+        )
+    )
+    try:
+        obj = _extract_json(reply)
+        return Verdict(
+            passed=bool(obj["pass"]),
+            confidence=float(obj.get("confidence", 0.0)),
+            reason=str(obj.get("reason", "")),
+        )
+    except (ValueError, KeyError, TypeError, json.JSONDecodeError):
+        # Unparseable verdict fails CLOSED: an unverified attempt is not
+        # accepted (retries are cheap; shipping unverified work is not).
+        return Verdict(passed=False, confidence=0.0, reason="unparseable judge reply")
+
+
+def judge_compare(
+    llm: LlmFn, instruction: str, candidates: dict[str, str]
+) -> tuple[str, str]:
+    """Pick the best of several failed-verification candidates.
+
+    candidates maps attempt-id -> evidence text. Returns (pick, reason);
+    falls back to the last candidate if the reply is unusable.
+    """
+    blocks = "\n".join(
+        f"Attempt {cid}:\n---\n{ev}\n---" for cid, ev in candidates.items()
+    )
+    reply = llm(COMPARE_PROMPT.format(instruction=instruction, candidates=blocks))
+    fallback = list(candidates)[-1]
+    try:
+        obj = _extract_json(reply)
+        pick = str(obj.get("pick", ""))
+        if pick not in candidates:
+            return fallback, "judge picked unknown id; returned last attempt"
+        return pick, str(obj.get("reason", ""))
+    except (ValueError, json.JSONDecodeError):
+        return fallback, "unparseable compare reply; returned last attempt"

benchmark/verified_retry/orchestrator.py

@@ -0,0 +1,100 @@
+"""The verified-retry loop: fresh attempt -> execute checks -> gated judge ->
+accept or retry; comparative pick if nothing verifies.
+
+Acceptance policy (in order):
+  1. The executor must not have crashed.
+  2. If executable checks exist, they MUST pass (execution outranks opinion).
+  3. The judge must say pass with confidence >= spec.accept_confidence.
+     When checks passed, the judge acts as a veto layer; when no checks
+     could be derived, the judge is the only verifier and the gate matters
+     most.
+Every attempt starts from a pristine copy of the task workspace - attempt
+independence is what makes retries worth anything.
+"""
+from __future__ import annotations
+
+import shutil
+import tempfile
+import time
+from pathlib import Path
+
+from .checks import derive_check_script, run_check_script
+from .judge import judge_attempt, judge_compare
+from .spec import Attempt, ExecutorFn, FinalResult, LlmFn, TaskSpec
+
+EVIDENCE_TAIL_CHARS = 40_000
+
+
+def _evidence(transcript: str, check_output: str) -> str:
+    tail = transcript[-EVIDENCE_TAIL_CHARS:]
+    return f"{tail}\n\n[executed acceptance checks]\n{check_output}"
+
+
+def run_verified_retry(
+    spec: TaskSpec,
+    executor: ExecutorFn,
+    llm: LlmFn,
+    archive_dir: Path | None = None,
+) -> FinalResult:
+    archive_root = Path(archive_dir or tempfile.mkdtemp(prefix="vr-")) / spec.stamp()
+    archive_root.mkdir(parents=True, exist_ok=True)
+
+    # Derive checks from the instruction ONLY, before any attempt exists.
+    check_script = derive_check_script(llm, spec.instruction)
+
+    attempts: list[Attempt] = []
+    for i in range(1, spec.max_attempts + 1):
+        workspace = archive_root / f"attempt-{i}"
+        shutil.copytree(spec.workspace_src, workspace)
+
+        started = time.monotonic()
+        result = executor(workspace, spec.instruction)
+        result.wall_seconds = time.monotonic() - started
+
+        checks = run_check_script(check_script, workspace, spec.check_timeout_seconds)
+
+        verdict = None
+        accepted = False
+        if not result.crashed and (not checks.available or checks.passed):
+            verdict = judge_attempt(
+                llm,
+                spec.instruction,
+                _evidence(result.transcript, checks.output),
+                checks.output if checks.available else "",
+            )
+            accepted = verdict.passed and verdict.confidence >= spec.accept_confidence
+
+        attempts.append(
+            Attempt(
+                index=i,
+                workspace=workspace,
+                executor=result,
+                checks=checks,
+                verdict=verdict,
+                accepted=accepted,
+            )
+        )
+        if accepted:
+            return FinalResult(
+                accepted=True, attempt_index=i, workspace=workspace, attempts=attempts
+            )
+
+    # Nothing verified: comparative judge picks the least-bad candidate.
+    # The result is explicitly flagged unverified (accepted=False).
+    non_crashed = [a for a in attempts if not a.executor.crashed] or attempts
+    candidates = {
+        str(a.index): _evidence(a.executor.transcript, a.checks.output)
+        for a in non_crashed
+    }
+    if len(candidates) == 1:
+        pick, reason = next(iter(candidates)), "single candidate; comparison skipped"
+    else:
+        pick, reason = judge_compare(llm, spec.instruction, candidates)
+    picked = next(a for a in attempts if str(a.index) == pick)
+    return FinalResult(
+        accepted=False,
+        attempt_index=picked.index,
+        workspace=picked.workspace,
+        attempts=attempts,
+        pick_reason=reason,
+    )