Option B: typed MessageNode.kind for replay resume points

environments/compact/compact/annotate.py

@@ -0,0 +1,42 @@
+"""Producer-side helpers: locate the compaction resume points a harness should tag.
+
+A compacting rollout splits into branches (one per context rewrite). Each compaction exposes
+two replay resume points:
+
+- ``compaction_after``  — the post-compaction branch start (the rewritten ``[system, user(notes)]``).
+  Resuming here, the model continues solving *from* the compaction message.
+- ``compaction_before`` — the leaf of the branch that compaction summarized (the prior turn's
+  response). Resuming here, the model is back in the pre-compaction context and its continuation
+  *writes* the compaction itself (then keeps solving).
+
+This module only *finds* the nodes; where the tag is stored is the A/B decision (Option A:
+``trace.info``; Option B: ``MessageNode.kind``) and lives at the harness write site.
+"""
+
+from __future__ import annotations
+
+from verifiers.v1 import graph
+from verifiers.v1.trace import Trace
+
+
+def compaction_after_nodes(trace: Trace) -> list[int]:
+    """Post-compaction branch starts: the first node of each forked branch. A node with >1
+    child is a fork point; ``children[0]`` is the original line, ``children[1:]`` are the
+    rewritten (post-compaction) branches. The compacting harness rewrites every turn, so each
+    is a compaction boundary."""
+    children: dict[int | None, list[int]] = {}
+    for nid, node in enumerate(trace.nodes):
+        children.setdefault(node.parent, []).append(nid)
+    starts: list[int] = []
+    for kids in children.values():
+        if len(kids) > 1:
+            starts.extend(kids[1:])
+    return starts
+
+
+def compaction_before_nodes(trace: Trace) -> list[int]:
+    """Pre-compaction points: every branch leaf except the final-answer branch. Each such leaf
+    is the turn whose output was summarized into the next branch's compaction message, so
+    resuming there puts the model right before it writes a compaction."""
+    leaves = sorted(graph.leaves(trace))
+    return leaves[:-1] if len(leaves) > 1 else []

environments/compact/compact/harness.py

@@ -16,6 +16,8 @@
 from verifiers.v1.runtimes import ProgramResult, Runtime
 from verifiers.v1.trace import Trace
 
+from compact.annotate import compaction_after_nodes, compaction_before_nodes
+
 PROGRAM_SOURCE = (Path(__file__).resolve().parent / "program.py").read_text()
 
 
@@ -51,4 +53,13 @@ async def launch(
                 {"mcpServers": {name: {"url": url} for name, url in mcp_urls.items()}}
             )
         program = await runtime.prepare_uv_script(PROGRAM_SOURCE, self.config.env)
-        return await runtime.run_program([*program, trace.task.prompt], env)
+        result = await runtime.run_program([*program, trace.task.prompt], env)
+
+        # Tag compaction resume points on the finished graph (Option B: typed MessageNode.kind).
+        # Two points per compaction: the post-compaction branch start (resume to continue from
+        # the compaction message) and the pre-compaction branch leaf (resume to regenerate it).
+        for nid in compaction_after_nodes(trace):
+            trace.nodes[nid].kind = "compaction_after"
+        for nid in compaction_before_nodes(trace):
+            trace.nodes[nid].kind = "compaction_before"
+        return result

environments/replay/pyproject.toml

