From 20d998a91c4a2387ece0b580c89e7b3c4e66cfd1 Mon Sep 17 00:00:00 2001 From: faresobeid <111092724+faresobeid@users.noreply.github.com> Date: Sat, 3 Oct 2026 21:39:08 +0000 Subject: [PATCH 1/4] feat(v1): by-handle payload segments on traces --- verifiers/v1/__init__.py | 4 ++++ verifiers/v1/clients/train.py | 4 ++++ verifiers/v1/graph.py | 43 +++++++++++++++++++++++++++++++++ verifiers/v1/serve/delta.py | 1 + verifiers/v1/trace.py | 27 +++++++++++++++++++-- verifiers/v1/types.py | 45 +++++++++++++++++++++++++++++++++++ 6 files changed, 122 insertions(+), 2 deletions(-) diff --git a/verifiers/v1/__init__.py b/verifiers/v1/__init__.py index d669c24f34..19f7577a40 100644 --- a/verifiers/v1/__init__.py +++ b/verifiers/v1/__init__.py @@ -127,6 +127,8 @@ Message, MessageContent, Messages, + NodeSegment, + PayloadSegment, Request, Response, Sampling, @@ -259,6 +261,8 @@ "Branch", "TurnTokens", "SamplingMask", + "PayloadSegment", + "NodeSegment", "Timing", "TimeSpan", "TimeSplit", diff --git a/verifiers/v1/clients/train.py b/verifiers/v1/clients/train.py index 60275cb16c..3fb5e282c2 100644 --- a/verifiers/v1/clients/train.py +++ b/verifiers/v1/clients/train.py @@ -23,6 +23,7 @@ from verifiers.v1.types import ( AssistantMessage, FinishReason, + PayloadSegment, Response, SamplingConfig, SamplingMask, @@ -181,6 +182,9 @@ def response_from_generate( sampling_mask=SamplingMask.from_sampling_mask(mask) if (mask := result.get("sampling_mask")) else None, + payload=[PayloadSegment(**segment) for segment in segments] + if (segments := result.get("payload")) + else None, ), ) diff --git a/verifiers/v1/graph.py b/verifiers/v1/graph.py index 270f087d3e..99348775be 100644 --- a/verifiers/v1/graph.py +++ b/verifiers/v1/graph.py @@ -43,6 +43,8 @@ from verifiers.v1.types import ( AssistantMessage, Message, + NodeSegment, + PayloadSegment, Response, SamplingMask, TextContentPart, @@ -692,6 +694,43 @@ def _attribute_routed_experts( off = end +def _attribute_payload( + trace: Trace, + prefix_node_ids: list[int], + new_node_ids: list[int], + path_len: int, + segments: list[PayloadSegment] | None, +) -> None: + """Append this turn's by-handle rows to `Trace.payload`, node by node. The request's + positions tile `[path_len:]` over the nodes created this turn. The one position before, + the prefix's final token, belongs to the prefix's last node when that node was sampled: + the previous turn never forwarded its final token, so this prefill is the first to + produce its row. Rows of other reused prefix positions (a turn that re-forwarded its + history) are dropped. Rows are never rewritten, only appended.""" + if not segments: + return + spans: list[ + tuple[int, int, int, int] + ] = [] # node, request lo, request hi, node start + last = next( + (nid for nid in reversed(prefix_node_ids) if trace.nodes[nid].token_ids), None + ) + if last is not None and trace.nodes[last].sampled: + spans.append( + (last, path_len - 1, path_len, path_len - len(trace.nodes[last].token_ids)) + ) + start = path_len + for nid in new_node_ids: + end = start + len(trace.nodes[nid].token_ids) + spans.append((nid, start, end, start)) + start = end + for segment in segments: + for nid, lo, hi, node_start in spans: + part = segment.clip(lo, hi, shift=-node_start) + if part is not None: + trace.payload.append(NodeSegment(node=nid, segment=part)) + + def _attribute_sampling_mask( trace: Trace, assistant_id: int, payload: SamplingMask | None ) -> None: @@ -971,6 +1010,10 @@ def _commit_turn(turn: PendingTurn, response: Response) -> int: trace, assistant_id, tokens.sampling_mask if tokens else None ) + _attribute_payload( + trace, prefix, new_node_ids, path_len, tokens.payload if tokens else None + ) + return assistant_id diff --git a/verifiers/v1/serve/delta.py b/verifiers/v1/serve/delta.py index e41bad72dd..a629d6ae11 100644 --- a/verifiers/v1/serve/delta.py +++ b/verifiers/v1/serve/delta.py @@ -46,6 +46,7 @@ LIST_FIELDS = ( "nodes", "calls", + "payload", "errors", "extra_usage", "request_rewrites", diff --git a/verifiers/v1/trace.py b/verifiers/v1/trace.py index c3b4f6c58d..b24457227f 100644 --- a/verifiers/v1/trace.py +++ b/verifiers/v1/trace.py @@ -27,6 +27,8 @@ FinishReason, Message, Messages, + NodeSegment, + PayloadSegment, Sampling, SamplingMask, Tool, @@ -45,9 +47,11 @@ "routed_experts", "sampling_mask", } - } + }, + "payload": True, } -"""Raw tensor fields kept on the msgpack wire but excluded from disk serialization.""" +"""Raw tensor fields and short-lived payload handles, kept on the msgpack wire but excluded +from disk serialization.""" class TimeSpan(BaseModel): @@ -199,6 +203,9 @@ class Branch(BaseModel): """Whether this physical path contributes a training sample.""" mm_token_type_id_map: dict[int, int] = Field(default_factory=dict) """The trace's `mm_token_type_id_map`, carried so `mm_token_type_ids` is self-contained.""" + payload: list[PayloadSegment] = Field(default_factory=list) + """The trace's payload segments on this path, positions relative to `token_ids`, in + the order they were appended (a later segment wins where two overlap).""" @property def messages(self) -> Messages: @@ -395,6 +402,10 @@ class Trace(BaseModel, Generic[DataT, StateT, AgentConfigT]): """The message graph, including physical and semantic parent links.""" calls: list[ModelCall] = Field(default_factory=list) """Every model call; automatically recorded at intercept time + linked into `nodes`.""" + payload: list[NodeSegment] = Field(default_factory=list) + """By-handle per-token side arrays (router-replay ids, sampling masks), append-only: + each turn's `commit` attributes the inference server's segments to the nodes they cover. + Handles point into a short-lived shared directory; kept off disk records.""" mm_token_type_id_map: dict[int, int] = Field(default_factory=dict) """Special-token id -> modality marker (1 = image placeholder, 2 = video placeholder) from the renderer that tokenized this trace, stamped at turn commit. Applied to a @@ -534,6 +545,11 @@ def branches(self) -> list[Branch]: path.append(nid) nid = self.nodes[nid].parent path.reverse() + node_starts: dict[int, int] = {} + offset = 0 + for n in path: + node_starts[n] = offset + offset += len(self.nodes[n].token_ids) is_compaction_attempt = any( link.type == "compaction_attempt" for link in self.nodes[leaf].semantic_parents @@ -548,6 +564,13 @@ def branches(self) -> list[Branch]: or leaf in accepted_compaction_attempts ), mm_token_type_id_map=self.mm_token_type_id_map, + payload=[ + entry.segment.model_copy( + update={"pos": entry.segment.pos + node_starts[entry.node]} + ) + for entry in self.payload + if entry.node in node_starts + ], ) ) return branches diff --git a/verifiers/v1/types.py b/verifiers/v1/types.py index 79a01de60a..01a594f41a 100644 --- a/verifiers/v1/types.py +++ b/verifiers/v1/types.py @@ -222,6 +222,48 @@ def from_sampling_mask(cls, sampling_mask: list[list[int]]) -> "SamplingMask": return cls(ids=ids, counts=counts) +class PayloadSegment(BaseModel): + """`rows` consecutive rows of one per-token side array (`field`, e.g. `routed_experts`, + `sampling_mask`) that the inference server wrote to `file` at byte `offset`, as a + C-contiguous `[rows, *shape]` `dtype` array. `pos` is the token position of the first + row in whatever sequence holds the segment: the request on `TurnTokens`, the node on + `Trace.payload`, the branch on `Branch.payload`. Only the handle travels; the trainer + reads the rows.""" + + field: str + file: str + offset: int + pos: int + rows: int + dtype: str + shape: list[int] + + @property + def end(self) -> int: + return self.pos + self.rows + + def clip(self, lo: int, hi: int, shift: int = 0) -> "PayloadSegment | None": + """The rows inside positions `[lo, hi)`, moved by `shift`; None when empty.""" + start, end = max(self.pos, lo), min(self.end, hi) + if start >= end: + return None + row_bytes = np.dtype(self.dtype).itemsize * int(np.prod(self.shape)) + return self.model_copy( + update={ + "offset": self.offset + (start - self.pos) * row_bytes, + "pos": start + shift, + "rows": end - start, + } + ) + + +class NodeSegment(BaseModel): + """A payload segment attributed to `Trace.nodes[node]`, positions node-local.""" + + node: int + segment: PayloadSegment + + class TurnTokens(BaseModel): """Training tokens from renderer tokenization or provider-returned token IDs.""" @@ -257,6 +299,9 @@ class TurnTokens(BaseModel): # Transient carrier (excluded): per-completion-token sampling masks, # attributed to the assistant node by the turn's `commit`, then dropped. sampling_mask: SamplingMask | None = Field(default=None, exclude=True) + # Transient carrier (excluded): by-handle side arrays from `generate`, request-absolute + # positions, attributed per node onto `Trace.payload` by the turn's `commit`. + payload: list[PayloadSegment] | None = Field(default=None, exclude=True) class Response(BaseModel): From 62b0e32a2ec528d7696b47774bc532aed51507f3 Mon Sep 17 00:00:00 2001 From: faresobeid <111092724+faresobeid@users.noreply.github.com> Date: Tue, 6 Oct 2026 17:17:10 +0000 Subject: [PATCH 2/4] test(v1): cover by-handle payload attribution and branch rebase --- tests/v1/test_graph.py | 57 ++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 57 insertions(+) diff --git a/tests/v1/test_graph.py b/tests/v1/test_graph.py index a4d8a25779..a139ae7623 100644 --- a/tests/v1/test_graph.py +++ b/tests/v1/test_graph.py @@ -649,3 +649,60 @@ def test_prompt_supplied_assistant_messages_are_not_sampled_turns(): assert [n.sampled for n in trace.nodes] == [False, False, False, True] assert trace.num_turns == 1 assert trace.assistant_messages == [response] + + +def test_payload_segments_attributed_per_node_and_rebased_per_branch(): + """By-handle rows tile the nodes a turn creates. A bridged turn's first row belongs to the + prior sampled turn's final token; a reused non-sampled prefix row is dropped. Each branch + rebases its nodes' segments onto its own token positions.""" + trace = vf.Trace( + agent=vf.AgentInfo(config=vf.AgentConfig()), + task=vf.TraceTask(type="Task", data=vf.TaskData(idx=0, prompt="x")), + ) + user = vf.UserMessage(content="u1") + + def commit(messages, content, prompt_ids, spans, pos, rows): + segment = vf.PayloadSegment( + field="routed_experts", + file="f", + offset=pos * 2, + pos=pos, + rows=rows, + dtype="uint8", + shape=[2], + ) + graph.prepare_turn(trace, messages).commit( + vf.Response( + id=content, + created=0, + model="t", + message=vf.AssistantMessage(content=content), + finish_reason="stop", + tokens=TurnTokens( + prompt_ids=prompt_ids, + completion_ids=[40, 41], + message_spans=spans, + payload=[segment], + ), + ) + ) + + def rows(branch): + return [(s.pos, s.rows, s.offset) for s in branch.payload] + + commit([user], "a1", [10, 11, 12], [(0, 3)], 0, 4) + # Bridged: request row 4 is a1's unforwarded final token. + commit( + [user, vf.AssistantMessage(content="a1"), vf.UserMessage(content="u2")], + "a2", + [10, 11, 12, 40, 41, 30, 31], + [(0, 3), None, (5, 7)], + 4, + 4, + ) + # Fork off the shared user prompt: re-forwarded rows 0-2 are dropped. + commit([user], "b1", [10, 11, 12], [(0, 3)], 0, 4) + + a2, b1 = (b for b in trace.branches if len(b.token_ids) in (9, 5)) + assert rows(a2) == [(0, 3, 0), (3, 1, 6), (4, 1, 8), (5, 2, 10), (7, 1, 14)] + assert rows(b1) == [(0, 3, 0), (3, 1, 6)] From 0846cb5210fd1d9a1351921d73f7e0ba8efcc4d6 Mon Sep 17 00:00:00 2001 From: faresobeid <111092724+faresobeid@users.noreply.github.com> Date: Fri, 16 Oct 2026 17:29:29 +0000 Subject: [PATCH 3/4] style(v1): keep the span-tuple comment off the annotation --- uv.lock | 3 +-- verifiers/v1/graph.py | 5 ++--- 2 files changed, 3 insertions(+), 5 deletions(-) diff --git a/uv.lock b/uv.lock index c24f3388e8..d261864181 100644 --- a/uv.lock +++ b/uv.lock @@ -1,5 +1,5 @@ version = 1 -revision = 5 +revision = 4 requires-python = ">=3.11, <3.15" resolution-markers = [ "python_full_version >= '3.14' and sys_platform == 'win32'", @@ -5327,7 +5327,6 @@ wheels = [ [[package]] name = "verifiers" source = { editable = "." } -default-groups = ["dev", "examples"] dependencies = [ { name = "aiohttp" }, { name = "anthropic" }, diff --git a/verifiers/v1/graph.py b/verifiers/v1/graph.py index 99348775be..5c392f45ed 100644 --- a/verifiers/v1/graph.py +++ b/verifiers/v1/graph.py @@ -709,9 +709,8 @@ def _attribute_payload( history) are dropped. Rows are never rewritten, only appended.""" if not segments: return - spans: list[ - tuple[int, int, int, int] - ] = [] # node, request lo, request hi, node start + # (node, request lo, request hi, node start) + spans: list[tuple[int, int, int, int]] = [] last = next( (nid for nid in reversed(prefix_node_ids) if trace.nodes[nid].token_ids), None ) From 6e235e77067495cfd8cb2e3c200f102d49900687 Mon Sep 17 00:00:00 2001 From: faresobeid <111092724+faresobeid@users.noreply.github.com> Date: Sat, 24 Oct 2026 18:31:04 +0000 Subject: [PATCH 4/4] fix(v1): check node token count with len() now that token_ids is an array --- verifiers/v1/graph.py | 3 ++- verifiers/v1/types.py | 2 -- 2 files changed, 2 insertions(+), 3 deletions(-) diff --git a/verifiers/v1/graph.py b/verifiers/v1/graph.py index d37acee796..402fc6be45 100644 --- a/verifiers/v1/graph.py +++ b/verifiers/v1/graph.py @@ -798,7 +798,8 @@ def _attribute_payload( # (node, request lo, request hi, node start) spans: list[tuple[int, int, int, int]] = [] last = next( - (nid for nid in reversed(prefix_node_ids) if trace.nodes[nid].token_ids), None + (nid for nid in reversed(prefix_node_ids) if len(trace.nodes[nid].token_ids)), + None, ) if last is not None and trace.nodes[last].sampled: spans.append( diff --git a/verifiers/v1/types.py b/verifiers/v1/types.py index ab94b624a6..c4f19882d6 100644 --- a/verifiers/v1/types.py +++ b/verifiers/v1/types.py @@ -324,8 +324,6 @@ def clip(self, lo: int, hi: int, shift: int = 0) -> "PayloadSegment | None": start, end = max(self.pos, lo), min(self.end, hi) if start >= end: return None - import numpy as np - row_bytes = np.dtype(self.dtype).itemsize * int(np.prod(self.shape)) return self.model_copy( update={