diff --git a/README.md b/README.md index 4ddc1a2..7027895 100644 --- a/README.md +++ b/README.md @@ -38,6 +38,7 @@ kbench snapshot your-node # live serving flags, for models.yaml kbench perf brain # perf sweep only (no task authoring needed) kbench run brain --limit 50 # quality + perf -> results/*.json +kbench import-verifiers brain --task kbench results # every score card recorded so far kbench card results/ # re-render one ``` @@ -48,6 +49,13 @@ Raw Inspect logs land in `logs/` (gitignored). View them with: .venv/bin/inspect view --log-dir logs ``` +Prime Verifiers v1 `traces.jsonl` can enter the same scorecard with +`kbench import-verifiers`. Bench reads the upstream episode artifact; it does +not execute or copy Verifiers' harness. The importer stores episode scores, +task-content fingerprints and short assistant excerpts, never task data, tool +arguments or full conversations. Inspect AI remains the native runner for +Bench tasks. + ## Layout ``` @@ -58,6 +66,7 @@ kbench/ perf.py async load generator, any OpenAI-compatible endpoint results.py the committed result schema run.py orchestration: target + tasks + perf -> score card + verifiers_import.py upstream Verifiers episode JSONL -> QualityResult data/ public/ a few example samples, shown on the site private/ the real holdout — never leaves this repo diff --git a/docs/DECISIONS.md b/docs/DECISIONS.md index 9277a22..bc341f4 100644 --- a/docs/DECISIONS.md +++ b/docs/DECISIONS.md @@ -198,3 +198,33 @@ reverse-engineer the private set (D3) by probing it one sample at a time. **Why.** The product does not need a public code-execution service. Manual approval is the intended workflow and keeps ownership of scarce hardware and private evaluation data clear. The Lumbridge control plane stores the inert review record with status `suggested`; only an owner can advance its lifecycle. **Supersedes D8 for product behavior.** The sandbox described in D8 is no longer a launch dependency because no public suggestion is automatically executed. If automated third-party execution is ever proposed again, it requires a new explicit decision and the full D8 isolation boundary first. + +--- + +## D11 — Import Verifiers episodes; do not add another harness + +**Decision.** Inspect AI remains Bench's native execution engine. Online Tera +and Prime-RL evaluations run in upstream Prime Verifiers and enter Bench through +the durable Verifiers v1 `traces.jsonl` artifact. Bench parses that artifact +into its existing `QualityResult` and `SampleOutcome` fields; the scorecard +schema is not changed. + +One Verifiers episode becomes one Bench sample. Only trainable traces contribute +to its score, so a modeled user or judge seat does not dilute the assistant's +result. By default the score is the mean summed weighted reward of those traces; +a named trace metric can be selected explicitly. The card stores episode and +trace ids, task identity hashes, reward names, pass/fail and at most the normal +400-character assistant excerpt. It never copies task data, prompts, tool +arguments, error messages or the full conversation. + +**Why.** Re-running a Verifiers environment through an Inspect-shaped adapter +would duplicate tool-loop and multi-agent behavior and could change the thing +being measured. The completed episode is already the evidence boundary. Bench's +unique work is longitudinal comparison against an exact model/host target, not +owning every runner. + +**Cost.** Inspect and Verifiers logs have different notions of a sample and a +headline score. The import command therefore requires a target and task name, +records its pass threshold and source-file digest, and labels sampling as +Verifiers rather than pretending Inspect produced it. Comparability still +requires the same target snapshot, task fingerprint, score source and threshold. diff --git a/kbench/cli.py b/kbench/cli.py index fc1b3ca..4a43828 100644 --- a/kbench/cli.py +++ b/kbench/cli.py @@ -12,11 +12,12 @@ from rich.console import Console from rich.table import Table from . import tasks as task_catalog -from .compare import compare as compare_runs, direction -from .sweep import analyse, point_from_run +from .compare import compare as compare_runs +from .compare import direction from .registry import load_registry from .results import RESULTS_DIR, Run from .run import build_run, run_perf, run_quality +from .sweep import analyse, point_from_run app = typer.Typer( add_completion=False, @@ -307,6 +308,51 @@ def card_cmd(path: Path) -> None: _print_card_dict(data) +@app.command("import-verifiers") +def import_verifiers_cmd( + target: str = typer.Argument(..., help="registered model/host target that produced the traces"), + path: Path = typer.Argument(..., help="Verifiers run directory or traces.jsonl"), + task: str = typer.Option(..., help="Bench task name for this imported environment"), + tier: str = typer.Option("signal", help="reference | signal | canary"), + pass_threshold: float = typer.Option(1.0, help="score required for an episode to pass"), + primary_metric: str | None = typer.Option( + None, help="trace metric to import; defaults to summed weighted rewards" + ), + save: bool = typer.Option(True, help="write the existing results/*.json scorecard"), +) -> None: + """Import upstream Verifiers episodes; execute no model and no harness.""" + from .verifiers_import import FORMAT, VerifiersImportError, import_quality, source_sha256 + + registry = load_registry() + registered = registry.target(target) + try: + quality = import_quality( + path, + task=task, + tier=tier, + pass_threshold=pass_threshold, + primary_metric=primary_metric, + ) + digest = source_sha256(path) + except VerifiersImportError as exc: + raise typer.BadParameter(str(exc), param_hint="path") from exc + + run = build_run(registry, registered, quality=[quality], perf=None) + run.runner["quality_import"] = { + "format": FORMAT, + "source_sha256": digest, + } + run.sampling = { + "source": "verifiers", + "primary_metric": primary_metric, + "pass_threshold": pass_threshold, + } + _print_card(run) + if save: + saved = run.save() + console.print(f"\n[green]saved[/green] {saved.relative_to(Path.cwd())}") + + def _print_card(run: Run) -> None: _print_card_dict(run.to_dict()) diff --git a/kbench/verifiers_import.py b/kbench/verifiers_import.py new file mode 100644 index 0000000..d0bbdab --- /dev/null +++ b/kbench/verifiers_import.py @@ -0,0 +1,287 @@ +"""Import Prime Verifiers v1 episode JSONL into Bench's existing scorecard. + +Verifiers owns rollout execution and its trace schema. Bench reads the durable +``traces.jsonl`` artifact with the standard library, reduces one episode to one +sample outcome, and keeps its own result schema unchanged. Raw prompts, task +data, tool arguments and full conversations never enter the committed card. +""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path +from typing import Any + +from .results import QualityResult, SampleOutcome, truncate + +FORMAT = "verifiers-v1-episode-jsonl" + + +class VerifiersImportError(ValueError): + pass + + +def traces_path(path: str | Path) -> Path: + candidate = Path(path) + if candidate.is_dir(): + candidate = candidate / "traces.jsonl" + if not candidate.is_file(): + raise VerifiersImportError(f"Verifiers traces file not found: {candidate}") + return candidate + + +def source_sha256(path: str | Path) -> str: + return hashlib.sha256(traces_path(path).read_bytes()).hexdigest() + + +def _reward_value(rewards: Any) -> tuple[float | None, list[str]]: + if not isinstance(rewards, dict): + return None, [] + total = 0.0 + found = False + names: list[str] = [] + for name, reward in sorted(rewards.items()): + if reward is None: + continue + if isinstance(reward, (int, float)) and not isinstance(reward, bool): + score, weight = float(reward), 1.0 + elif isinstance(reward, dict) and isinstance(reward.get("score"), (int, float)): + score = float(reward["score"]) + weight = float(reward.get("weight", 1.0)) + else: + continue + found = True + names.append(str(name)) + total += score * weight + return (total if found else None), names + + +def _trace_value(trace: dict[str, Any], primary_metric: str | None) -> tuple[float | None, list[str]]: + if primary_metric: + metrics = trace.get("metrics") or {} + value = metrics.get(primary_metric) if isinstance(metrics, dict) else None + if isinstance(value, (int, float)) and not isinstance(value, bool): + return float(value), [primary_metric] + return None, [] + return _reward_value(trace.get("rewards")) + + +def _content_text(content: Any) -> str: + if isinstance(content, str): + return content + if isinstance(content, list): + parts: list[str] = [] + for part in content: + if isinstance(part, str): + parts.append(part) + elif isinstance(part, dict) and isinstance(part.get("text"), str): + parts.append(part["text"]) + return "".join(parts) + return "" + + +def _last_assistant_text(trace: dict[str, Any]) -> str | None: + for node in reversed(trace.get("nodes") or []): + if not isinstance(node, dict): + continue + message = node.get("message") or {} + if isinstance(message, dict) and message.get("role") == "assistant": + text = _content_text(message.get("content")).strip() + if text: + return text + completion = trace.get("completion") + return completion.strip() if isinstance(completion, str) and completion.strip() else None + + +def _eligible_traces(episode: dict[str, Any]) -> list[dict[str, Any]]: + traces = episode.get("traces") + if not isinstance(traces, list): + raise VerifiersImportError("episode field 'traces' must be a list") + typed = [trace for trace in traces if isinstance(trace, dict)] + trainable = [ + trace + for trace in typed + if (trace.get("agent") or {}).get("trainable", True) is not False + ] + # Eval files from older v1 builds may omit trainable. If an env explicitly + # marks every trace non-trainable, keep the traces rather than manufacturing + # an empty sample. + return trainable or typed + + +def _task_identity(episode: dict[str, Any]) -> str: + task = episode.get("task") or {} + if not isinstance(task, dict): + task = {} + durable = task.get("key") or task.get("hash") + if durable: + return str(durable) + # Hash task data locally if an older trace lacks key/hash. The card receives + # only this digest; private task content is never copied into it. + canonical = json.dumps(task.get("data"), sort_keys=True, separators=(",", ":")) + return hashlib.sha256(canonical.encode()).hexdigest() + + +def _dataset_fingerprint(episodes: list[dict[str, Any]]) -> str | None: + if not episodes: + return None + identities = [ + f"{(episode.get('env') or {}).get('id', '')}\x1f{_task_identity(episode)}" + for episode in episodes + ] + digest = hashlib.sha256() + for identity in sorted(identities): + digest.update(identity.encode()) + digest.update(b"\x1e") + return f"sha256:{digest.hexdigest()[:16]}" + + +def _duration(episodes: list[dict[str, Any]]) -> float | None: + total = 0.0 + measured = False + for episode in episodes: + starts: list[float] = [] + ends: list[float] = [] + for trace in episode.get("traces") or []: + timing = trace.get("timing") if isinstance(trace, dict) else None + if not isinstance(timing, dict): + continue + for span in timing.values(): + if not isinstance(span, dict): + continue + start, end = span.get("start"), span.get("end") + if isinstance(start, (int, float)) and start > 0: + starts.append(float(start)) + if isinstance(end, (int, float)) and end > 0: + ends.append(float(end)) + if starts and ends and max(ends) >= min(starts): + total += max(ends) - min(starts) + measured = True + return round(total, 6) if measured else None + + +def read_episodes(path: str | Path) -> list[dict[str, Any]]: + source = traces_path(path) + episodes: list[dict[str, Any]] = [] + ids: set[str] = set() + for line_number, line in enumerate(source.read_text().splitlines(), start=1): + if not line.strip(): + continue + try: + episode = json.loads(line) + except json.JSONDecodeError as exc: + raise VerifiersImportError(f"{source}:{line_number}: invalid JSON: {exc.msg}") from exc + if not isinstance(episode, dict): + raise VerifiersImportError(f"{source}:{line_number}: episode must be an object") + episode_id = str(episode.get("id") or "") + if not episode_id: + raise VerifiersImportError(f"{source}:{line_number}: episode has no id") + if episode_id in ids: + raise VerifiersImportError(f"{source}:{line_number}: duplicate episode id {episode_id!r}") + ids.add(episode_id) + _eligible_traces(episode) # validate the durable v1 boundary now + episodes.append(episode) + if not episodes: + raise VerifiersImportError(f"{source} contains no episodes") + return episodes + + +def import_quality( + path: str | Path, + task: str, + tier: str = "signal", + pass_threshold: float = 1.0, + primary_metric: str | None = None, +) -> QualityResult: + """Reduce Verifiers episodes to one existing Bench QualityResult.""" + if tier not in {"reference", "signal", "canary"}: + raise VerifiersImportError("tier must be reference, signal or canary") + episodes = read_episodes(path) + outcomes: list[SampleOutcome] = [] + versions: set[str] = set() + + for episode in episodes: + traces = _eligible_traces(episode) + values: list[float] = [] + reward_names: set[str] = set() + excerpts: list[str] = [] + trace_ids: list[str] = [] + agents: list[str] = [] + trace_ok = True + error_types = { + str(error.get("type")) + for error in (episode.get("errors") or []) + if isinstance(error, dict) and error.get("type") + } + for trace in traces: + value, names = _trace_value(trace, primary_metric) + if value is not None: + values.append(value) + reward_names.update(names) + trace_ids.append(str(trace.get("id") or "")) + agent = trace.get("agent") or {} + agents.append(str(agent.get("name") or "agent")) + trace_ok = trace_ok and bool(trace.get("ok", False)) + verifiers = trace.get("verifiers") or {} + if isinstance(verifiers, dict) and verifiers.get("version"): + versions.add(str(verifiers["version"])) + error_types.update( + str(error.get("type")) + for error in (trace.get("errors") or []) + if isinstance(error, dict) and error.get("type") + ) + text = _last_assistant_text(trace) + if text: + excerpts.append(f"{agents[-1]}: {text}" if len(traces) > 1 else text) + + score = sum(values) / len(values) if values else 0.0 + episode_ok = bool(episode.get("ok", False)) and trace_ok + outcomes.append( + SampleOutcome( + sample_id=str(episode["id"]), + score=round(score, 6), + passed=episode_ok and score >= pass_threshold, + excerpt=truncate("\n\n".join(excerpts) or None), + metadata={ + "source": FORMAT, + "env": str((episode.get("env") or {}).get("id") or ""), + "task_key": _task_identity(episode), + "trace_ids": trace_ids, + "agents": agents, + "episode_ok": episode_ok, + "score_source": f"metric:{primary_metric}" if primary_metric else "weighted_rewards", + "reward_names": sorted(reward_names), + "error_types": sorted(error_types), + "ungraded": not values, + }, + ) + ) + + accuracy = sum(outcome.passed for outcome in outcomes) / len(outcomes) + mean_score = sum(outcome.score for outcome in outcomes) / len(outcomes) + metrics = { + "accuracy": round(accuracy, 6), + "mean_score": round(mean_score, 6), + "pass_threshold": float(pass_threshold), + } + return QualityResult( + task=task, + tier=tier, + dataset_version=("verifiers:" + ",".join(sorted(versions))) if versions else None, + dataset_fingerprint=_dataset_fingerprint(episodes), + n_samples=len(episodes), + metrics=metrics, + samples=outcomes, + duration_s=_duration(episodes), + ) + + +__all__ = [ + "FORMAT", + "VerifiersImportError", + "import_quality", + "read_episodes", + "source_sha256", + "traces_path", +] diff --git a/tests/test_verifiers_import.py b/tests/test_verifiers_import.py new file mode 100644 index 0000000..38b3f96 --- /dev/null +++ b/tests/test_verifiers_import.py @@ -0,0 +1,122 @@ +"""Prime Verifiers v1 episode JSONL enters Bench without a second harness.""" + +from __future__ import annotations + +import json +import sys +import tempfile +import unittest +from pathlib import Path + +ROOT = Path(__file__).resolve().parent.parent +sys.path.insert(0, str(ROOT)) + +from kbench.verifiers_import import ( + VerifiersImportError, + import_quality, + read_episodes, +) + + +def trace( + trace_id: str, + reward: float, + *, + agent: str = "agent", + trainable: bool = True, + answer: str = "done", +) -> dict: + return { + "id": trace_id, + "verifiers": {"version": "0.3.1"}, + "agent": {"name": agent, "trainable": trainable}, + "ok": True, + "rewards": {"task": {"score": reward, "weight": 1.0}}, + "metrics": {"strict": reward}, + "nodes": [{"message": {"role": "assistant", "content": answer}}], + "timing": {"agent": {"start": 10.0, "end": 12.0}}, + } + + +def episode(episode_id: str, traces: list[dict], task_key: str = "task-1") -> dict: + return { + "id": episode_id, + "env": {"id": "example-tool-loop-v1"}, + "task": { + "type": "OfficeJobTask", + "key": task_key, + "data": {"private_prompt": "must never enter the score card"}, + }, + "ok": True, + "errors": [], + "traces": traces, + } + + +class VerifiersImportTests(unittest.TestCase): + def write(self, root: Path, rows: list[dict]) -> Path: + path = root / "traces.jsonl" + path.write_text("".join(json.dumps(row) + "\n" for row in rows)) + return path + + def test_one_episode_becomes_one_existing_sample_outcome(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = self.write(Path(tmp), [episode("ep-1", [trace("tr-1", 1.0)])]) + result = import_quality(path, "example-tool-loop") + self.assertEqual(result.n_samples, 1) + self.assertEqual(result.metrics["accuracy"], 1.0) + self.assertEqual(result.samples[0].sample_id, "ep-1") + self.assertEqual(result.samples[0].excerpt, "done") + self.assertNotIn("private_prompt", json.dumps(result.samples[0].metadata)) + self.assertEqual(result.dataset_version, "verifiers:0.3.1") + + def test_multi_agent_episode_ignores_non_trainable_user_simulator(self) -> None: + rows = [ + episode( + "ep-1", + [ + trace("assistant", 0.75, agent="assistant"), + trace("user", 0.0, agent="user", trainable=False), + ], + ) + ] + with tempfile.TemporaryDirectory() as tmp: + result = import_quality(self.write(Path(tmp), rows), "user-sim", pass_threshold=0.5) + self.assertEqual(result.samples[0].score, 0.75) + self.assertTrue(result.samples[0].passed) + self.assertEqual(result.samples[0].metadata["agents"], ["assistant"]) + + def test_named_metric_can_be_the_score_source(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = self.write(Path(tmp), [episode("ep-1", [trace("tr-1", 0.25)])]) + result = import_quality(path, "strict", pass_threshold=0.2, primary_metric="strict") + self.assertEqual(result.samples[0].score, 0.25) + self.assertEqual(result.samples[0].metadata["score_source"], "metric:strict") + + def test_fingerprint_uses_task_identity_not_line_order(self) -> None: + rows = [ + episode("ep-1", [trace("tr-1", 1.0)], "a"), + episode("ep-2", [trace("tr-2", 0.0)], "b"), + ] + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + a = import_quality(self.write(root, rows), "t").dataset_fingerprint + b = import_quality(self.write(root, list(reversed(rows))), "t").dataset_fingerprint + self.assertEqual(a, b) + + def test_duplicate_episode_ids_are_refused(self) -> None: + row = episode("same", [trace("tr-1", 1.0)]) + with tempfile.TemporaryDirectory() as tmp: + path = self.write(Path(tmp), [row, row]) + with self.assertRaises(VerifiersImportError): + read_episodes(path) + + def test_directory_resolves_the_upstream_traces_filename(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + self.write(root, [episode("ep-1", [trace("tr-1", 1.0)])]) + self.assertEqual(len(read_episodes(root)), 1) + + +if __name__ == "__main__": + unittest.main()