From f274865d7120b21dc2b32d01baf9772c581f7433 Mon Sep 17 00:00:00 2001
From: Jaden Fix <163031992+jadenfix@users.noreply.github.com>
Date: Fri, 9 Oct 2026 12:03:14 -0700
Subject: [PATCH 1/6] [skycap] `skycap check`: report whether a record stays on
 one path, and why each fork happened

Reads a record directory's documents (no sidecars) and, per trajectory,
reports the path count and a cause for each fork: resample, edited reply,
re-rendered (token mode), different message, or tools or model. Exits 0 if
every trajectory is linear, 1 if any forks, 2 if a record can't be read.

record.load's graph rebuild moves into record.add_nodes so check shares it.
---
 skycap/README.md            |  24 +++
 skycap/src/skycap/check.py  | 235 +++++++++++++++++++++++++++
 skycap/src/skycap/cli.py    |  40 ++++-
 skycap/src/skycap/record.py |  20 ++-
 skycap/tests/test_check.py  | 307 ++++++++++++++++++++++++++++++++++++
 5 files changed, 620 insertions(+), 6 deletions(-)
 create mode 100644 skycap/src/skycap/check.py
 create mode 100644 skycap/tests/test_check.py

diff --git a/skycap/README.md b/skycap/README.md
index b2c9b4fb1d..fe759d6ce7 100644
--- a/skycap/README.md
+++ b/skycap/README.md
@@ -283,6 +283,30 @@ doesn't hold; readers treat those as absent (see [`docs/format.md`](docs/format.
 The local record is untouched. Excluding `tokens` is allowed, but a viewer of
 the mirror then shows message text only.
 
