feat: delta-native message-graph trajectory

verifiers/v1/README.md

@@ -16,7 +16,10 @@ abstractions and on-disk output. Everything is pydantic-typed; `import verifiers
 - **Minimal & pythonic** — the high-level abstractions without the implementation bulk;
   plain classes + decorators (`@vf.reward` / `@vf.metric` / ...).
 - **Training-ready traces** — exact token ids + logprobs straight from an agentic rollout
-  (renderer client), with branching recovered for compaction / subagents.
+  (renderer client); one training sample per branch, recovered for compaction / subagents.
+- **Delta-native trace graph** — each message is stored once as a node linked to its
+  predecessor, so a trace's size is linear in turns, not quadratic; branches fall out of
+  walking the graph, and a training sample is a cheap concat of node tokens along a path.
 - **Hub-native + v0-compatible** — ids install on demand from the Environments Hub, and
   classic v0 envs run through the same CLIs via a bridge.
 
@@ -145,9 +148,9 @@ uv run eval code-golf-v1 -n 1 -r 2  # group rewards: a @vf.group_reward scores N
 
 A rollout isn't always linear. The `compact` harness rewrites its context every turn — a
 fresh `[system, user]` carrying its running notes plus the last tool output — so each turn
-is its own *branch*. `branching` recovers them from the flat trajectory and
-`trace.branches` / `num_branches` expose it (a linear harness is one branch; the compact
-harness is one per turn — it also handles subagents):
+is its own *branch*. Branches fall out of the message graph — each leaf's root→leaf path is
+one branch, exposed by `trace.branches` / `num_branches` (a linear harness is one branch;
+the compact harness is one per turn — it also handles subagents):
 
 ```bash
 uv run eval wiki-search-v1 -n 1 --harness.id compact  # fresh prompt each turn → num_branches == turns
