integrations: Chronicle record/replay + unbiased overhead benchmark

.gitignore

@@ -19,3 +19,6 @@ env/
 # Local
 *.log
 .env
+
+# Chronicle integration demo output (regenerated)
+integrations/chronicle/.fixtures/

integrations/chronicle/README.md

@@ -0,0 +1,66 @@
+# Chronicle integration (record and replay a testbench MAS)
+
+This folder attaches [chronicle](https://github.com/theagentplane/chronicle) to the
+testbench **from the outside**. Nothing in `src/testbench` imports chronicle (a CI test
+enforces that), so this code lives here under `integrations/` instead, where it is free
+to import both.
+
+## The one idea
+
+Every testbench agent depends on a single model seam: `LLMClient.complete(model, messages)`.
+Chronicle wraps that seam. On a live run it writes down each call; on replay it hands the
+recorded answer straight back instead of calling the model.
+
+```
+agent -> ChronicleClient.complete(...) -> real client   (RECORD: also writes an Envelope)
+agent -> ChronicleClient.complete(...) -> recorded answer (REPLAY: no real client call)
+```
+
+We wrap the client, not the event sink, because replay has to **return** a recorded value
+into the agent. The event sink only observes; it cannot feed a value back.
+
+## Terms (same as chronicle)
+
+- **Boundary**: one model call. Here the boundary name is the agent role (`supervisor`,
+  `researcher`, `analyst`, `writer`), read from the `ROLE:` system message.
+- **Envelope**: the record of one boundary crossing: its input messages, its output text,
+  token usage, and model. It records that I/O, not any side effects.
+- **Fixture**: the committed folder of Envelopes for one run (`graph.json` + one JSON per
+  boundary).
+- **Stub / live**: on replay a *stubbed* boundary returns its recorded output without
+  running; a *live* boundary runs the real client. A **cut-point** is the one boundary you
+  set live to test a change while the rest stays stubbed.
+
+## Run the demos
+
+Both MAS use the *same* `ChronicleClient` with no changes.
+
+```bash
+python integrations/chronicle/demo.py        # orchestrator-workers (linear, 4 calls)
+python integrations/chronicle/demo_loop.py   # evaluator-optimizer (loops, 6 calls)
+```
+
+Each prints three phases and how many times the real model client was actually called:
+
+```
+1) RECORD                real model calls: 4/6   (fixture written)
+2) REPLAY (stub all)     real model calls: 0     (identical result)
+3) CUT-POINT             real model calls: 1     (one boundary ran live)
+```
+
+The loop demo cut-points `generator@2` (the second draft only), which shows how chronicle
+tells repeated calls apart by **invocation index**.
+
+## Run the tests
+
+```bash
+python -m pytest integrations/chronicle/test_replay_regression.py -q
+```
+
+## Files
+
+- `chronicle_client.py`       : the `ChronicleClient` adapter (wraps any `LLMClient`).
+- `demo.py`                   : orchestrator-workers: record, replay, cut-point.
+- `demo_loop.py`              : evaluator-optimizer: same, on a looping MAS.
+- `test_replay_regression.py` : the phases above as assertions, for both MAS.
+- `.fixtures/`                : demo output, gitignored (regenerated on each run).

integrations/chronicle/benchmark.py

@@ -0,0 +1,209 @@
+"""Unbiased recording-overhead and determinism measurement for Chronicle, run on the
+neutral testbench MAS (which imports nothing from Chronicle).
+
+Because the workload does not know Chronicle exists (a CI test enforces that), the
+overhead measured here cannot be tuned to flatter the tool. This is the number to cite
+for recording overhead on a realistic multi-agent system, alongside the fault-detection
+numbers from Chronicle's own incident benchmark.
+
+    python integrations/chronicle/benchmark.py
+    python integrations/chronicle/benchmark.py --json out.json --tex table.tex
+
+Reports, for each MAS:
+- recording overhead (in-memory) in microseconds per crossing and as a fraction of a
+  typical model call,
+- deterministic replay (N replays reproduce the run) with zero real model calls.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import statistics
+import sys
+import tempfile
+from dataclasses import asdict, dataclass
+from pathlib import Path
+from time import perf_counter
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import chronicle
+from chronicle import ReplayPlan
+
+from chronicle_client import ChronicleClient
+from testbench.core import LLMClient, ModelResponse, RunConfig, make_client
+from testbench.evaluator_optimizer.loop import refine
+from testbench.orchestrator_workers.graph import build_graph, initial_state
+
+MODEL_LATENCY_MS = 300.0
+REPS = 150
+ROUNDS = 7
+REPLAYS = 20
+
+ORCH_TASK = "What is token governance and why does it matter for agents?"
+LOOP_TASK = "Explain token governance to a new engineer in two sentences."
+
+
+class _Spy:
+    """Counts how many calls actually reach the real client."""
+
+    def __init__(self, inner: LLMClient) -> None:
+        self.inner = inner
+        self.calls = 0
+
+    def complete(self, model, messages, *, max_output_tokens=None) -> ModelResponse:
+        self.calls += 1
+        return self.inner.complete(model, messages, max_output_tokens=max_output_tokens)
+
+
+def _prep_orch(client: LLMClient):
+    # Compile the LangGraph once, so timing measures the run, not graph construction.
+    graph = build_graph(client, sink=None, cfg=RunConfig())
+    return lambda: graph.invoke(initial_state(ORCH_TASK))["brief"]
+
+
+def _prep_loop(client: LLMClient):
+    return lambda: (
+        lambda r: f"{r.answer}|{r.score}|{r.iterations}"
+    )(refine(LOOP_TASK, client, sink=None, cfg=RunConfig()))
+
+
+# name -> (prepare(client) -> zero-arg runnable returning a stable outcome, crossings)
+WORKLOADS = {
+    "Orchestrator": (_prep_orch, 4),
+    "Loop": (_prep_loop, 6),
+}
+
+
+def _best_time(fn, reps: int = REPS, rounds: int = ROUNDS) -> float:
+    for _ in range(min(reps, 30)):
+        fn()
+    best = float("inf")
+    for _ in range(rounds):
+        start = perf_counter()
+        for _ in range(reps):
+            fn()
+        best = min(best, (perf_counter() - start) / reps)
+    return best
+
+
+@dataclass
+class Result:
+    mas: str
+    crossings: int
+    t_base_us: float
+    t_rec_us: float
+    overhead_us_per_crossing: float
+    overhead_pct_at_model: float
+    deterministic: bool
+    replay_model_calls: int
+
+
+def measure(name: str, prepare, crossings: int) -> Result:
+    # Baseline: the plain client, no Chronicle in the path. Graph built once.
+    base_run = prepare(make_client(RunConfig()))
+    t_base = _best_time(base_run)
+
+    # Recording: ChronicleClient, in-memory (store=None) to isolate compute from disk.
+    rec_run = prepare(ChronicleClient(make_client(RunConfig())))
+
+    def rec() -> None:
+        with chronicle.record(name, store=None):
+            rec_run()
+
+    t_rec = _best_time(rec)
+
+    # Determinism + zero model calls: record once, replay REPLAYS times.
+    with tempfile.TemporaryDirectory() as tmp:
+        fixture = str(Path(tmp) / name)
+        spy = _Spy(make_client(RunConfig()))
+        run = prepare(ChronicleClient(spy))
+        with chronicle.record(name, export=fixture):
+            first = run()
+        outcomes, replay_calls = [], 0
+        for i in range(REPLAYS):
+            spy.calls = 0
+            with chronicle.replay_trace(fixture, ReplayPlan()):
+                outcomes.append(run())
+            if i == 0:
+                replay_calls = spy.calls
+        deterministic = all(o == first for o in outcomes)
+
+    per_crossing_us = (t_rec - t_base) * 1e6 / crossings if crossings else 0.0
+    return Result(
+        mas=name,
+        crossings=crossings,
+        t_base_us=t_base * 1e6,
+        t_rec_us=t_rec * 1e6,
+        overhead_us_per_crossing=per_crossing_us,
+        overhead_pct_at_model=per_crossing_us / (MODEL_LATENCY_MS * 1000.0) * 100.0,
+        deterministic=deterministic,
+        replay_model_calls=replay_calls,
+    )
+
+
+def run_all() -> list[Result]:
+    return [measure(name, prep, crossings) for name, (prep, crossings) in WORKLOADS.items()]
+
+
+def print_table(results: list[Result]) -> None:
+    print()
+    print(f"  {'MAS':<13} {'cross':>5} {'t_base':>9} {'t_rec':>9} {'rec/cross':>10} "
+          f"{'%@model':>8} {'determ':>7} {'replay calls':>13}")
+    print("  " + "-" * 82)
+    for r in results:
+        print(f"  {r.mas:<13} {r.crossings:>5} {r.t_base_us:>8.1f}u {r.t_rec_us:>8.1f}u "
+              f"{r.overhead_us_per_crossing:>9.1f}u {r.overhead_pct_at_model:>7.3f}% "
+              f"{'yes' if r.deterministic else 'NO':>7} {r.replay_model_calls:>13}")
+    print("  " + "-" * 82)
+    med = statistics.median(r.overhead_us_per_crossing for r in results)
+    med_pct = statistics.median(r.overhead_pct_at_model for r in results)
+    calls = sum(r.replay_model_calls for r in results)
+    print(f"\n  unbiased recording overhead : {med:.1f} us/crossing "
+          f"(= {med_pct:.3f}% of a {MODEL_LATENCY_MS:.0f} ms model call)")
+    print(f"  replay model calls          : {calls}  (across all workloads)")
+    print(f"  deterministic replay        : {'yes' if all(r.deterministic for r in results) else 'NO'}"
+          f"  ({REPLAYS} replays each)\n")
+
+
+def to_latex(results: list[Result]) -> str:
+    med = statistics.median(r.overhead_us_per_crossing for r in results)
+    med_pct = statistics.median(r.overhead_pct_at_model for r in results)
+    lines = [
+        "% Auto-generated by integrations/chronicle/benchmark.py (neutral testbench)",
+        "\\newcommand{\\unbiasedoverheadus}{%.1f}" % med,
+        "\\newcommand{\\unbiasedoverheadpct}{%.3f}" % med_pct,
+        "",
+        "\\begin{tabular}{lrrr}",
+        "\\toprule",
+        "MAS & Crossings & Rec./cross. (\\textmu s) & Overhead \\\\",
+        "\\midrule",
+    ]
+    for r in results:
+        lines.append(
+            f"{r.mas} & {r.crossings} & {r.overhead_us_per_crossing:.1f} & "
+            f"{r.overhead_pct_at_model:.3f}\\% \\\\"
+        )
+    lines += ["\\bottomrule", "\\end{tabular}"]
+    return "\n".join(lines)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="Unbiased Chronicle overhead on the testbench")
+    parser.add_argument("--json", type=Path)
+    parser.add_argument("--tex", type=Path)
+    args = parser.parse_args()
+
+    results = run_all()
+    print_table(results)
+    if args.json:
+        args.json.write_text(json.dumps([asdict(r) for r in results], indent=2), encoding="utf-8")
+        print(f"  wrote {args.json}")
+    if args.tex:
+        args.tex.write_text(to_latex(results), encoding="utf-8")
+        print(f"  wrote {args.tex}")
+
+
+if __name__ == "__main__":
+    main()