feat(v1): episode-level derived aggregates and typed EnvInfo

verifiers/v1/cli/dashboard/eval.py

@@ -393,18 +393,19 @@ def _breakdown(scored: list[Trace], done: list[Trace]) -> Table | None:
     phase_count: dict[str, int] = {}
     model_secs = harness_secs = 0.0
     for trace in done:
-        prompt, completion, cached, reasoning, _ = _tokens(trace)
-        total_in += prompt
-        total_out += completion
-        if cached is not None:
-            total_cached += cached
-            have_cached = True
-        if reasoning is not None:
-            total_reasoning += reasoning
-            have_reasoning = True
-        if trace.usage is not None and trace.usage.cost is not None:
-            total_cost += trace.usage.cost
-            have_cost = True
+        total_in += trace.num_input_tokens
+        total_out += trace.num_output_tokens
+        usage = trace.usage
+        if usage is not None:
+            if usage.cached_input_tokens is not None:
+                total_cached += usage.cached_input_tokens
+                have_cached = True
+            if usage.reasoning_tokens is not None:
+                total_reasoning += usage.reasoning_tokens
+                have_reasoning = True
+            if usage.cost is not None:
+                total_cost += usage.cost
+                have_cost = True
         # Judge / auxiliary scoring calls (off the message graph) shown separately from the agent's.
         judge = Usage.aggregate(trace.extra_usage)
         if judge is not None:
@@ -464,26 +465,6 @@ def _breakdown(scored: list[Trace], done: list[Trace]) -> Table | None:
     return grid if grid.row_count else None
 
 
-def _tokens(trace: Trace) -> tuple[int, int, int | None, int | None, int]:
-    """Input/output tokens summed across all branches: per branch, output is every assistant
-    (completion) token generated across its turns and input is the fed-in tokens counted once
-    (system + user + tool) — the final sequence minus everything the model generated. A rollout
-    yields one training sample per branch (a linear trace is a single branch; compaction and
-    subagents add more), so the totals sum them — matching `Trace.num_input_tokens` /
-    `Trace.num_output_tokens`, whose sum is `num_total_tokens`.
-
-    Both counts come from provider-reported usage. Returns the branch count from the same derived
-    view so each dashboard tick materializes it once."""
-    usage = trace.usage
-    cached = usage.cached_input_tokens if usage else None
-    reasoning = usage.reasoning_tokens if usage else None
-    branches = trace.branches
-    nbranches = len(branches)
-    prompt = sum(b.num_input_tokens for b in branches)
-    completion = sum(b.num_output_tokens for b in branches)
-    return prompt, completion, cached, reasoning, nbranches
-
-
 def _stage(trace: Trace) -> str:
     """The stage a live (not-yet-done) rollout is in, derived from its trace's timing
     spans — the engine opens and closes each span exactly at the stage transitions, so
@@ -551,7 +532,9 @@ def Rows(groups: list[list[RunSlot]], now: float, runtime_type: str) -> Table:
             base = f"name={task.name[:32]}" if task.name else f"idx={task.idx}"
             if not slot.traces:
                 if slot.done:  # the env's rollout() itself failed before any trace
-                    error = slot.episode.error if slot.episode is not None else None
+                    error = (
+                        slot.episode.last_error if slot.episode is not None else None
+                    )
                     group_rows.append(
                         (
                             "error",
@@ -612,8 +595,12 @@ def Rows(groups: list[list[RunSlot]], now: float, runtime_type: str) -> Table:
                     )
                     or now
                 )
-                prompt, completion, cached, reasoning, nbranches = _tokens(t)
-                cost = t.usage.cost if t.usage else None
+                prompt, completion = t.num_input_tokens, t.num_output_tokens
+                nbranches = t.num_branches
+                usage = t.usage
+                cached = usage.cached_input_tokens if usage else None
+                reasoning = usage.reasoning_tokens if usage else None
+                cost = usage.cost if usage else None
                 tokens = ""
                 if prompt or completion:
                     tokens = f"{format_count(prompt)}/{format_count(completion)} tokens"

verifiers/v1/cli/replay.py

@@ -84,7 +84,7 @@ async def run_replay(config: ReplayConfig, source: Path, out: Path) -> list[Trac
     episodes = read_episodes(
         source, Trace[WireTaskData, state_cls(task_cls), WireAgentConfig]
     )
-    sourced = [(trace, e.env) for e in episodes for trace in e.traces]
+    sourced = [(trace, e.env.id) for e in episodes for trace in e.traces]
     if config.num_traces is not None:
         sourced = sourced[: config.num_traces]
     traces = [trace for trace, _ in sourced]

verifiers/v1/env.py

@@ -20,7 +20,7 @@
     _declared_agent_configs,
     default_agent_harness,
 )
-from verifiers.v1.episode import Episode
+from verifiers.v1.episode import EnvInfo, Episode
 from verifiers.v1.errors import EnvError, boundary
 from verifiers.v1.harness import Harness, HarnessConfig
 from verifiers.v1.interception import (
@@ -257,7 +257,7 @@ async def run_episode(
         completed subset, its exception on the episode's `errors`. `on_trace` observes
         each agent-run's trace at mint; `on_discard` its abandonment (a per-agent
         retry mints a replacement)."""
-        episode = Episode(env=self.config.env_id)
+        episode = Episode(env=EnvInfo(id=self.config.env_id))
         agents = self._episode_agents(ctx, episode.traces, on_trace, on_discard)
         try:
             async with asyncio.timeout(self.config.timeout.episode):

verifiers/v1/episode.py

@@ -3,51 +3,79 @@
 import uuid
 from typing import Generic
 
-from pydantic import Field
+from pydantic import BaseModel, Field
 
 from verifiers.v1.configs.agent import WireAgentConfig
 from verifiers.v1.state import State, StateT
 from verifiers.v1.task import DataT, WireTaskData
 from verifiers.v1.trace import AgentConfigT, Error, Trace
-from verifiers.v1.types import StrictBaseModel
+from verifiers.v1.types import StrictBaseModel, Usage
 
 
-class Episode(StrictBaseModel, Generic[DataT, StateT, AgentConfigT]):
-    """One run of a task, whole: its identity and standing (`id`, `env`, `errors`)
-    next to its flat `traces` — the object `finalize()` receives, the engine
-    returns, and the durability envelope: one episode is one `traces.jsonl` line
-    and one serve reply, so it persists and arrives whole or not at all — a torn
-    line is the whole episode owed again, and a failure before any trace minted
-    still leaves its errors here. Episode standing lives ONLY here (zero
-    redundancy on the traces); per-trace facts (`agent`, per-trace errors) stay
-    on the traces, which remain the atomic unit.
+class EnvInfo(StrictBaseModel):
+    """The env that ran the episode, self-describing without the run's config."""
 
-    `errors` are failures not attributable to any one trace (the env's
-    `run`/`finalize` hooks, plus prior attempts' when retried).
+    id: str = ""
+    """`EnvConfig.env_id`, e.g. `agentic-judge+gsm8k-v1`."""
 
-    The type parameters serve the wire loaders: `WireEpisode` reads any taskset's
-    episodes without importing the taskset."""
+
+class Episode(BaseModel, Generic[DataT, StateT, AgentConfigT]):
+    """The artifact Env.run produces. Contains multiple agents' traces."""
 
     id: str = Field(default_factory=lambda: uuid.uuid4().hex)
-    env: str = ""
-    """The env that ran the episode (`EnvConfig.env_id`, e.g.
-    `agentic-judge+gsm8k-v1`)."""
+
+    env: EnvInfo = Field(default_factory=EnvInfo)
+    """The env that produced this episode."""
     ok: bool = False
-    """THE success sentinel — the resume unit's keep-verdict, stamped by the
-    engine when the final attempt's hooks and every trace concluded clean.
-    Distinct from `errors` emptiness: a retried-and-recovered episode is `ok`
-    and still keeps its earlier attempts' errors."""
+    """Whether the episode completed successfully."""
     errors: list[Error] = Field(default_factory=list)
+    """Every error captured across attempts, oldest to newest."""
     traces: list[Trace[DataT, StateT, AgentConfigT]] = Field(default_factory=list)
+    """Every agent's trace, in completion order."""
 
     @property
-    def error(self) -> Error | None:
+    def last_error(self) -> Error | None:
+        """The last episode-level error captured across attempts."""
         return self.errors[-1] if self.errors else None
 
+    @property
+    def usage(self) -> Usage | None:
+        """Provider-reported usage summed across every trace's model calls;
+        judge/off-graph usage stays on the traces (`Trace.extra_usage`)."""
+        return Usage.aggregate(u for t in self.traces if (u := t.usage) is not None)
+
+    @property
+    def num_input_tokens(self) -> int:
+        """Fed-in tokens (system + user + tool), summed across traces."""
+        return sum(t.num_input_tokens for t in self.traces)
+
+    @property
+    def num_output_tokens(self) -> int:
+        """Model-generated tokens across all turns, summed across traces."""
+        return sum(t.num_output_tokens for t in self.traces)
+
+    @property
+    def num_total_tokens(self) -> int:
+        """Final sequence lengths per branch, summed across traces."""
+        return sum(t.num_total_tokens for t in self.traces)
+
+    @property
+    def num_turns(self) -> int:
+        """Sampled turns, summed across traces."""
+        return sum(t.num_turns for t in self.traces)
+
+    @property
+    def by_agent(self) -> dict[str, list[Trace[DataT, StateT, AgentConfigT]]]:
+        """Traces grouped by agent name (e.g. n solvers), in completion order."""
+        grouped: dict[str, list[Trace[DataT, StateT, AgentConfigT]]] = {}
+        for trace in self.traces:
+            grouped.setdefault(trace.agent.name, []).append(trace)
+        return grouped
+
     @classmethod
     def of(cls, trace: Trace, env: str = "") -> "Episode":
         """The single-agent record: one trace as its own episode."""
-        return cls(env=env, traces=[trace], ok=trace.ok)
+        return cls(env=EnvInfo(id=env), traces=[trace], ok=trace.ok)
 
 
 WireEpisode = Episode[WireTaskData, State, WireAgentConfig]

verifiers/v1/trace.py

@@ -7,7 +7,7 @@
 from typing import TYPE_CHECKING, Annotated, Any, Generic, Literal
 
 import numpy as np
-from pydantic import Field, PrivateAttr
+from pydantic import BaseModel, Field, PrivateAttr
 from renderers.base import MultiModalData
 from typing_extensions import TypeVar
 
@@ -298,7 +298,7 @@ def num_input_tokens(self) -> int:
         return self.num_total_tokens - self.num_output_tokens
 
 
-class Trace(StrictBaseModel, Generic[DataT, StateT, AgentConfigT]):
+class Trace(BaseModel, Generic[DataT, StateT, AgentConfigT]):
     version: int = TRACE_VERSION
     """The trace schema this trace serializes as."""
     id: str = Field(default_factory=lambda: uuid.uuid4().hex)