@@ -0,0 +1,13 @@
+[project]
+name = "replay"
+version = "0.1.0"
+description = "replay — a replay-buffer taskset that resumes old rollouts from compaction points."
+requires-python = ">=3.10"
+dependencies = ["verifiers"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["replay"]

environments/replay/replay/__init__.py

@@ -0,0 +1,8 @@
+"""replay — resume old rollouts from tagged compaction points (a replay buffer).
+
+Option B: compaction tags are read from the typed ``MessageNode.kind`` field (see
+``replay/selector.py``)."""
+
+from replay.taskset import ReplayTask, ReplayTaskset, ReplayTasksetConfig
+
+__all__ = ["ReplayTaskset", "ReplayTasksetConfig", "ReplayTask"]

environments/replay/replay/selector.py

@@ -0,0 +1,45 @@
+"""Consumer-side resume-point selection for the replay buffer (Option B: ``MessageNode`` fields).
+
+Reads the compaction tags the producing harness stamped on ``MessageNode.kind`` and turns them
+into resume points the ReplayTaskset can sample. ``seed_messages`` builds the replay prefix (the
+root->node path); ``snapshot_ref_of`` returns the durable sandbox handle to restore (None until
+per-turn snapshot capture is wired).
+"""
+
+from __future__ import annotations
+
+from verifiers.v1.trace import Trace
+from verifiers.v1.types import Messages
+
+
+def get_tag(trace: Trace, node_id: int) -> str | None:
+    """Read a node's replay tag (Option B: from the typed ``MessageNode.kind`` field)."""
+    return trace.nodes[node_id].kind
+
+
+def snapshot_ref_of(trace: Trace, node_id: int) -> str | None:
+    """Durable sandbox snapshot ref for a node (Option B: from ``MessageNode.snapshot_ref``).
+    None when snapshotting was off/unsupported (the skeleton never captures one yet)."""
+    return trace.nodes[node_id].snapshot_ref
+
+
+def seed_messages(trace: Trace, node_id: int) -> Messages:
+    """The replay prefix: messages along root->node_id, in order."""
+    path: Messages = []
+    nid: int | None = node_id
+    while nid is not None:
+        path.append(trace.nodes[nid].message)
+        nid = trace.nodes[nid].parent
+    path.reverse()
+    return path
+
+
+def resume_points(trace: Trace, *, kinds: set[str]) -> list[dict]:
+    """Compaction resume points whose tag is in ``kinds``. Each: ``node`` id and ``kind``
+    ("compaction_before" | "compaction_after")."""
+    points: list[dict] = []
+    for nid in range(len(trace.nodes)):
+        tag = get_tag(trace, nid)
+        if tag in kinds:
+            points.append({"node": nid, "kind": tag})
+    return points

environments/replay/replay/taskset.py

@@ -0,0 +1,94 @@
+"""A minimal replay-buffer taskset (skeleton).
+
+Materializes replay tasks from a buffer of stored rollouts (offline mode): each compaction
+resume point in each stored trace becomes one task whose ``prompt`` is the replay prefix
+(root->resume node). Run it with the default harness (``--harness.id default``), which accepts
+a Messages prompt and continues the rollout from the prefix:
+
+- ``compaction_after``  -> the prefix ends at the compaction message; the model continues solving.
+- ``compaction_before`` -> the prefix is the pre-compaction context; the model first writes the
+  compaction (the compact system prompt instructs it), then continues.
+
+Scoring reuses the ORIGINAL task's verifier (skeleton stub below). Exec/sandbox envs restore the
+resume point's snapshot in ``setup`` before the harness runs (skeleton: ref is None -> no-op).
+
+NOTE (online/growing buffer): ``load_tasks`` runs once at env-server start, so this offline
+materialization sees only the buffer present at startup. The online variant samples a stored
+trace + resume point per rollout in a custom replay harness (rollout-time), keeping num_tasks
+virtual; left as a follow-up.
+"""
+
+from __future__ import annotations
+
+import glob
+import json
+
+from verifiers.v1.decorators import reward
+from verifiers.v1.runtimes import Runtime
+from verifiers.v1.task import Task
+from verifiers.v1.taskset import Taskset, TasksetConfig
+from verifiers.v1.trace import Trace, WireTrace
+
+from replay.selector import resume_points, seed_messages, snapshot_ref_of
+
+
+class ReplayTask(Task):
+    """A replayed prefix plus the provenance needed to score it with the original verifier."""
+
+    source_trace_id: str = ""
+    resume_node: int = -1
+    resume_kind: str = ""
+    snapshot_ref: str | None = None
+    original_task: dict = {}
+    original_reward: float = 0.0
+
+
+class ReplayTasksetConfig(TasksetConfig):
+    buffer_glob: str = ""
+    """Glob of stored-rollout JSONL files, e.g. ``.../rollouts/step_*/train_rollouts.jsonl``."""
+    kinds: list[str] = ["compaction_after", "compaction_before"]
+    """Which resume-point kinds to materialize tasks from."""
+
+
+class ReplayTaskset(Taskset[ReplayTask, ReplayTasksetConfig]):
+    def load_tasks(self) -> list[ReplayTask]:
+        tasks: list[ReplayTask] = []
+        kinds = set(self.config.kinds)
+        for path in sorted(glob.glob(self.config.buffer_glob)):
+            with open(path) as f:
+                for line in f:
+                    line = line.strip()
+                    if not line:
+                        continue
+                    src: Trace = WireTrace.model_validate(json.loads(line))
+                    for pt in resume_points(src, kinds=kinds):
+                        tasks.append(
+                            ReplayTask(
+                                idx=len(tasks),
+                                prompt=seed_messages(src, pt["node"]),
+                                source_trace_id=src.id,
+                                resume_node=pt["node"],
+                                resume_kind=pt["kind"],
+                                snapshot_ref=snapshot_ref_of(src, pt["node"]),
+                                original_task=src.task.model_dump(),
+                                original_reward=src.reward,
+                            )
+                        )
+        return tasks
+
+    async def setup(self, task: ReplayTask, runtime: Runtime) -> None:
+        # Exec/sandbox replay: restore the sandbox to the resume point before the harness runs.
+        # Skeleton: snapshot capture isn't wired yet, so ref is None -> nothing to restore.
+        if task.snapshot_ref is not None:
+            await runtime.restore(task.snapshot_ref)
+
+    @reward
+    def replay_reward(self, trace: Trace, task: ReplayTask) -> float:
+        # TODO: reuse the ORIGINAL env's verifier — reconstruct its taskset + the original Task
+        # from ``task.original_task`` and delegate (``inner.score(trace, runtime)``); for the
+        # judge-style mode, compare the model's verdict against ``task.original_reward``.
+        # Skeleton placeholder so the env is runnable end-to-end.
+        return 0.0
+
+
+__all__ = ["ReplayTaskset", "ReplayTasksetConfig", "ReplayTask"]

verifiers/v1/graph.py

 model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True) 
 model_config = ConfigDict(extra="forbid", arbitrary_types_allowed=True) 
