[skycap] skycap check: report whether a record stays on one path, and why each fork happened

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

skycap/src/skycap/check.py

@@ -0,0 +1,223 @@
+"""`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, 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
+  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, Sequence
+from dataclasses import dataclass, replace
+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 MatchKey
+
+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, AttributeError, AssertionError, zstandard.ZstdError)
+
+
+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(
+            document.get("id", "?"),
+            "unreadable",
+            error=f"format_version {version}, this skycap reads {record.FORMAT_VERSION}",
+        )
+    graph = MessageGraph()
+    record.add_nodes(graph, document)
+    # 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"],
+        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:
+            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
+
+
+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)]
+    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]
+    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"
+    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:
+    """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)

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())

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: