From 65b1c43c1b5fa74efec75d2f2b93d8a335828fcf Mon Sep 17 00:00:00 2001 From: zengyuanl Date: Thu, 20 Aug 2026 20:25:42 +0000 Subject: [PATCH] =?UTF-8?q?feat(benchmark):=20verified-retry=20orchestrato?= =?UTF-8?q?r=20=E2=80=94=20production-legal=20best-of-N=20with=20layered?= =?UTF-8?q?=20verifier?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Layered verification measured out of the TB-2.1 K3-max study: (1) executable checks derived blind from the task statement and run in the workspace, (2) confidence-gated LLM judge (accept >=0.85) as veto/fallback, (3) comparative pick over unverified candidates. Fresh workspace per attempt, early stop at first verified success, fail-closed on unparseable verdicts. Executor and LLM injected; stdlib-only core; 9 unit tests, no docker/network. Co-Authored-By: Claude Fable 5 --- benchmark/verified_retry/README.md | 46 ++++++ benchmark/verified_retry/__init__.py | 0 benchmark/verified_retry/checks.py | 68 +++++++++ benchmark/verified_retry/judge.py | 105 +++++++++++++ benchmark/verified_retry/orchestrator.py | 100 ++++++++++++ benchmark/verified_retry/spec.py | 98 ++++++++++++ benchmark/verified_retry/tests/__init__.py | 0 .../tests/test_verified_retry.py | 144 ++++++++++++++++++ 8 files changed, 561 insertions(+) create mode 100644 benchmark/verified_retry/README.md create mode 100644 benchmark/verified_retry/__init__.py create mode 100644 benchmark/verified_retry/checks.py create mode 100644 benchmark/verified_retry/judge.py create mode 100644 benchmark/verified_retry/orchestrator.py create mode 100644 benchmark/verified_retry/spec.py create mode 100644 benchmark/verified_retry/tests/__init__.py create mode 100644 benchmark/verified_retry/tests/test_verified_retry.py diff --git a/benchmark/verified_retry/README.md b/benchmark/verified_retry/README.md new file mode 100644 index 000000000..03ab59cc6 --- /dev/null +++ b/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` diff --git a/benchmark/verified_retry/__init__.py b/benchmark/verified_retry/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/benchmark/verified_retry/checks.py b/benchmark/verified_retry/checks.py new file mode 100644 index 000000000..750d45f96 --- /dev/null +++ b/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) diff --git a/benchmark/verified_retry/judge.py b/benchmark/verified_retry/judge.py new file mode 100644 index 000000000..f8fc889ad --- /dev/null +++ b/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": ""}}. 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": "", "reason": ""}}.""" + + +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" diff --git a/benchmark/verified_retry/orchestrator.py b/benchmark/verified_retry/orchestrator.py new file mode 100644 index 000000000..e35ed7b8a --- /dev/null +++ b/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, + ) diff --git a/benchmark/verified_retry/spec.py b/benchmark/verified_retry/spec.py new file mode 100644 index 000000000..f9696937c --- /dev/null +++ b/benchmark/verified_retry/spec.py @@ -0,0 +1,98 @@ +"""Core datatypes for the verified-retry orchestrator. + +Design provenance: parameters and layer choices come from the TB-2.1 +Kimi-K3-Max study (2026-08). Measured there: + - LLM judge from evidence alone: 82.5% verdict accuracy, 35.6% false-pass. + - Confidence gate at >=0.85 removes 14/16 false-passes (keeps 43/70 true). + - Comparative pick over candidates: 87% accuracy vs 62.5% random. + - Executable checks are the only layer that matches a benchmark verifier; + every observed false-pass came from trusting narrated (not executed) + evidence. +""" +from __future__ import annotations + +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path + +# An LLM is any prompt -> completion-text callable; tests inject fakes and +# production wraps an API client. Keeping this a bare callable keeps the +# orchestrator import-clean of any provider SDK. +LlmFn = Callable[[str], str] + + +@dataclass +class ExecutorResult: + """Outcome of one executor (agent) run inside a workspace.""" + + exit_code: int + transcript: str + wall_seconds: float = 0.0 + + @property + def crashed(self) -> bool: + return self.exit_code != 0 + + +# An executor runs the task in the given workspace and returns its result. +ExecutorFn = Callable[[Path, str], ExecutorResult] + + +@dataclass +class CheckResult: + """Result of executing the derived check script in a workspace.""" + + available: bool # False when no executable checks could be derived + passed: bool + output: str = "" + + +@dataclass +class Verdict: + """LLM judge verdict on a single attempt.""" + + passed: bool + confidence: float + reason: str = "" + + +@dataclass +class Attempt: + index: int + workspace: Path + executor: ExecutorResult + checks: CheckResult + verdict: Verdict | None + accepted: bool + + +@dataclass +class FinalResult: + """What the orchestrator hands back to the caller.""" + + accepted: bool # True iff an attempt passed verification + attempt_index: int # 1-based index of the returned attempt + workspace: Path # the returned attempt's (archived) workspace + attempts: list[Attempt] = field(default_factory=list) + pick_reason: str = "" # set when the comparative judge chose (unverified) + + @property + def n_attempts(self) -> int: + return len(self.attempts) + + +@dataclass +class TaskSpec: + task_id: str + instruction: str + workspace_src: Path # pristine task workspace; never mutated + max_attempts: int = 4 + # Acceptance requires the judge to say pass with at least this + # confidence (0.85 per the study's threshold sweep). Retries are cheap; + # false-passes are not. + accept_confidence: float = 0.85 + check_timeout_seconds: float = 300.0 + + def stamp(self) -> str: + return f"{self.task_id}-{int(time.time())}" diff --git a/benchmark/verified_retry/tests/__init__.py b/benchmark/verified_retry/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/benchmark/verified_retry/tests/test_verified_retry.py b/benchmark/verified_retry/tests/test_verified_retry.py new file mode 100644 index 000000000..449f8604e --- /dev/null +++ b/benchmark/verified_retry/tests/test_verified_retry.py @@ -0,0 +1,144 @@ +"""Unit tests for the verified-retry orchestrator (no network, no docker).""" +from __future__ import annotations + +import json +import sys +from pathlib import Path + +import pytest + +sys.path.insert(0, str(Path(__file__).resolve().parents[2])) + +from verified_retry.checks import derive_check_script, run_check_script +from verified_retry.judge import judge_attempt +from verified_retry.orchestrator import run_verified_retry +from verified_retry.spec import ExecutorResult, TaskSpec + +INSTRUCTION = "Write the exact string 42 into a file named answer.txt." +CHECK_SCRIPT = 'test -f answer.txt && [ "$(cat answer.txt)" = "42" ] && echo ok' + + +class FakeLlm: + """Scripted LLM: derives the check script, judges by peeking at check + output (deterministic), and always compares by picking attempt 2.""" + + def __init__(self, judge_confidence: float = 0.95, derive: str | None = CHECK_SCRIPT): + self.judge_confidence = judge_confidence + self.derive = derive + self.calls: list[str] = [] + + def __call__(self, prompt: str) -> str: + self.calls.append(prompt.split("\n", 1)[0][:60]) + if "automated acceptance check" in prompt: + return self.derive if self.derive is not None else "NO_CHECKS" + if "verifying whether ONE attempt" in prompt: + good = "ok" in prompt.split("[executed acceptance checks]")[-1] or "wrote 42" in prompt + return json.dumps( + {"pass": good, "confidence": self.judge_confidence, "reason": "scripted"} + ) + if "MOST LIKELY" in prompt: + return json.dumps({"pick": "2", "reason": "scripted compare"}) + raise AssertionError(f"unexpected prompt: {prompt[:80]}") + + +def make_executor(succeed_on: int | None, crash_on: set[int] = frozenset()): + """Executor stub: writes the right answer starting at attempt `succeed_on`, + a wrong answer otherwise; also asserts workspace freshness.""" + counter = {"n": 0} + + def executor(workspace: Path, instruction: str) -> ExecutorResult: + counter["n"] += 1 + n = counter["n"] + marker = workspace / "pollution.txt" + assert not marker.exists(), "workspace not pristine: prior attempt leaked" + marker.write_text("attempt ran here") + if n in crash_on: + return ExecutorResult(exit_code=1, transcript=f"attempt {n} crashed") + value = "42" if (succeed_on is not None and n >= succeed_on) else "41" + (workspace / "answer.txt").write_text(value) + return ExecutorResult(exit_code=0, transcript=f"attempt {n} wrote {value}") + + return executor, counter + + +def spec(tmp_path: Path, **kw) -> TaskSpec: + src = tmp_path / "src" + src.mkdir(exist_ok=True) + (src / "README").write_text(INSTRUCTION) + return TaskSpec( + task_id="t", instruction=INSTRUCTION, workspace_src=src, + **{"max_attempts": 4, **kw}, + ) + + +def test_accepts_first_verified_attempt_and_stops(tmp_path): + executor, counter = make_executor(succeed_on=1) + res = run_verified_retry(spec(tmp_path), executor, FakeLlm(), tmp_path / "a") + assert res.accepted and res.attempt_index == 1 and counter["n"] == 1 + assert (res.workspace / "answer.txt").read_text() == "42" + + +def test_retries_until_checks_pass(tmp_path): + executor, counter = make_executor(succeed_on=3) + res = run_verified_retry(spec(tmp_path), executor, FakeLlm(), tmp_path / "a") + assert res.accepted and res.attempt_index == 3 and counter["n"] == 3 + assert not res.attempts[0].accepted and not res.attempts[0].checks.passed + + +def test_never_solves_runs_all_attempts_and_flags_unverified(tmp_path): + executor, counter = make_executor(succeed_on=None) + res = run_verified_retry(spec(tmp_path), executor, FakeLlm(), tmp_path / "a") + assert not res.accepted and counter["n"] == 4 + assert res.attempt_index == 2 and res.pick_reason == "scripted compare" + + +def test_crash_is_never_accepted(tmp_path): + executor, counter = make_executor(succeed_on=2, crash_on={1}) + res = run_verified_retry(spec(tmp_path), executor, FakeLlm(), tmp_path / "a") + assert res.accepted and res.attempt_index == 2 + assert res.attempts[0].executor.crashed and res.attempts[0].verdict is None + + +def test_low_confidence_judge_vetoes_despite_passing_checks(tmp_path): + executor, counter = make_executor(succeed_on=1) + res = run_verified_retry( + spec(tmp_path), executor, FakeLlm(judge_confidence=0.6), tmp_path / "a" + ) + assert not res.accepted and counter["n"] == 4 # every attempt gated out + + +def test_no_checks_available_falls_back_to_judge_only(tmp_path): + executor, _ = make_executor(succeed_on=1) + llm = FakeLlm(derive=None) # NO_CHECKS + res = run_verified_retry(spec(tmp_path), executor, llm, tmp_path / "a") + assert res.accepted and res.attempts[0].checks.available is False + + +def test_check_script_runner_pass_and_fail(tmp_path): + ws = tmp_path / "ws" + ws.mkdir() + (ws / "answer.txt").write_text("42") + assert run_check_script(CHECK_SCRIPT, ws).passed + (ws / "answer.txt").write_text("41") + assert not run_check_script(CHECK_SCRIPT, ws).passed + + +def test_derive_handles_fences_and_no_checks(): + assert derive_check_script(lambda p: f"```bash\n{CHECK_SCRIPT}\n```", "x") == CHECK_SCRIPT + assert derive_check_script(lambda p: "NO_CHECKS", "x") is None + + +def test_unparseable_judge_fails_closed(): + v = judge_attempt(lambda p: "gibberish with no json", "task", "evidence", "") + assert not v.passed and v.confidence == 0.0 + + +def test_single_attempt_unverified_skips_compare_call(tmp_path): + executor, _ = make_executor(succeed_on=None) + llm = FakeLlm() + res = run_verified_retry( + spec(tmp_path, max_attempts=1), executor, llm, tmp_path / "a" + ) + assert not res.accepted and res.attempt_index == 1 + assert res.pick_reason == "single candidate; comparison skipped" + assert not any(c.startswith("Multiple independent") for c in llm.calls)