@@ -21,7 +21,7 @@
 import hashlib
 import json
 from dataclasses import dataclass
-from typing import TYPE_CHECKING, Any
+from typing import TYPE_CHECKING, Any, Literal
 
 import numpy as np
 from pydantic import ConfigDict, Field, field_serializer, field_validator
@@ -60,6 +60,14 @@ def _decode_ndarray(d: dict) -> np.ndarray:
     return np.frombuffer(d["data"], dtype=np.dtype(d["dtype"])).reshape(d["shape"])
 
 
+NodeTag = Literal["compaction_before", "compaction_after", "subagent"] | None
+"""Replay resume-point tag for a node — branch provenance the typed graph can't otherwise
+express. ``compaction_before``/``compaction_after`` mark the two replay points around a
+compaction (regenerate it vs. continue from it); ``subagent`` marks a subagent fork. Written by
+the harness during generation, read by the replay-buffer resume-point selector. None for an
+ordinary node."""
+
+
 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
@@ -73,6 +81,16 @@ class MessageNode(StrictBaseModel):
     """True iff a model call produced this message (the response passed to `commit`); False for
     every prompt-supplied message — including assistant/tool messages fabricated as context
     the model never generated, which role alone can't tell apart from real turns."""
+    kind: NodeTag = None
+    """Replay resume-point tag, written by the harness during generation: branch provenance
+    (compaction_before/compaction_after/subagent); None for an ordinary node. Read by the
+    replay-buffer resume-point selector. A plain str/None, so it rides the wire and the JSON
+    dump automatically (no `_NODE_DUMP_EXCLUDE` entry needed)."""
+    snapshot_ref: str | None = None
+    """Durable handle to the sandbox state captured after this node's turn (registry tag /
+    modal snapshot id / object-store key); restore it to resume a replay from here. Written by
+    the per-turn snapshot hook during generation; None when snapshotting was off/unsupported.
+    A plain str/None, so it rides the wire and the JSON dump automatically."""
     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

verifiers/v1/runtimes/base.py

@@ -258,6 +258,22 @@ async def read(self, path: str) -> bytes:
     async def write(self, path: str, data: bytes) -> None:
         """Write a file into the runtime's workspace, creating parent dirs."""
 
+    # --- snapshot / restore (replay buffer) ---
+
+    async def snapshot(self) -> str | None:
+        """Persist the current sandbox state and return a DURABLE ref (e.g. a registry tag,
+        modal snapshot id, or object-store key). The ref must outlive this runtime instance:
+        it is read back by a *different*, later rollout — possibly on another machine, possibly
+        replaying an offline buffer from a prior run. Override per runtime when the sandbox
+        supports checkpointing. Skeleton: returns None (no snapshot taken)."""
+        return None
+
+    async def restore(self, ref: str) -> None:
+        """Restore the sandbox to a previously snapshotted state before continuing a replay
+        from mid-rollout. Override per runtime. Skeleton: refuse a real ref rather than
+        silently resuming on the wrong state (a None ref is a no-op handled by the caller)."""
+        raise NotImplementedError(f"{self.type}: snapshot/restore not available yet")
+
     # --- networking ---
 
     @property