@@ -164,10 +167,11 @@ uv run eval gsm8k-v1 -n 1 --client.type renderers \  # renderers: client-side to
   --client.base-url http://localhost:8000/v1          # token-in/out traces (needs a vLLM engine)
 ```
 
-With `renderers`, each `trace.trajectory[i].tokens` carries the exact `prompt_ids` /
-`completion_ids` / `completion_logprobs` the engine saw — training-ready token data
-straight from an agentic rollout, with zero agent changes. (When the engine returns ids on
-the response itself, the openai client picks them up too — no renderer required.)
+With `renderers`, each graph node carries the exact tokens the engine saw — `token_ids`
+plus a per-token trainable `mask` and `logprobs` — so concatenating a branch's nodes is a
+ready training sample, straight from an agentic rollout with zero agent changes. (When the
+engine returns ids on the response itself, the openai client picks them up too — no
+renderer required.)
 
 ### Limits & retries
 

verifiers/v1/__init__.py

@@ -43,14 +43,13 @@
 from verifiers.v1.task import Resources, Task, WireTask
 from verifiers.v1.taskset import Taskset, TasksetConfig, ToolsConfig
 from verifiers.v1.tools import Tools, run_mcp_server
+from verifiers.v1.graph import MessageNode
 from verifiers.v1.trace import (
     Branch,
     Error,
     TimeSpan,
     Timing,
     Trace,
-    Turn,
-    TurnTokens,
 )
 from verifiers.v1.types import (
     AssistantMessage,
@@ -62,6 +61,7 @@
     SystemMessage,
     Tool,
     ToolCall,
+    TurnTokens,
     ToolMessage,
     Usage,
     UserMessage,
@@ -90,7 +90,7 @@
     "WireTask",
     "Resources",
     "Trace",
-    "Turn",
+    "MessageNode",
     "Branch",
     "TurnTokens",
     "Timing",

verifiers/v1/cli/dashboard.py

@@ -81,15 +81,10 @@ def _tokens(trace: Trace) -> str:
     final context the model saw. (Output can exceed the final context — reasoning tokens
     count toward completions but aren't re-fed — so it's not derived by subtraction.)"""
     branches = trace.branches
-    if not branches or not branches[0].turns:
+    if not branches or not branches[0].nodes:
         return ""
-    usages = [
-        t.response.usage for t in branches[0].turns if t.response.usage is not None
-    ]
-    if not usages:
-        return ""
-    output = sum(u.completion_tokens for u in usages)
-    return f"{format_count(usages[-1].prompt_tokens)}/{format_count(output)} tokens"
+    b = branches[0]
+    return f"{format_count(b.prompt_len)}/{format_count(b.completion_len)} tokens"
 
 
 def _groups(rollouts: list[Rollout]) -> list[list[Rollout]]:

verifiers/v1/clients/renderer.py

@@ -62,6 +62,12 @@ def response_from_generate(result: dict, model: str) -> Response:
     ] or None
     prompt_ids = result.get("prompt_ids") or []
     completion_ids = result.get("completion_ids") or []
+    # Per-message token spans (the renderer's attribution) let the trace graph store each
+    # message's tokens once; carried transiently on TurnTokens and consumed by `graph.add_turn`.
+    attribution = result.get("prompt_attribution")
+    message_spans = (
+        attribution.message_token_spans() if attribution is not None else None
+    )
     return Response(
         id=result.get("request_id", ""),
         created=0,
@@ -79,6 +85,7 @@ def response_from_generate(result: dict, model: str) -> Response:
             prompt_ids=prompt_ids,
             completion_ids=completion_ids,
             completion_logprobs=result.get("completion_logprobs") or [],
+            message_spans=message_spans,
         ),
     )
 

verifiers/v1/graph.py

@@ -0,0 +1,179 @@
+"""Message-graph trajectory: store each message once, recover branches by walking.
+
+A rollout is a graph of `MessageNode`s — one per distinct message, each linked to its
+predecessor. The conversation is a path from a root to a leaf; branches (compaction,
+subagents) are simply multiple leaves, so branching falls out of the walk. Each node stores
+only the tokens it *adds* to the cumulative sequence, keeping size linear in turns and
+making a branch's training sample a cheap concat of node `token_ids`/`mask`/`logprobs` along
+its path.
+
+Token attribution (renderer client): the renderer reports, per prompt, each message's token
+span (`RenderedTokens.message_token_spans()`, carried on `TurnTokens.message_spans`). A new
+input message's node gets its span plus the leading template scaffold since the previous
+message; the trailing scaffold (the generation prompt) goes on the assistant node, prefixed
+to its sampled completion. By construction `concat(node.token_ids along a path)` reproduces
+the exact `prompt_ids + completion_ids` the model saw.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from typing import TYPE_CHECKING
+
+from pydantic import Field
+
+from verifiers.v1.types import (
+    AssistantMessage,
+    FinishReason,
+    Message,
+    Response,
+    StrictBaseModel,
+    ToolMessage,
+)
+
+if TYPE_CHECKING:
+    from verifiers.v1.trace import Branch, Trace
+
+
+class MessageNode(StrictBaseModel):
+    """One message in the graph: a message plus the tokens it adds to the cumulative
+    sequence. Concatenating a root→leaf path's nodes reconstructs that branch's full token
+    sequence; the mask/logprobs make it a training sample."""
+
+    parent: int | None = None
+    """Index into `Trace.nodes` of the predecessor message; None for a root."""
+    message: Message
+    """The message this node carries (system / user / assistant / tool)."""
+    token_ids: list[int] = Field(default_factory=list)
+    """This message's delta contribution to the cumulative token sequence: its leading
+    template scaffold + its own tokens — for an assistant, the generation-prompt scaffold
+    followed by the sampled completion. Concatenated along a path, these reproduce the exact
+    `prompt_ids + completion_ids` the model saw."""
+    mask: list[bool] = Field(default_factory=list)
+    """Per-token, parallel to `token_ids`: True for trainable, model-sampled tokens (only an
+    assistant node's completion span); False for template scaffold and every input-message
+    token."""
+    logprobs: list[float] = Field(default_factory=list)
+    """Sampling logprobs for the sampled tokens — length equals the number of True entries in
+    `mask`; empty for input messages."""
+    finish_reason: FinishReason = None
+    """The response's finish reason (assistant nodes only) — kept for truncation detection."""
+
+
+def message_hash(message: Message) -> str:
+    """Stable content hash on the fields that round-trip through a prompt — role, content
+    (None and "" equal), assistant tool calls, tool call id; `reasoning_content` ignored.
+    Two messages hash equal iff they're the same conversational message, so a re-stated
+    prefix message dedups to one node. The dedup key for sharing a prefix across
+    turns/branches; salt-free so it is identical across processes and after deserialization."""
+    parts: list[str] = [type(message).__name__, message.content or ""]
+    if isinstance(message, AssistantMessage):
+        for tc in message.tool_calls or []:
+            parts += [tc.id, tc.name, tc.arguments]
+    elif isinstance(message, ToolMessage):
+        parts.append(message.tool_call_id)
+    return hashlib.blake2b("\x00".join(parts).encode(), digest_size=16).hexdigest()
+
+
+def _head_index(trace: "Trace") -> dict[tuple[int | None, str], int]:
+    """`(parent, msg_hash) -> node_id`, rebuilt lazily from `nodes` after deserialization."""
+    if not trace._head_index and trace.nodes:
+        trace._head_index = {
+            (node.parent, message_hash(node.message)): nid
+            for nid, node in enumerate(trace.nodes)
+        }
+    return trace._head_index
+
+
+def add_turn(trace: "Trace", prompt: "list[Message]", response: Response) -> None:
+    """Insert one model turn (its prompt messages + its response) into the graph. Reuses any
+    existing prefix nodes (by `(parent, hash)`), creates a node per new message attributing
+    its tokens, and appends a fresh assistant node holding the generation-prompt scaffold +
+    the sampled completion.
+
+    Token attribution anchors new tokens to the cumulative *stored* length of the reused
+    prefix (`path_len`), not message spans — the previous assistant's closing scaffold lives
+    in its later input-form span but not its stored generation form, so anchoring on spans
+    would drop it. The new tokens (`prompt_ids[path_len:]`) are split among the new input
+    messages by span (leading template scaffold folds into the following message), and the
+    trailing generation prompt goes on the assistant node before its sampled completion. By
+    construction `concat(node.token_ids along the path) == prompt_ids + completion_ids`."""
+    tokens = response.tokens
+    prompt_ids = list(tokens.prompt_ids) if tokens else []
+    spans = tokens.message_spans if tokens else None
+    idx = _head_index(trace)
+
+    parent: int | None = None
+    path_len = 0  # cumulative stored token length of the reused prefix
+    # cursor: in prompt_ids, the end of the previous *new* message's tokens
+    cursor: int | None = None
+    for i, msg in enumerate(prompt):
+        key = (parent, message_hash(msg))
+        existing = idx.get(key)
+        if cursor is None and existing is not None:  # still extending the shared prefix
+            parent = existing
+            path_len += len(trace.nodes[existing].token_ids)
+            continue
+        start = path_len if cursor is None else cursor
+        span = spans[i] if spans and i < len(spans) else None
+        end = span[1] if span else start
+        node_tokens = prompt_ids[start:end]
+        trace.nodes.append(
+            MessageNode(
+                parent=parent,
+                message=msg,
+                token_ids=node_tokens,
+                mask=[False] * len(node_tokens),
+            )
+        )
+        parent = len(trace.nodes) - 1
+        idx[key] = parent
+        cursor = end
+
+    # Assistant node: trailing scaffold (the generation prompt) + the sampled completion.
+    comp_ids = list(tokens.completion_ids) if tokens else []
+    gen_start = path_len if cursor is None else cursor
+    gen_prompt = prompt_ids[gen_start:]
+    trace.nodes.append(
+        MessageNode(
+            parent=parent,
+            message=response.message,
+            token_ids=[*gen_prompt, *comp_ids],
+            mask=[False] * len(gen_prompt) + [True] * len(comp_ids),
+            logprobs=list(tokens.completion_logprobs) if tokens else [],
+            finish_reason=response.finish_reason,
+        )
+    )
+    # Register the assistant so the next turn's prompt (which restates it) reuses this node.
+    idx[(parent, message_hash(response.message))] = len(trace.nodes) - 1
+
+
+# --- walking the graph (views) ---------------------------------------------------------
+
+
+def _path_to(trace: "Trace", leaf: int) -> list[int]:
+    """Node ids from the root down to `leaf` (inclusive), in order."""
+    path: list[int] = []
+    nid: int | None = leaf
+    while nid is not None:
+        path.append(nid)
+        nid = trace.nodes[nid].parent
+    path.reverse()
+    return path
+
+
+def leaves(trace: "Trace") -> list[int]:
+    """Node ids that are no node's parent — one per branch (the last node of each)."""
+    has_child = {n.parent for n in trace.nodes if n.parent is not None}
+    return [i for i in range(len(trace.nodes)) if i not in has_child]
+
+
+def branches_from_nodes(trace: "Trace") -> list["Branch"]:
+    """Each leaf's root→leaf node path becomes a `Branch` — one per leaf (one branch when
+    linear, several under compaction or subagents)."""
+    from verifiers.v1.trace import Branch
+
+    return [
+        Branch(index=i, nodes=[trace.nodes[nid] for nid in _path_to(trace, leaf)])
+        for i, leaf in enumerate(leaves(trace))
+    ]