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

tests/v1/test_e2e.py

@@ -367,7 +367,7 @@ async def test_multi_agent_env(run_v1, tmp_path):
     # agent info (completion order — the gathered seats land in either order).
     (line,) = (tmp_path / "traces.jsonl").read_text().splitlines()
     row = json.loads(line)
-    assert row["env"] == "duet-v1"
+    assert row["env"] == {"id": "duet-v1"}
     by_name = {t["agent"]["name"]: t for t in row["traces"]}
     assert set(by_name) == {"a", "b"}
     assert by_name["a"]["agent"]["trainable"] is True

verifiers/v1/__init__.py

@@ -115,10 +115,10 @@
 from verifiers.v1.trace import (
     TRACE_VERSION,
     AgentInfo,
+    AgentSpan,
     Branch,
     Error,
     EvalRunInfo,
-    GenerationSpan,
     ModelCall,
     Reward,
     RunInfo,
@@ -144,7 +144,6 @@
     Response,
     Sampling,
     SamplingConfig,
-    StrictBaseModel,
     SystemMessage,
     TextContentPart,
     Tool,
@@ -177,7 +176,6 @@
     "Response",
     "Sampling",
     "SamplingConfig",
-    "StrictBaseModel",
     "SystemMessage",
     "TextContentPart",
     "Tool",
@@ -213,7 +211,7 @@
     "Timing",
     "TimeSpan",
     "TimeSplit",
-    "GenerationSpan",
+    "AgentSpan",
     "Error",
     # decorators
     "stop",

verifiers/v1/artifacts.py

@@ -8,9 +8,7 @@
 from pathlib import PurePosixPath
 from typing import TYPE_CHECKING
 
-from pydantic import Field
-
-from verifiers.v1.types import StrictBaseModel
+from pydantic import BaseModel, Field
 
 if TYPE_CHECKING:
     from verifiers.v1.runtimes import Runtime
@@ -25,7 +23,7 @@
 agent's image, so the repo is already there and only its output has to travel."""
 
 
-class Artifact(StrictBaseModel):
+class Artifact(BaseModel):
     """One path to restore at the same location in another runtime."""
 
     source: str

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:
@@ -413,13 +414,13 @@ def _breakdown(scored: list[Trace], done: list[Trace]) -> Table | None:
             if judge.cost is not None:
                 total_judge_cost += judge.cost
             have_judge = True
-        for phase in ("boot", "setup", "generation", "finalize", "scoring"):
+        for phase in ("boot", "setup", "agent", "finalize", "scoring"):
             span = getattr(trace.timing, phase)
             if span.end:  # phase was timed for this rollout
                 phase_secs[phase] = phase_secs.get(phase, 0.0) + span.duration
                 phase_count[phase] = phase_count.get(phase, 0) + 1
-        model_secs += trace.timing.generation.model.duration
-        harness_secs += trace.timing.generation.harness.duration
+        model_secs += trace.timing.agent.model.duration
+        harness_secs += trace.timing.agent.harness.duration
     if (
         total_in
         or total_out
@@ -448,12 +449,12 @@ def _breakdown(scored: list[Trace], done: list[Trace]) -> Table | None:
             usage.append(cost)
         grid.add_row("usage", "  ·  ".join(usage))
     time_segments = []
-    for phase in ("boot", "setup", "generation", "finalize", "scoring"):
+    for phase in ("boot", "setup", "agent", "finalize", "scoring"):
         count = phase_count.get(phase)
         if not count:
             continue
         segment = f"{phase} {format_time(phase_secs[phase] / count)}"
-        if phase == "generation":
+        if phase == "agent":
             segment += (
                 f" (model {format_time(model_secs / count)}"
                 f" + harness {format_time(harness_secs / count)})"
@@ -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
@@ -495,7 +476,7 @@ def _stage(trace: Trace) -> str:
     for stage, span in (
         ("scoring", trace.timing.scoring),
         ("finalize", trace.timing.finalize),
-        ("running", trace.timing.generation),
+        ("running", trace.timing.agent),
         ("setup", trace.timing.setup),
         ("boot", trace.timing.boot),
     ):
@@ -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",
@@ -602,7 +585,7 @@ def Rows(groups: list[list[RunSlot]], now: float, runtime_type: str) -> Table:
                 end = (
                     t.timing.scoring.end
                     or t.timing.finalize.end
-                    or t.timing.generation.end
+                    or t.timing.agent.end
                     # a rollout that errored in boot/setup has only that span's end — freeze there
                     # once done, else (still running) the timer would grow off `now` forever
                     or (
@@ -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/debug.py

@@ -123,16 +123,16 @@ def record_debug_error(
     action_timeout: float | None,
 ) -> None:
     now = time.time()
-    for span in (trace.timing.boot, trace.timing.setup, trace.timing.generation):
+    for span in (trace.timing.boot, trace.timing.setup, trace.timing.agent):
         if span.start and not span.end:
             span.end = now
-    in_action = bool(trace.timing.generation.start)
+    in_action = bool(trace.timing.agent.start)
     stage = (
         "debug action" if in_action else "setup" if trace.timing.setup.start else "boot"
     )
     timeout = action_timeout if in_action else setup_timeout
     error_start = (
-        trace.timing.generation.start
+        trace.timing.agent.start
         if in_action
         else trace.timing.setup.start or trace.timing.boot.start
     )
@@ -234,9 +234,9 @@ async def debug_task(task: Task, config: DebugConfig) -> tuple[Trace, bool]:
         await runtime.prepare_execution([])
         trace.timing.setup.end = time.time()
 
-        trace.timing.generation.start = time.time()
+        trace.timing.agent.start = time.time()
         debug.update(await run_action(runtime, config))
-        trace.timing.generation.end = time.time()
+        trace.timing.agent.end = time.time()
         if not debug.get("ok"):
             record_action_failure(trace, debug)
         trace.stop(str(debug["reason"]))
@@ -246,7 +246,7 @@ async def debug_task(task: Task, config: DebugConfig) -> tuple[Trace, bool]:
     except Exception as e:  # noqa: BLE001 - persist any framework failure on the trace
         record_debug_error(trace, debug, e, setup_timeout, config.timeout.total)
     finally:
-        trace.split_generation()
+        trace.split_agent_time()
         trace.info["debug"] = debug
         try:
             await runtime.stop()

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/envs/agentic_judge/env.py

@@ -19,10 +19,9 @@
 import tomllib
 from pathlib import Path
 
-from pydantic import Field, field_validator
+from pydantic import BaseModel, Field, field_validator
 
 import verifiers.v1 as vf
-from verifiers.v1.types import StrictBaseModel
 from verifiers.v1.utils.compile import validate_pairing
 
 VERDICT_FILE = "/tmp/verdict.json"
@@ -39,7 +38,7 @@
 {prompt}"""
 
 
-class Criterion(StrictBaseModel):
+class Criterion(BaseModel):
     """One rubric criterion — the plugged rubric judge's format, mirrored so the
     same `criteria` files grade both judges."""
 

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 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(BaseModel):
+    """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]