Option B: typed MessageNode.kind for replay resume points

environments/compact/compact/annotate.py

@@ -0,0 +1,86 @@
+"""Replay resume-point annotation + selection (Option B: typed ``MessageNode.kind`` field).
+
+During generation the harness tags interesting nodes; the replay-buffer selector reads the
+tags back. Tool *identity* (which tool) and "is this a tool call" come from the typed graph,
+so only two things need a tag: branch provenance (compaction/subagent) and tool failure
+status (``failed_tool_call``), which the trace otherwise drops.
+
+Option B stores the tag on the node itself (``MessageNode.kind``; see ``verifiers/v1/graph.py``).
+The label rides *with* the node — no keying, typed/validated, and visible anywhere traces are
+inspected — at the cost of a change to the core ``MessageNode`` schema.
+"""
+
+from __future__ import annotations
+
+from verifiers.v1.graph import MessageNode
+from verifiers.v1.trace import Trace
+from verifiers.v1.types import AssistantMessage, ToolMessage
+
+
+def branch_start_nodes(trace: Trace) -> list[int]:
+    """First node of each forked branch. A node with >1 child is a fork point; ``children[0]``
+    is the original line and ``children[1:]`` are the forks. The compacting harness rewrites
+    context every turn, so each fork start 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 tool_nodes(trace: Trace) -> dict[str, int]:
+    """``tool_call_id -> node_id`` for every tool result — the key for joining the program's
+    per-call failure log back onto the graph."""
+    return {
+        node.message.tool_call_id: nid
+        for nid, node in enumerate(trace.nodes)
+        if isinstance(node.message, ToolMessage)
+    }
+
+
+def tool_name(trace: Trace, node: MessageNode) -> str | None:
+    """The tool/function name for a tool-result node: ``ToolMessage.name`` when the dialect
+    recovered it, else the issuing assistant's matching ``tool_calls[].name``."""
+    m = node.message
+    if not isinstance(m, ToolMessage):
+        return None
+    if m.name:
+        return m.name
+    nid = node.parent
+    while nid is not None:
+        parent = trace.nodes[nid].message
+        if isinstance(parent, AssistantMessage):
+            for tc in parent.tool_calls or []:
+                if tc.id == m.tool_call_id:
+                    return tc.name
+            break
+        nid = trace.nodes[nid].parent
+    return None
+
+
+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 resume_points(trace: Trace, *, kinds: set[str]) -> list[dict]:
+    """Enumerate replay resume points, filtered to ``kinds``. Tool calls come from the typed
+    graph; compaction comes from the node tag. Each point carries its ``node`` id, ``kind``,
+    and for tool calls the ``tool`` name and whether it ``failed``."""
+    points: list[dict] = []
+    for nid, node in enumerate(trace.nodes):
+        if isinstance(node.message, ToolMessage):
+            points.append(
+                {
+                    "node": nid,
+                    "kind": "tool_call",
+                    "tool": tool_name(trace, node),
+                    "failed": get_tag(trace, nid) == "failed_tool_call",
+                }
+            )
+        elif get_tag(trace, nid) == "compaction":
+            points.append({"node": nid, "kind": "compaction"})
+    return [p for p in points if p["kind"] in kinds]

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 branch_start_nodes, tool_nodes
+
 PROGRAM_SOURCE = (Path(__file__).resolve().parent / "program.py").read_text()
 
 
@@ -39,10 +41,12 @@ async def launch(
         secret: str,
         mcp_urls: dict[str, str],
     ) -> ProgramResult:
+        tool_log_path = "/tmp/vf_compact_tool_log.json"
         env = {
             "OPENAI_BASE_URL": endpoint,
             "OPENAI_API_KEY": secret,
             "OPENAI_MODEL": ctx.model,
+            "TOOL_LOG": tool_log_path,
         }
         if mcp_urls:
             # The program connects to the tool servers over HTTP; hand it a standard
@@ -51,4 +55,15 @@ 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 replay resume points on the finished graph (Option B: typed MessageNode.kind).
+        # The harness is the only writer; the program was the sensor for tool failures.
+        for nid in branch_start_nodes(trace):  # this harness rewrites context every turn
+            trace.nodes[nid].kind = "compaction"
+        by_id = tool_nodes(trace)
+        for rec in json.loads(await runtime.read(tool_log_path)):
+            nid = by_id.get(rec["tool_call_id"])
+            if nid is not None:
+                trace.nodes[nid].kind = rec["tag"]
+        return result

environments/compact/compact/program.py

@@ -27,9 +27,14 @@
 import re
 import sys
 from contextlib import AsyncExitStack
+from pathlib import Path
 
 from openai import AsyncOpenAI
 
+# Where to surface per-tool-call failures the trace can't keep (the harness reads this back
+# and tags the matching nodes). The harness sets it; absent when running the program standalone.
+TOOL_LOG = os.environ.get("TOOL_LOG")
+
 SYSTEM = (
     "You solve a task across several turns; your NOTES are your only lasting memory. The "
     "first turn shows the task; after that you see only your notes — plus, the turn right "
@@ -103,6 +108,9 @@ async def main() -> None:
     config = json.loads(os.environ.get("MCP_CONFIG", "{}"))
     notes: str | None = None  # the durable memory carried across turns
     tool_output: str | None = None  # the last tool result, kept for exactly one turn
+    tool_log: list[dict] = []  # per-tool-call status the trace drops; surfaced for the harness
+    if TOOL_LOG:
+        Path(TOOL_LOG).write_text("[]")  # ensure the harness always finds the sensor's log
     async with AsyncExitStack() as stack:
         tools, dispatch = (
             await connect_mcp(stack, config) if config.get("mcpServers") else ([], {})
@@ -129,13 +137,19 @@ async def main() -> None:
                 results = []
                 for call in message.tool_calls:
                     args = json.loads(call.function.arguments or "{}")
-                    result = (
-                        await call_mcp(dispatch, call.function.name, args)
-                        if call.function.name in dispatch
-                        else f"error: unknown tool {call.function.name!r}"
-                    )
+                    if call.function.name in dispatch:
+                        try:
+                            result, tag = await call_mcp(dispatch, call.function.name, args), None
+                        except Exception as e:  # the failure signal the trace can't keep
+                            result, tag = f"error: {e}", "failed_tool_call"
+                    else:
+                        result, tag = f"error: unknown tool {call.function.name!r}", "failed_tool_call"
+                    if tag:
+                        tool_log.append({"tool_call_id": call.id, "tag": tag})
                     results.append(f"{call.function.name}:\n{result}")
                 tool_output = "\n\n".join(results)
+                if TOOL_LOG:  # the harness reads this back to tag failed-tool-call nodes
+                    Path(TOOL_LOG).write_text(json.dumps(tool_log))
 
 
 if __name__ == "__main__":

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,13 @@ def _decode_ndarray(d: dict) -> np.ndarray:
     return np.frombuffer(d["data"], dtype=np.dtype(d["dtype"])).reshape(d["shape"])
 
 
+NodeTag = Literal["compaction", "subagent", "failed_tool_call"] | None
+"""Replay resume-point tag for a node: branch provenance (compaction/subagent) or tool failure
+status (failed_tool_call) — the two things the typed graph can't otherwise express. 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 +80,12 @@ 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/subagent) or tool failure status (failed_tool_call); None for an ordinary node.
+    Read by the replay-buffer resume-point selector. Tool identity and "is this a tool call"
+    come from the typed message, so they need no tag. A plain str/None, so it rides the wire and
+    the JSON dump automatically (no `_NODE_DUMP_EXCLUDE` entry needed)."""
     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