+## Check a harness
+
+A harness that only appends to its history should record one path per
+trajectory. `skycap check` reads a record directory and reports, for each
+trajectory, how many paths it has and why each fork happened:
+
+```console
+$ uv run skycap check ./record
+tr_a  finished  linear
+tr_b  finished  2 paths
+  node 2 (under node 0, vs node 1): edited reply: the harness sent back the model's reply with reasoning_content dropped
+1 of 2 trajectories linear; forks: 1 edited reply
+```
+
+A fork is a `resample` (the model replied again to a history it had already
+answered), an `edited reply` (the harness replayed a model reply with fields
+changed, such as stripped reasoning), `re-rendered` (token mode: the same
+message, but its render didn't reproduce the sampled tokens), a `different
+message` (a compaction, a subagent, a rewritten turn), or `tools or model` (the
+same message sent with a different tool set or model). In token mode it also
+counts unbridged calls, whose prompt was re-rendered rather than extended.
+It exits 0 if every trajectory is linear, 1 if any forks and 2 if a record
+can't be read. Pass trajectory ids after the directory to check only those.
+
 ## Develop
 
 ```bash
diff --git a/skycap/src/skycap/check.py b/skycap/src/skycap/check.py
new file mode 100644
index 0000000000..259f974c8b
--- /dev/null
+++ b/skycap/src/skycap/check.py
@@ -0,0 +1,235 @@
+"""`skycap check`: whether a harness keeps a history that should be linear on one path.
+
+A harness that only ever appends to its history produces one root-to-leaf path
+per trajectory. Every fork is a place where a request's history stopped
+matching what the graph already had, and comparing the new branch's first node
+with the sibling it should have matched says why:
+
+* ``resample``: the model replied again to a history it had already answered
+  (a retry, ``n > 1``, or a reply to a history the harness cut short).
+* ``edited reply``: the harness sent back a model reply with fields changed,
+  such as stripped reasoning or a repaired tool call.
+* ``re-rendered``: token mode only. The same message, but its re-render didn't
+  reproduce the tokens the model sampled, so it couldn't continue from them.
+* ``different message``: a different message after the same history, such as
+  a compaction, a subagent or a rewritten turn.
+* ``tools or model``: the same message, sent with another tool set or model.
+  A match covers both, so this starts a new root.
+
+A second root is compared with the earlier roots the same way.
+
+This reads only the record documents, never the token sidecars, so it is cheap
+to run over a whole record directory.
+"""
+
+from __future__ import annotations
+
+from collections import Counter
+from collections.abc import Iterator, Mapping, Sequence
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Literal
+
+import zstandard
+
+from skycap import record
+from skycap.graph import MessageGraph, Node
+from skycap.hashing import canonical_bytes, canonical_message, rendered_fields
+
+Kind = Literal["resample", "edited reply", "re-rendered", "different message", "tools or model"]
+
+
+@dataclass(frozen=True, slots=True)
+class Fork:
+    """Where a trajectory's history left the path it was on, and why."""
+
+    #: The node the fork is under; ``None`` for a second root.
+    parent: int | None
+    #: The first node of the new branch.
+    node: int
+    #: The earlier sibling it was compared with: the one its history should have matched.
+    sibling: int
+    kind: Kind
+    #: What differs, in words.
+    detail: str
+
+
+@dataclass(frozen=True, slots=True)
+class Report:
+    """One trajectory's paths and forks, or why its record couldn't be read."""
+
+    trajectory_id: str
+    status: str
+    paths: int = 0
+    forks: tuple[Fork, ...] = ()
+    #: Token mode: calls whose prompt was rendered from the messages instead of
+    #: extending the previous call's tokens (``CallInfo.bridged is False``).
+    unbridged_calls: int = 0
+    #: Set when the record couldn't be read; nothing else is then known.
+    error: str | None = None
+
+    @property
+    def linear(self) -> bool:
+        """One path, or none for a trajectory that made no calls."""
+        return self.error is None and self.paths <= 1
+
+
+#: The errors a malformed or foreign record raises while it is read.
+_READ_ERRORS = (OSError, ValueError, KeyError, TypeError, AssertionError, zstandard.ZstdError)
+
+
+def check_document(document: dict[str, Any]) -> Report:
+    """Check one record document, as ``record.read_document`` returns it."""
+    version = document.get("format_version")
+    if version != record.FORMAT_VERSION:
+        return Report(
+            document.get("id", "?"),
+            "unreadable",
+            error=f"format_version {version}, this skycap reads {record.FORMAT_VERSION}",
+        )
+    graph = MessageGraph()
+    record.add_nodes(graph, document)
+    tokens = (document.get("capture") or {}).get("mode") == "tokens"
+    normalize = _token_message if tokens else canonical_message
+    # Each message as the graph's matching sees it, normalized once.
+    messages = {node.id: normalize(node.message) for node in graph}
+    forks = [fork for parent in [None, *(node.id for node in graph)] for fork in _forks_under(graph, parent, messages)]
+    return Report(
+        trajectory_id=document["id"],
+        status=document["status"],
+        paths=len(graph.leaves()),
+        forks=tuple(sorted(forks, key=lambda fork: fork.node)),
+        unbridged_calls=graph.unbridged_calls(),
+    )
+
+
+def check_dir(record_dir: Path, trajectory_ids: Sequence[str] = ()) -> list[Report]:
+    """Check the given trajectories, or every one in ``record_dir``.
+
+    A record that can't be read becomes a report with ``error`` set rather
+    than stopping the rest.
+    """
+    reports = []
+    for trajectory_id in trajectory_ids or list(record.list_ids(record_dir)):
+        try:
+            reports.append(check_document(record.read_document(record_dir, trajectory_id)))
+        except _READ_ERRORS as error:
+            reports.append(Report(trajectory_id, "unreadable", error=f"{type(error).__name__}: {error}"))
+    return reports
+
+
+def _token_message(message: Mapping[str, Any]) -> dict[str, Any]:
+    """A token-mode message as its match hash sees it: the rendered fields (``MatchKey.fields``)."""
+    return canonical_message(rendered_fields(message))
+
+
+def _forks_under(graph: MessageGraph, parent: int | None, messages: dict[int, dict[str, Any]]) -> Iterator[Fork]:
+    """One fork per child after the first: each started a path its earlier siblings don't share."""
+    children = [graph.nodes[i] for i in graph.children(parent)]
+    by_match: dict[str, list[Node]] = {}
+    by_message: dict[bytes, Node] = {}
+    for index, node in enumerate(children):
+        if index:
+            sibling = _sibling_to_compare(children, index, by_match, by_message, messages)
+            kind, detail = _cause(sibling, node, messages)
+            yield Fork(parent=parent, node=node.id, sibling=sibling.id, kind=kind, detail=detail)
+        by_match.setdefault(node.match_hash, []).append(node)
+        by_message[canonical_bytes(messages[node.id])] = node
+
+
+def _sibling_to_compare(
+    children: list[Node],
+    index: int,
+    by_match: dict[str, list[Node]],
+    by_message: dict[bytes, Node],
+    messages: dict[int, dict[str, Any]],
+) -> Node:
+    """The earlier sibling ``children[index]``'s history should have matched: the closest one.
+
+    One with the same match hash first (only the tokens or the sampling
+    differ), preferring a model reply, then one with the same message (only the
+    tools or model differ). Otherwise the same-role sibling, preferring a model
+    reply: for a model node the latest, since every model sibling is a resample
+    of it; for a client node the one with the fewest differing fields.
+    """
+    node, earlier = children[index], children[:index]
+    same_match = by_match.get(node.match_hash)
+    if same_match:
+        return ([s for s in same_match if s.author == "model"] or same_match)[-1]
+    same_message = by_message.get(canonical_bytes(messages[node.id]))
+    if same_message is not None:
+        return same_message
+    same_role = [s for s in earlier if s.role == node.role]
+    if not same_role:
+        return earlier[-1]
+    if node.author == "model":
+        return ([s for s in same_role if s.author == "model"] or same_role)[-1]
+    return min(
+        same_role,
+        key=lambda s: (len(_changes(messages[s.id], messages[node.id])), s.author != "model", -s.id),
+    )
+
+
+def _cause(sibling: Node, node: Node, messages: dict[int, dict[str, Any]]) -> tuple[Kind, str]:
+    before, after = messages[sibling.id], messages[node.id]
+    if node.author == "model":
+        if node.match_hash == sibling.match_hash:
+            if sibling.author == "model":
+                return "resample", "the same reply again, under other sampling parameters or tokenized differently"
+            return "resample", "the model sampled a message the harness had already written"
+        if sibling.role == node.role:
+            return "resample", "a second reply to the same history"
+        return "resample", f"a reply to a history the other path continues with {_a(sibling.role)} message"
+    if node.match_hash == sibling.match_hash:
+        return "re-rendered", "the same message, rendered to different tokens"
+    if before == after:
+        return "tools or model", "the same message, sent with a different tool set or model"
+    if sibling.role != node.role:
+        return "different message", f"{_a(node.role)} message where the other path has {_a(sibling.role)} message"
+    changes = ", ".join(_changes(before, after))
+    if sibling.author == "model":
+        return "edited reply", f"the harness sent back the model's reply with {changes}"
+    return "different message", f"a different {node.role} message: {changes}"
+
+
+def _changes(before: dict[str, Any], after: dict[str, Any]) -> list[str]:
+    """Which top-level fields differ, e.g. ``["content changed", "reasoning_content dropped"]``."""
+    changes = []
+    for key in sorted(before.keys() | after.keys()):
+        if key not in after:
+            changes.append(f"{key} dropped")
+        elif key not in before:
+            changes.append(f"{key} added")
+        elif before[key] != after[key]:
+            changes.append(f"{key} changed")
+    return changes
+
+
+def _a(role: str | None) -> str:
+    word = role or "unnamed"
+    return f"{'an' if word[0] in 'aeio' else 'a'} {word}"
+
+
+def format_reports(reports: Sequence[Report]) -> str:
+    """One line per trajectory, one per fork under it, and a summary line."""
+    lines = []
+    for report in reports:
+        if report.error is not None:
+            lines.append(f"{report.trajectory_id}  unreadable: {report.error}")
+            continue
+        shape = "empty" if report.paths == 0 else "linear" if report.linear else f"{report.paths} paths"
+        lines.append(f"{report.trajectory_id}  {report.status}  {shape}")
+        for fork in report.forks:
+            where = "new root" if fork.parent is None else f"under node {fork.parent}"
+            lines.append(f"  node {fork.node} ({where}, vs node {fork.sibling}): {fork.kind}: {fork.detail}")
+        if report.unbridged_calls:
+            lines.append(f"  {report.unbridged_calls} unbridged call(s): prompt re-rendered, not extended")
+    readable = [report for report in reports if report.error is None]
+    summary = f"{sum(report.linear for report in readable)} of {len(readable)} trajectories linear"
+    kinds = Counter(fork.kind for report in readable for fork in report.forks)
+    if kinds:
+        summary += "; forks: " + ", ".join(f"{count} {kind}" for kind, count in kinds.most_common())
+    if len(readable) < len(reports):
+        summary += f"; {len(reports) - len(readable)} unreadable"
+    lines.append(summary)
+    return "\n".join(lines)
diff --git a/skycap/src/skycap/cli.py b/skycap/src/skycap/cli.py
index 1e52ec746b..18dc84d171 100644
--- a/skycap/src/skycap/cli.py
+++ b/skycap/src/skycap/cli.py
@@ -1,4 +1,4 @@
-"""`skycap serve`: run one capture server against one upstream."""
+"""`skycap serve`: run one capture server against one upstream. `skycap check`: report forks in a record."""
 
 from __future__ import annotations
 
@@ -9,10 +9,12 @@
 import os
 import signal
 import sys
+from pathlib import Path
 
 from aiohttp import web
 
-from skycap import __version__
+from skycap import __version__, record
+from skycap.check import check_dir, format_reports
 from skycap.exposure import Exposure, load_exposure
 from skycap.server import CaptureServer
 from skycap.service import build_backend, serve
@@ -129,6 +131,20 @@ def build_parser() -> argparse.ArgumentParser:
         help="harness routes answer only a caller sending the trajectory's own key (create returns it) "
         "as `Authorization: Bearer <key>`, as an OpenAI client does with its api_key",
     )
+
+    check = commands.add_parser(
+        "check",
+        help="report whether recorded trajectories stayed on one path, and the cause of each fork",
+        description="Report whether recorded trajectories stayed on one path, and the cause of each fork, so a "
+        "harness whose history should be linear can be checked against a recorded run.",
+        epilog="causes: resample (another reply to the same history), edited reply (a model reply sent back with "
+        "fields changed), re-rendered (token mode: the same message rendered to different tokens), different "
+        "message (a compaction, subagent or rewritten turn), tools or model (the same message with another tool "
+        "set or model). Exit status: 0 if every trajectory is linear, 1 if any forks, 2 if a record can't be read.",
+    )
+    check.set_defaults(subparser=check)
+    check.add_argument("record_dir", type=Path, help="a --record-dir written by `skycap serve`")
+    check.add_argument("trajectory_ids", nargs="*", help="trajectories to check (default: every one in the directory)")
     return parser
 
 
@@ -172,6 +188,8 @@ def build_server(args: argparse.Namespace) -> CaptureServer:
 
 def main(argv: list[str] | None = None) -> int:
     args = build_parser().parse_args(argv)
+    if args.command == "check":
+        return check(args.subparser, args.record_dir, args.trajectory_ids)
     logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s: %(message)s")
     if args.command == "serve":
         if args.expose is None and args.expose_kwargs:
@@ -205,5 +223,23 @@ def ready(url: str, harness_url: str | None) -> None:
     )
 
 
+def check(parser: argparse.ArgumentParser, record_dir: Path, trajectory_ids: list[str]) -> int:
+    """`skycap check`. Exits 0 if every trajectory is linear, 1 if any forks, 2 if a record can't be read."""
+    if not record_dir.is_dir():
+        parser.error(f"{record_dir} is not a directory")
+    ids = list(dict.fromkeys(trajectory_ids))
+    known = set(record.list_ids(record_dir))
+    unknown = [i for i in ids if i not in known]
+    if unknown:
+        parser.error(f"no trajectory {', '.join(unknown)} in {record_dir}")
+    reports = check_dir(record_dir, ids)
+    if not reports:
+        parser.error(f"no trajectories in {record_dir}")
+    print(format_reports(reports))
+    if any(report.error is not None for report in reports):
+        return 2
+    return 0 if all(report.linear for report in reports) else 1
+
+
 if __name__ == "__main__":
     sys.exit(main())
diff --git a/skycap/src/skycap/record.py b/skycap/src/skycap/record.py
index a981cc6ce0..ed9baa439b 100644
--- a/skycap/src/skycap/record.py
+++ b/skycap/src/skycap/record.py
@@ -33,7 +33,7 @@
 import orjson
 import zstandard
 
-from skycap.graph import CallInfo, NodeTokens
+from skycap.graph import CallInfo, MessageGraph, NodeTokens
 from skycap.trajectory import Failure, Trajectory
 
 logger = logging.getLogger(__name__)
@@ -247,7 +247,20 @@ def load(record_dir: Path, trajectory_id: str) -> Trajectory:
     retries = document.get("retries") or {}
     trajectory.replay.replayed = retries.get("replayed", 0)
     trajectory.replay.coalesced = retries.get("coalesced", 0)
-    graph = trajectory.graph
+    add_nodes(trajectory.graph, document, arrays)
+    return trajectory
+
+
+def add_nodes(
+    graph: MessageGraph,
+    document: dict[str, Any],
+    arrays: dict[str, dict[str, np.ndarray] | None] | None = None,
+) -> None:
+    """Rebuild a written document's tools and nodes into ``graph``.
+
+    Without ``arrays`` (the sidecars ``load`` reads) the nodes carry no tokens,
+    which is enough for anything that only reads the graph's shape.
+    """
     graph.tools.update(document["tools"])
     for entry in document["nodes"]:
         node, _ = graph.add(
@@ -258,11 +271,10 @@ def load(record_dir: Path, trajectory_id: str) -> Trajectory:
             match_hash=entry["match_hash"],
             delta_hash=entry["delta_hash"],
             created_at=entry["created_at"],
-            tokens=_node_tokens(entry.get("tokens"), arrays),
+            tokens=None if arrays is None else _node_tokens(entry.get("tokens"), arrays),
         )
         assert node.id == entry["id"], "record nodes are stored in creation order"
         node.calls.extend(CallInfo(**call) for call in entry["calls"])
-    return trajectory
 
 
 def _node_tokens(meta: dict[str, Any] | None, arrays: dict[str, dict[str, np.ndarray] | None]) -> NodeTokens | None:
diff --git a/skycap/tests/test_check.py b/skycap/tests/test_check.py
new file mode 100644
index 0000000000..ac3c6d31aa
--- /dev/null
+++ b/skycap/tests/test_check.py
@@ -0,0 +1,307 @@
+"""`skycap check`: each kind of fork a harness can cause, read back from a written record."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+import orjson
+import pytest
+import zstandard
+
+from skycap import cli, record
+from skycap.check import check_dir, format_reports
+from skycap.graph import CallInfo
+from skycap.trajectory import Trajectory
+from tests.conftest import Stack, openai_client
+from tests.fake_renderer import END, NL, encode
+from tests.test_tokens import token_stack
+
+SYS = {"role": "system", "content": "You are terse."}
+SEARCH = [{"type": "function", "function": {"name": "search", "parameters": {}}}]
+
+
+def user(text: str) -> dict:
+    return {"role": "user", "content": text}
+
+
+def assistant(text: str, **extra) -> dict:
+    return {"role": "assistant", "content": text, **extra}
+
+
+class Harness:
+    """Drives one text-mode trajectory's graph a call at a time, then writes it."""
+
+    def __init__(self, trajectory_id: str) -> None:
+        self.trajectory = Trajectory(id=trajectory_id)
+        self.clock = 0.0
+
+    def call(self, messages, reply, *, tools=None, model="policy", bridged=None, **sampling) -> None:
+        self.clock += 1.0
+        info = CallInfo(t_start=self.clock, t_end=self.clock + 0.5, model=model, sampling=sampling, bridged=bridged)
+        self.trajectory.graph.commit_text(messages, reply, tools=tools, model=model, call=info)
+
+    def write(self, record_dir: Path) -> None:
+        self.trajectory.seal("finished", {"reward": 1.0})
+        record.write(record_dir, self.trajectory)
+
+
+def report_lines(record_dir: Path, *harnesses: Harness) -> list[str]:
+    for harness in harnesses:
+        harness.write(record_dir)
+    return format_reports(check_dir(record_dir)).splitlines()
+
+
+def test_an_append_only_history_is_linear(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([SYS, user("q1")], assistant("a1"))
+    h.call([SYS, user("q1"), assistant("a1"), user("q2")], assistant("a2"))
+
+    assert report_lines(tmp_path, h) == ["tr  finished  linear", "1 of 1 trajectories linear"]
+
+
+def test_a_trajectory_without_calls_is_empty(tmp_path: Path) -> None:
+    assert report_lines(tmp_path, Harness("tr")) == ["tr  finished  empty", "1 of 1 trajectories linear"]
+
+
+def test_stripped_reasoning_is_an_edited_reply(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([user("q")], assistant("answer", reasoning_content="let me think"))
+    h.call([user("q"), assistant("answer"), user("more")], assistant("more answer"))
+
+    assert report_lines(tmp_path, h) == [
+        "tr  finished  2 paths",
+        "  node 2 (under node 0, vs node 1): edited reply: "
+        "the harness sent back the model's reply with reasoning_content dropped",
+        "0 of 1 trajectories linear; forks: 1 edited reply",
+    ]
+
+
+def test_an_edit_is_compared_with_the_reply_it_came_from(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([user("q")], assistant("A", reasoning_content="r1"))
+    h.call([user("q")], assistant("B", reasoning_content="r2"))
+    h.call([user("q"), assistant("A"), user("more")], assistant("ok"))
+
+    assert report_lines(tmp_path, h)[1:3] == [
+        "  node 2 (under node 0, vs node 1): resample: a second reply to the same history",
+        "  node 3 (under node 0, vs node 1): edited reply: "
+        "the harness sent back the model's reply with reasoning_content dropped",
+    ]
+
+
+def test_the_same_reply_under_other_sampling_is_a_resample(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([user("q")], assistant("a"), top_p=1.0)
+    h.call([user("q")], assistant("a"), top_p=0.5)
+
+    assert report_lines(tmp_path, h)[1] == (
+        "  node 2 (under node 0, vs node 1): resample: "
+        "the same reply again, under other sampling parameters or tokenized differently"
+    )
+
+
+def test_a_sample_equal_to_a_harness_written_message_is_a_resample(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([user("q"), assistant("x"), user("again")], assistant("y"))
+    h.call([user("q")], assistant("x"))
+
+    assert report_lines(tmp_path, h)[1] == (
+        "  node 4 (under node 0, vs node 1): resample: the model sampled a message the harness had already written"
+    )
+
+
+def test_a_reply_to_a_shortened_history_is_a_resample(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([SYS, user("task")], assistant("a"))
+    h.call([SYS], assistant("b"))
+
+    assert report_lines(tmp_path, h)[1] == (
+        "  node 3 (under node 0, vs node 1): resample: "
+        "a reply to a history the other path continues with a user message"
+    )
+
+
+def test_compaction_is_a_different_message(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([SYS, user("task")], assistant("step1"))
+    h.call([SYS, user("task"), assistant("step1"), user("summarize")], assistant("summary"))
+    h.call([SYS, user("summary"), user("continue")], assistant("step2"))
+
+    assert report_lines(tmp_path, h)[1] == (
+        "  node 5 (under node 0, vs node 1): different message: a different user message: content changed"
+    )
+
+
+def test_a_subagent_with_its_own_system_prompt_is_a_new_root(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([SYS, user("task")], assistant("delegating"))
+    h.call([{"role": "system", "content": "You are a searcher."}, user("find x")], assistant("found"))
+
+    assert report_lines(tmp_path, h)[1] == (
+        "  node 3 (new root, vs node 0): different message: a different system message: content changed"
+    )
+
+
+def test_a_changed_tool_set_is_compared_with_the_root_it_repeats(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([SYS, user("task")], assistant("plan"))
+    h.call([{"role": "system", "content": "You are a searcher."}, user("find x")], assistant("found"))
+    h.call([SYS, user("task"), assistant("plan"), user("go")], assistant("done"), tools=SEARCH)
+
+    assert report_lines(tmp_path, h)[2] == (
+        "  node 6 (new root, vs node 0): tools or model: the same message, sent with a different tool set or model"
+    )
+
+
+def test_a_message_of_another_role_names_both(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([SYS, user("task")], assistant("plan"))
+    h.call([SYS, {"role": "assistant", "content": "prefill"}], assistant("x"))
+
+    assert report_lines(tmp_path, h)[1] == (
+        "  node 3 (under node 0, vs node 1): different message: "
+        "an assistant message where the other path has a user message"
+    )
+
+
+def test_unbridged_calls_are_counted(tmp_path: Path) -> None:
+    h = Harness("tr")
+    h.call([user("q1")], assistant("a1"))
+    h.call([user("q1"), assistant("a1"), user("q2")], assistant("a2"), bridged=False)
+
+    assert report_lines(tmp_path, h) == [
+        "tr  finished  linear",
+        "  1 unbridged call(s): prompt re-rendered, not extended",
+        "1 of 1 trajectories linear",
+    ]
+
+
+async def test_a_reply_whose_re_render_changes_its_tokens_is_re_rendered(tmp_path: Path) -> None:
+    # Without END the bridge can't extend the completion, and the full render of
+    # the replayed reply ends it with END, so the same message gets new tokens.
+    def completion(prompt, sampling):
+        return [*encode("ans"), NL, *encode("wer")]
+
+    async with token_stack(completion=completion, record_dir=tmp_path) as stack:
+        created = await stack.create()
+        llm = openai_client(created["base_url"])
+        reply = (await llm.chat.completions.create(model="policy", messages=[user("q")])).choices[0].message
+        replayed = reply.model_dump(exclude_none=True)
+        await llm.chat.completions.create(model="policy", messages=[user("q"), replayed, user("more")])
+        await stack.finish(created["id"])
+
+    assert format_reports(check_dir(tmp_path)).splitlines() == [
+        f"{created['id']}  finished  2 paths",
+        "  node 2 (under node 0, vs node 1): re-rendered: the same message, rendered to different tokens",
+        "  1 unbridged call(s): prompt re-rendered, not extended",
+        "0 of 1 trajectories linear; forks: 1 re-rendered",
+    ]
+
+
+async def test_a_resample_after_a_re_render_is_compared_with_the_model_reply(tmp_path: Path) -> None:
+    def completion(prompt, sampling):
+        return [*encode("ans"), NL, *encode("wer")]
+
+    async with token_stack(completion=completion, record_dir=tmp_path) as stack:
+        created = await stack.create()
+        llm = openai_client(created["base_url"])
+        reply = (await llm.chat.completions.create(model="policy", messages=[user("q")], top_p=1.0)).choices[0]
+        replayed = reply.message.model_dump(exclude_none=True)
+        await llm.chat.completions.create(model="policy", messages=[user("q"), replayed, user("more")])
+        # The same tokens again under another top_p: a model sibling, not the re-rendered copy.
+        await llm.chat.completions.create(model="policy", messages=[user("q")], top_p=0.5)
+        await stack.finish(created["id"])
+
+    lines = format_reports(check_dir(tmp_path)).splitlines()
+    assert lines[1:3] == [
+        "  node 2 (under node 0, vs node 1): re-rendered: the same message, rendered to different tokens",
+        "  node 5 (under node 0, vs node 1): resample: "
+        "the same reply again, under other sampling parameters or tokenized differently",
+    ]
+
+
+async def test_token_mode_names_only_fields_the_renderer_uses(tmp_path: Path) -> None:
+    def completion(prompt, sampling):
+        return [*encode("THINK:hmm|CALL:search:{}"), END]
+
+    async with token_stack(completion=completion, record_dir=tmp_path) as stack:
+        created = await stack.create()
+        llm = openai_client(created["base_url"])
+        first = await llm.chat.completions.create(model="policy", messages=[user("q")], tools=SEARCH)
+        reply = first.choices[0].message.model_dump(exclude_none=True)
+        # Dropping reasoning forks. Client metadata such as provider_specific_fields
+        # isn't rendered, so adding it doesn't, and it mustn't be named.
+        replay = {key: value for key, value in reply.items() if key != "reasoning_content"}
+        replay["provider_specific_fields"] = {"x": 1}
+        tool_result = {"role": "tool", "tool_call_id": reply["tool_calls"][0]["id"], "content": "found"}
+        body = {"model": "policy", "messages": [user("q"), replay, tool_result], "tools": SEARCH}
+        async with stack.http.post(f"{created['base_url']}/chat/completions", json=body) as response:
+            assert response.status == 200
+        await stack.finish(created["id"])
+
+    assert format_reports(check_dir(tmp_path)).splitlines()[1] == (
+        "  node 2 (under node 0, vs node 1): edited reply: "
+        "the harness sent back the model's reply with reasoning_content dropped"
+    )
+
+
+async def test_check_reads_what_a_capture_server_wrote(recorded_stack: Stack, capsys: pytest.CaptureFixture) -> None:
+    async def run(edit_reply: bool) -> str:
+        created = await recorded_stack.create()
+        llm = openai_client(created["base_url"])
+        first = await llm.chat.completions.create(model="policy", messages=[user("q")])
+        replayed = "edited" if edit_reply else first.choices[0].message.content
+        await llm.chat.completions.create(model="policy", messages=[user("q"), assistant(replayed), user("next")])
+        await recorded_stack.finish(created["id"])
+        return created["id"]
+
+    appended, edited = await run(edit_reply=False), await run(edit_reply=True)
+    record_dir = recorded_stack.server.record_dir
+    assert record_dir is not None
+
+    assert cli.main(["check", str(record_dir), appended]) == 0
+    assert capsys.readouterr().out.splitlines() == [f"{appended}  finished  linear", "1 of 1 trajectories linear"]
+    assert cli.main(["check", str(record_dir), edited, edited]) == 1
+    assert capsys.readouterr().out.splitlines() == [
+        f"{edited}  finished  2 paths",
+        "  node 2 (under node 0, vs node 1): edited reply: "
+        "the harness sent back the model's reply with content changed",
+        "0 of 1 trajectories linear; forks: 1 edited reply",
+    ]
+
+
+def test_an_unreadable_record_is_reported_and_exits_2(tmp_path: Path, capsys: pytest.CaptureFixture) -> None:
+    Harness("tr_ok").write(tmp_path)
+    (tmp_path / "tr_bad.json.zst").write_bytes(b"not zstd")
+    future = {"format_version": 99, "id": "tr_new"}
+    (tmp_path / "tr_new.json.zst").write_bytes(zstandard.ZstdCompressor().compress(orjson.dumps(future)))
+
+    assert cli.main(["check", str(tmp_path)]) == 2
+    lines = capsys.readouterr().out.splitlines()
+    assert lines[0].startswith("tr_bad  unreadable: ZstdError: ")
+    assert lines[1:] == [
+        "tr_new  unreadable: format_version 99, this skycap reads 1",
+        "tr_ok  finished  empty",
+        "1 of 1 trajectories linear; 2 unreadable",
+    ]
+
+
+@pytest.mark.parametrize(
+    ("record_dir", "ids", "message"),
+    [
+        ("absent", [], "absent is not a directory"),
+        ("record", [], "no trajectories in"),
+        ("record", ["tr_x"], "no trajectory tr_x in"),
+        ("record", ["../tr_x"], "no trajectory ../tr_x in"),
+    ],
+)
+def test_usage_errors_exit_2(
+    tmp_path: Path, capsys: pytest.CaptureFixture, record_dir: str, ids: list[str], message: str
+) -> None:
+    (tmp_path / "record").mkdir()
+    Harness("tr_x").write(tmp_path)  # beside the record directory, not in it
+
+    with pytest.raises(SystemExit) as raised:
+        cli.main(["check", str(tmp_path / record_dir), *ids])
+    assert raised.value.code == 2
+    assert message in capsys.readouterr().err

From c2bada5ecfefbe793e4847e4ca1eb5e43c346c80 Mon Sep 17 00:00:00 2001
From: Jaden Fix <163031992+jadenfix@users.noreply.github.com>
Date: Fri, 9 Oct 2026 12:03:15 -0700
Subject: [PATCH 2/6] [skycap] skycap check: index siblings, scan only branch
 points, report non-object records

Forks are found under the roots and graph.branch_points() rather than
every node. Siblings are indexed by match hash, message, and the latest
(model) sibling per role, so only a client edit compares against every
earlier sibling, by a change count rather than the formatted list: 5,000
replies under one node take 0.05s. A record whose document isn't a JSON
object is reported unreadable, named by its file, instead of stopping
the run.
---
 skycap/src/skycap/check.py | 110 ++++++++++++++++++++++---------------
 skycap/tests/test_check.py |   4 +-
 2 files changed, 70 insertions(+), 44 deletions(-)

diff --git a/skycap/src/skycap/check.py b/skycap/src/skycap/check.py
index 259f974c8b..657eddac6f 100644
--- a/skycap/src/skycap/check.py
+++ b/skycap/src/skycap/check.py
@@ -25,8 +25,9 @@
 from __future__ import annotations
 
 from collections import Counter
-from collections.abc import Iterator, Mapping, Sequence
-from dataclasses import dataclass
+from collections.abc import Iterable, Iterator, Mapping, Sequence
+from dataclasses import dataclass, replace
+from itertools import islice
 from pathlib import Path
 from typing import Any, Literal
 
@@ -80,6 +81,8 @@ def linear(self) -> bool:
 
 def check_document(document: dict[str, Any]) -> Report:
     """Check one record document, as ``record.read_document`` returns it."""
+    if not isinstance(document, dict):
+        return Report("?", "unreadable", error=f"the document is a JSON {type(document).__name__}, not an object")
     version = document.get("format_version")
     if version != record.FORMAT_VERSION:
         return Report(
@@ -93,7 +96,7 @@ def check_document(document: dict[str, Any]) -> Report:
     normalize = _token_message if tokens else canonical_message
     # Each message as the graph's matching sees it, normalized once.
     messages = {node.id: normalize(node.message) for node in graph}
-    forks = [fork for parent in [None, *(node.id for node in graph)] for fork in _forks_under(graph, parent, messages)]
+    forks = [fork for parent in [None, *graph.branch_points()] for fork in _forks_under(graph, parent, messages)]
     return Report(
         trajectory_id=document["id"],
         status=document["status"],
@@ -112,7 +115,9 @@ def check_dir(record_dir: Path, trajectory_ids: Sequence[str] = ()) -> list[Repo
     reports = []
     for trajectory_id in trajectory_ids or list(record.list_ids(record_dir)):
         try:
-            reports.append(check_document(record.read_document(record_dir, trajectory_id)))
+            report = check_document(record.read_document(record_dir, trajectory_id))
+            # A document too malformed to name itself is named by its file.
+            reports.append(report if report.error is None else replace(report, trajectory_id=trajectory_id))
         except _READ_ERRORS as error:
             reports.append(Report(trajectory_id, "unreadable", error=f"{type(error).__name__}: {error}"))
     return reports
@@ -126,48 +131,61 @@ def _token_message(message: Mapping[str, Any]) -> dict[str, Any]:
 def _forks_under(graph: MessageGraph, parent: int | None, messages: dict[int, dict[str, Any]]) -> Iterator[Fork]:
     """One fork per child after the first: each started a path its earlier siblings don't share."""
     children = [graph.nodes[i] for i in graph.children(parent)]
-    by_match: dict[str, list[Node]] = {}
-    by_message: dict[bytes, Node] = {}
+    earlier = _Siblings(messages)
     for index, node in enumerate(children):
         if index:
-            sibling = _sibling_to_compare(children, index, by_match, by_message, messages)
+            sibling = earlier.closest(node, islice(children, index))
             kind, detail = _cause(sibling, node, messages)
             yield Fork(parent=parent, node=node.id, sibling=sibling.id, kind=kind, detail=detail)
-        by_match.setdefault(node.match_hash, []).append(node)
-        by_message[canonical_bytes(messages[node.id])] = node
-
-
-def _sibling_to_compare(
-    children: list[Node],
-    index: int,
-    by_match: dict[str, list[Node]],
-    by_message: dict[bytes, Node],
-    messages: dict[int, dict[str, Any]],
-) -> Node:
-    """The earlier sibling ``children[index]``'s history should have matched: the closest one.
-
-    One with the same match hash first (only the tokens or the sampling
-    differ), preferring a model reply, then one with the same message (only the
-    tools or model differ). Otherwise the same-role sibling, preferring a model
-    reply: for a model node the latest, since every model sibling is a resample
-    of it; for a client node the one with the fewest differing fields.
-    """
-    node, earlier = children[index], children[:index]
-    same_match = by_match.get(node.match_hash)
-    if same_match:
-        return ([s for s in same_match if s.author == "model"] or same_match)[-1]
-    same_message = by_message.get(canonical_bytes(messages[node.id]))
-    if same_message is not None:
-        return same_message
-    same_role = [s for s in earlier if s.role == node.role]
-    if not same_role:
-        return earlier[-1]
-    if node.author == "model":
-        return ([s for s in same_role if s.author == "model"] or same_role)[-1]
-    return min(
-        same_role,
-        key=lambda s: (len(_changes(messages[s.id], messages[node.id])), s.author != "model", -s.id),
-    )
+        earlier.add(node)
+
+
+class _Siblings:
+    """The siblings seen so far under one parent, indexed for ``closest``."""
+
+    def __init__(self, messages: dict[int, dict[str, Any]]) -> None:
+        self.messages = messages
+        self.latest: Node | None = None
+        self.by_match: dict[str, list[Node]] = {}
+        self.by_message: dict[bytes, Node] = {}
+        #: The latest sibling of each role, and the latest model reply of each role.
+        self.by_role: dict[str | None, Node] = {}
+        self.model_by_role: dict[str | None, Node] = {}
+
+    def add(self, node: Node) -> None:
+        self.latest = node
+        self.by_match.setdefault(node.match_hash, []).append(node)
+        self.by_message[canonical_bytes(self.messages[node.id])] = node
+        self.by_role[node.role] = node
+        if node.author == "model":
+            self.model_by_role[node.role] = node
+
+    def closest(self, node: Node, earlier: Iterable[Node]) -> Node:
+        """The earlier sibling ``node``'s history should have matched.
+
+        One with the same match hash first (only the tokens or the sampling
+        differ), preferring a model reply, then one with the same message (only
+        the tools or model differ). Otherwise the same-role sibling, preferring
+        a model reply: for a model node the latest, since every model sibling is
+        a resample of it; for a client node the one with the fewest differing
+        fields, the only case that compares against every earlier sibling.
+        """
+        same_match = self.by_match.get(node.match_hash)
+        if same_match:
+            return ([s for s in same_match if s.author == "model"] or same_match)[-1]
+        same_message = self.by_message.get(canonical_bytes(self.messages[node.id]))
+        if same_message is not None:
+            return same_message
+        if node.role not in self.by_role:
+            assert self.latest is not None
+            return self.latest
+        if node.author == "model":
+            return self.model_by_role.get(node.role) or self.by_role[node.role]
+        message = self.messages[node.id]
+        return min(
+            (s for s in earlier if s.role == node.role),
+            key=lambda s: (_count_changes(self.messages[s.id], message), s.author != "model", -s.id),
+        )
 
 
 def _cause(sibling: Node, node: Node, messages: dict[int, dict[str, Any]]) -> tuple[Kind, str]:
@@ -192,6 +210,11 @@ def _cause(sibling: Node, node: Node, messages: dict[int, dict[str, Any]]) -> tu
     return "different message", f"a different {node.role} message: {changes}"
 
 
+def _count_changes(before: dict[str, Any], after: dict[str, Any]) -> int:
+    """How many top-level fields differ (``len(_changes(...))`` without building the words)."""
+    return len(before.keys() ^ after.keys()) + sum(before[k] != after[k] for k in before.keys() & after.keys())
+
+
 def _changes(before: dict[str, Any], after: dict[str, Any]) -> list[str]:
     """Which top-level fields differ, e.g. ``["content changed", "reasoning_content dropped"]``."""
     changes = []
@@ -207,7 +230,8 @@ def _changes(before: dict[str, Any], after: dict[str, Any]) -> list[str]:
 
 def _a(role: str | None) -> str:
     word = role or "unnamed"
-    return f"{'an' if word[0] in 'aeio' else 'a'} {word}"
+    vowel = word[0].lower() in "aeio" or word.lower().startswith("un")
+    return f"{'an' if vowel else 'a'} {word}"
 
 
 def format_reports(reports: Sequence[Report]) -> str:
diff --git a/skycap/tests/test_check.py b/skycap/tests/test_check.py
index ac3c6d31aa..ac4c698299 100644
--- a/skycap/tests/test_check.py
+++ b/skycap/tests/test_check.py
@@ -275,14 +275,16 @@ def test_an_unreadable_record_is_reported_and_exits_2(tmp_path: Path, capsys: py
     (tmp_path / "tr_bad.json.zst").write_bytes(b"not zstd")
     future = {"format_version": 99, "id": "tr_new"}
     (tmp_path / "tr_new.json.zst").write_bytes(zstandard.ZstdCompressor().compress(orjson.dumps(future)))
+    (tmp_path / "tr_list.json.zst").write_bytes(zstandard.ZstdCompressor().compress(orjson.dumps([1, 2, 3])))
 
     assert cli.main(["check", str(tmp_path)]) == 2
     lines = capsys.readouterr().out.splitlines()
     assert lines[0].startswith("tr_bad  unreadable: ZstdError: ")
     assert lines[1:] == [
+        "tr_list  unreadable: the document is a JSON list, not an object",
         "tr_new  unreadable: format_version 99, this skycap reads 1",
         "tr_ok  finished  empty",
-        "1 of 1 trajectories linear; 2 unreadable",
+        "1 of 1 trajectories linear; 3 unreadable",
     ]
 
 

From 64075559503bb9ea9fda91ac4f9c4dc16ae85dc7 Mon Sep 17 00:00:00 2001
From: Jaden Fix <163031992+jadenfix@users.noreply.github.com>
Date: Fri, 9 Oct 2026 12:03:16 -0700
Subject: [PATCH 3/6] [skycap] skycap check: reuse MatchKey.fields and
 test_graph's harness

check normalizes each message with MatchKey.fields instead of its own copy
of it. test_check builds on test_graph's Harness and message helpers rather
than repeating them; test_graph's Harness.call takes `bridged` for it.
---
 skycap/src/skycap/check.py | 17 ++++++-----------
 skycap/tests/test_check.py | 26 ++++++--------------------
 skycap/tests/test_graph.py |  4 ++--
 3 files changed, 14 insertions(+), 33 deletions(-)

diff --git a/skycap/src/skycap/check.py b/skycap/src/skycap/check.py
index 657eddac6f..f537419c99 100644
--- a/skycap/src/skycap/check.py
+++ b/skycap/src/skycap/check.py
@@ -25,7 +25,7 @@
 from __future__ import annotations
 
 from collections import Counter
-from collections.abc import Iterable, Iterator, Mapping, Sequence
+from collections.abc import Iterable, Iterator, Sequence
 from dataclasses import dataclass, replace
 from itertools import islice
 from pathlib import Path
@@ -35,7 +35,7 @@
 
 from skycap import record
 from skycap.graph import MessageGraph, Node
-from skycap.hashing import canonical_bytes, canonical_message, rendered_fields
+from skycap.hashing import MatchKey, canonical_bytes
 
 Kind = Literal["resample", "edited reply", "re-rendered", "different message", "tools or model"]
 
@@ -92,10 +92,10 @@ def check_document(document: dict[str, Any]) -> Report:
         )
     graph = MessageGraph()
     record.add_nodes(graph, document)
-    tokens = (document.get("capture") or {}).get("mode") == "tokens"
-    normalize = _token_message if tokens else canonical_message
-    # Each message as the graph's matching sees it, normalized once.
-    messages = {node.id: normalize(node.message) for node in graph}
+    # Each message's fields as the graph's matching sees them. The key's tools and model don't
+    # matter here: ``fields`` ignores them, and the match hashes already cover them.
+    key = MatchKey("", None, rendered_only=(document.get("capture") or {}).get("mode") == "tokens")
+    messages = {node.id: key.fields(node.message) for node in graph}
     forks = [fork for parent in [None, *graph.branch_points()] for fork in _forks_under(graph, parent, messages)]
     return Report(
         trajectory_id=document["id"],
@@ -123,11 +123,6 @@ def check_dir(record_dir: Path, trajectory_ids: Sequence[str] = ()) -> list[Repo
     return reports
 
 
-def _token_message(message: Mapping[str, Any]) -> dict[str, Any]:
-    """A token-mode message as its match hash sees it: the rendered fields (``MatchKey.fields``)."""
-    return canonical_message(rendered_fields(message))
-
-
 def _forks_under(graph: MessageGraph, parent: int | None, messages: dict[int, dict[str, Any]]) -> Iterator[Fork]:
     """One fork per child after the first: each started a path its earlier siblings don't share."""
     children = [graph.nodes[i] for i in graph.children(parent)]
diff --git a/skycap/tests/test_check.py b/skycap/tests/test_check.py
index ac4c698299..56aae239bc 100644
--- a/skycap/tests/test_check.py
+++ b/skycap/tests/test_check.py
@@ -10,35 +10,21 @@
 
 from skycap import cli, record
 from skycap.check import check_dir, format_reports
-from skycap.graph import CallInfo
 from skycap.trajectory import Trajectory
+from tests import test_graph
 from tests.conftest import Stack, openai_client
 from tests.fake_renderer import END, NL, encode
+from tests.test_graph import SEARCH, SYS, assistant, user
 from tests.test_tokens import token_stack
 
-SYS = {"role": "system", "content": "You are terse."}
-SEARCH = [{"type": "function", "function": {"name": "search", "parameters": {}}}]
 
-
-def user(text: str) -> dict:
-    return {"role": "user", "content": text}
-
-
-def assistant(text: str, **extra) -> dict:
-    return {"role": "assistant", "content": text, **extra}
-
-
-class Harness:
-    """Drives one text-mode trajectory's graph a call at a time, then writes it."""
+class Harness(test_graph.Harness):
+    """``test_graph``'s harness, driving one trajectory's graph, which it then writes."""
 
     def __init__(self, trajectory_id: str) -> None:
+        super().__init__()
         self.trajectory = Trajectory(id=trajectory_id)
-        self.clock = 0.0
-
-    def call(self, messages, reply, *, tools=None, model="policy", bridged=None, **sampling) -> None:
-        self.clock += 1.0
-        info = CallInfo(t_start=self.clock, t_end=self.clock + 0.5, model=model, sampling=sampling, bridged=bridged)
-        self.trajectory.graph.commit_text(messages, reply, tools=tools, model=model, call=info)
+        self.graph = self.trajectory.graph
 
     def write(self, record_dir: Path) -> None:
         self.trajectory.seal("finished", {"reward": 1.0})
diff --git a/skycap/tests/test_graph.py b/skycap/tests/test_graph.py
index 3da2b69ca5..42f05362d8 100644
--- a/skycap/tests/test_graph.py
+++ b/skycap/tests/test_graph.py
@@ -30,9 +30,9 @@ def __init__(self) -> None:
         self.graph = MessageGraph()
         self.clock = 0.0
 
-    def call(self, messages, reply, *, tools=None, model="policy", **sampling):
+    def call(self, messages, reply, *, tools=None, model="policy", bridged=None, **sampling):
         self.clock += 1.0
-        info = CallInfo(t_start=self.clock, t_end=self.clock + 0.5, model=model, sampling=sampling)
+        info = CallInfo(t_start=self.clock, t_end=self.clock + 0.5, model=model, sampling=sampling, bridged=bridged)
         return self.graph.commit_text(messages, reply, tools=tools, model=model, call=info)
 
     def shape(self) -> list[tuple[int | None, str, str]]:

From 364fdb66281108db5401a67370386ea20a2870d7 Mon Sep 17 00:00:00 2001
From: Jaden Fix <163031992+jadenfix@users.noreply.github.com>
Date: Fri, 9 Oct 2026 12:03:17 -0700
Subject: [PATCH 4/6] [skycap] skycap check: pick the compared sibling with one
 min() instead of an index

_Siblings kept five indexes to find the sibling a fork is compared with, and
_count_changes repeated _changes to count without formatting. One ranked
min() over the earlier siblings picks the same sibling in every test. It is
as fast up to hundreds of siblings under one node (0.07s at 500); only
thousands of model replies under one node are slower (7s at 5,000, from
0.09s). Client edits were already quadratic under the index (15s at 5,000).
---
 skycap/src/skycap/check.py | 95 +++++++++++++-------------------------
 1 file changed, 32 insertions(+), 63 deletions(-)

diff --git a/skycap/src/skycap/check.py b/skycap/src/skycap/check.py
index f537419c99..30dd3d9148 100644
--- a/skycap/src/skycap/check.py
+++ b/skycap/src/skycap/check.py
@@ -6,7 +6,8 @@
 with the sibling it should have matched says why:
 
 * ``resample``: the model replied again to a history it had already answered
-  (a retry, ``n > 1``, or a reply to a history the harness cut short).
+  (a retry, another sample of the same turn, or a reply to a history the harness
+  cut short).
 * ``edited reply``: the harness sent back a model reply with fields changed,
   such as stripped reasoning or a repaired tool call.
 * ``re-rendered``: token mode only. The same message, but its re-render didn't
@@ -25,9 +26,8 @@
 from __future__ import annotations
 
 from collections import Counter
-from collections.abc import Iterable, Iterator, Sequence
+from collections.abc import Iterator, Sequence
 from dataclasses import dataclass, replace
-from itertools import islice
 from pathlib import Path
 from typing import Any, Literal
 
@@ -35,7 +35,7 @@
 
 from skycap import record
 from skycap.graph import MessageGraph, Node
-from skycap.hashing import MatchKey, canonical_bytes
+from skycap.hashing import MatchKey
 
 Kind = Literal["resample", "edited reply", "re-rendered", "different message", "tools or model"]
 
@@ -126,62 +126,36 @@ def check_dir(record_dir: Path, trajectory_ids: Sequence[str] = ()) -> list[Repo
 def _forks_under(graph: MessageGraph, parent: int | None, messages: dict[int, dict[str, Any]]) -> Iterator[Fork]:
     """One fork per child after the first: each started a path its earlier siblings don't share."""
     children = [graph.nodes[i] for i in graph.children(parent)]
-    earlier = _Siblings(messages)
-    for index, node in enumerate(children):
-        if index:
-            sibling = earlier.closest(node, islice(children, index))
-            kind, detail = _cause(sibling, node, messages)
-            yield Fork(parent=parent, node=node.id, sibling=sibling.id, kind=kind, detail=detail)
-        earlier.add(node)
-
-
-class _Siblings:
-    """The siblings seen so far under one parent, indexed for ``closest``."""
-
-    def __init__(self, messages: dict[int, dict[str, Any]]) -> None:
-        self.messages = messages
-        self.latest: Node | None = None
-        self.by_match: dict[str, list[Node]] = {}
-        self.by_message: dict[bytes, Node] = {}
-        #: The latest sibling of each role, and the latest model reply of each role.
-        self.by_role: dict[str | None, Node] = {}
-        self.model_by_role: dict[str | None, Node] = {}
-
-    def add(self, node: Node) -> None:
-        self.latest = node
-        self.by_match.setdefault(node.match_hash, []).append(node)
-        self.by_message[canonical_bytes(self.messages[node.id])] = node
-        self.by_role[node.role] = node
-        if node.author == "model":
-            self.model_by_role[node.role] = node
-
-    def closest(self, node: Node, earlier: Iterable[Node]) -> Node:
-        """The earlier sibling ``node``'s history should have matched.
-
-        One with the same match hash first (only the tokens or the sampling
-        differ), preferring a model reply, then one with the same message (only
-        the tools or model differ). Otherwise the same-role sibling, preferring
-        a model reply: for a model node the latest, since every model sibling is
-        a resample of it; for a client node the one with the fewest differing
-        fields, the only case that compares against every earlier sibling.
-        """
-        same_match = self.by_match.get(node.match_hash)
-        if same_match:
-            return ([s for s in same_match if s.author == "model"] or same_match)[-1]
-        same_message = self.by_message.get(canonical_bytes(self.messages[node.id]))
-        if same_message is not None:
-            return same_message
-        if node.role not in self.by_role:
-            assert self.latest is not None
-            return self.latest
-        if node.author == "model":
-            return self.model_by_role.get(node.role) or self.by_role[node.role]
-        message = self.messages[node.id]
-        return min(
-            (s for s in earlier if s.role == node.role),
-            key=lambda s: (_count_changes(self.messages[s.id], message), s.author != "model", -s.id),
+    for index, node in enumerate(children[1:], 1):
+        sibling = _closest(node, children[:index], messages)
+        kind, detail = _cause(sibling, node, messages)
+        yield Fork(parent=parent, node=node.id, sibling=sibling.id, kind=kind, detail=detail)
+
+
+def _closest(node: Node, earlier: list[Node], messages: dict[int, dict[str, Any]]) -> Node:
+    """The earlier sibling ``node``'s history should have matched.
+
+    One with the same match hash (only the tokens or the sampling differ), else
+    the same message (only the tools or model differ), else the same role: for
+    a client node the one with the fewest changed fields. Ties go to a model
+    reply, then to the latest.
+    """
+    message = messages[node.id]
+
+    def rank(sibling: Node) -> tuple:
+        before = messages[sibling.id]
+        same_role = sibling.role == node.role
+        return (
+            sibling.match_hash != node.match_hash,
+            before != message,
+            not same_role,
+            len(_changes(before, message)) if same_role and node.author == "client" else 0,
+            sibling.author != "model",
+            -sibling.id,
         )
 
+    return min(earlier, key=rank)
+
 
 def _cause(sibling: Node, node: Node, messages: dict[int, dict[str, Any]]) -> tuple[Kind, str]:
     before, after = messages[sibling.id], messages[node.id]
@@ -205,11 +179,6 @@ def _cause(sibling: Node, node: Node, messages: dict[int, dict[str, Any]]) -> tu
     return "different message", f"a different {node.role} message: {changes}"
 
 
-def _count_changes(before: dict[str, Any], after: dict[str, Any]) -> int:
-    """How many top-level fields differ (``len(_changes(...))`` without building the words)."""
-    return len(before.keys() ^ after.keys()) + sum(before[k] != after[k] for k in before.keys() & after.keys())
-
-
 def _changes(before: dict[str, Any], after: dict[str, Any]) -> list[str]:
     """Which top-level fields differ, e.g. ``["content changed", "reasoning_content dropped"]``."""
     changes = []

From 2946ea2fd72f7699d522b997046a10808aea7adf Mon Sep 17 00:00:00 2001
From: Jaden Fix <163031992+jadenfix@users.noreply.github.com>
Date: Fri, 9 Oct 2026 12:03:18 -0700
Subject: [PATCH 5/6] [skycap] skycap check: a record whose fields have the
 wrong shape is unreadable, not fatal

A document that is a JSON object but whose capture is a list raised
AttributeError, which stopped the whole run instead of reporting that one
record. It is now a read error like the others.
---
 skycap/src/skycap/check.py | 2 +-
 skycap/tests/test_check.py | 5 ++++-
 2 files changed, 5 insertions(+), 2 deletions(-)

diff --git a/skycap/src/skycap/check.py b/skycap/src/skycap/check.py
index 30dd3d9148..3e8f04fcc0 100644
--- a/skycap/src/skycap/check.py
+++ b/skycap/src/skycap/check.py
@@ -76,7 +76,7 @@ def linear(self) -> bool:
 
 
 #: The errors a malformed or foreign record raises while it is read.
-_READ_ERRORS = (OSError, ValueError, KeyError, TypeError, AssertionError, zstandard.ZstdError)
+_READ_ERRORS = (OSError, ValueError, KeyError, TypeError, AttributeError, AssertionError, zstandard.ZstdError)
 
 
 def check_document(document: dict[str, Any]) -> Report:
diff --git a/skycap/tests/test_check.py b/skycap/tests/test_check.py
index 56aae239bc..fdc1876ed6 100644
--- a/skycap/tests/test_check.py
+++ b/skycap/tests/test_check.py
@@ -262,6 +262,8 @@ def test_an_unreadable_record_is_reported_and_exits_2(tmp_path: Path, capsys: py
     future = {"format_version": 99, "id": "tr_new"}
     (tmp_path / "tr_new.json.zst").write_bytes(zstandard.ZstdCompressor().compress(orjson.dumps(future)))
     (tmp_path / "tr_list.json.zst").write_bytes(zstandard.ZstdCompressor().compress(orjson.dumps([1, 2, 3])))
+    shape = {"format_version": 1, "id": "tr_shape", "tools": {}, "nodes": [], "capture": ["tokens"]}
+    (tmp_path / "tr_shape.json.zst").write_bytes(zstandard.ZstdCompressor().compress(orjson.dumps(shape)))
 
     assert cli.main(["check", str(tmp_path)]) == 2
     lines = capsys.readouterr().out.splitlines()
@@ -270,7 +272,8 @@ def test_an_unreadable_record_is_reported_and_exits_2(tmp_path: Path, capsys: py
         "tr_list  unreadable: the document is a JSON list, not an object",
         "tr_new  unreadable: format_version 99, this skycap reads 1",
         "tr_ok  finished  empty",
-        "1 of 1 trajectories linear; 3 unreadable",
+        "tr_shape  unreadable: AttributeError: 'list' object has no attribute 'get'",
+        "1 of 1 trajectories linear; 4 unreadable",
     ]
 
 

From b16fc844074b58e12f53177218c4b869be357093 Mon Sep 17 00:00:00 2001
From: Jaden Fix <163031992+jadenfix@users.noreply.github.com>
Date: Fri, 9 Oct 2026 12:03:19 -0700
Subject: [PATCH 6/6] [skycap] test_check: fold the subagent root test into the
 tool-set test that already builds it

---
 skycap/tests/test_check.py | 20 ++++++--------------
 1 file changed, 6 insertions(+), 14 deletions(-)

diff --git a/skycap/tests/test_check.py b/skycap/tests/test_check.py
index fdc1876ed6..35ac8fc7ca 100644
--- a/skycap/tests/test_check.py
+++ b/skycap/tests/test_check.py
@@ -118,25 +118,17 @@ def test_compaction_is_a_different_message(tmp_path: Path) -> None:
     )
 
 
-def test_a_subagent_with_its_own_system_prompt_is_a_new_root(tmp_path: Path) -> None:
-    h = Harness("tr")
-    h.call([SYS, user("task")], assistant("delegating"))
-    h.call([{"role": "system", "content": "You are a searcher."}, user("find x")], assistant("found"))
-
-    assert report_lines(tmp_path, h)[1] == (
-        "  node 3 (new root, vs node 0): different message: a different system message: content changed"
-    )
-
-
-def test_a_changed_tool_set_is_compared_with_the_root_it_repeats(tmp_path: Path) -> None:
+def test_a_subagent_and_a_changed_tool_set_start_new_roots(tmp_path: Path) -> None:
     h = Harness("tr")
     h.call([SYS, user("task")], assistant("plan"))
     h.call([{"role": "system", "content": "You are a searcher."}, user("find x")], assistant("found"))
     h.call([SYS, user("task"), assistant("plan"), user("go")], assistant("done"), tools=SEARCH)
 
-    assert report_lines(tmp_path, h)[2] == (
-        "  node 6 (new root, vs node 0): tools or model: the same message, sent with a different tool set or model"
-    )
+    # The tool change repeats the first root's message, so it is compared with that root, not the latest.
+    assert report_lines(tmp_path, h)[1:3] == [
+        "  node 3 (new root, vs node 0): different message: a different system message: content changed",
+        "  node 6 (new root, vs node 0): tools or model: the same message, sent with a different tool set or model",
+    ]
 
 
 def test_a_message_of_another_role_names_both(tmp_path: Path) -> None: