"""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", ]