feat(loopmetrics): session inputs note, truncation sidecar, and usage dedupe that survives resumed sessions

tools/loopmetrics/INPUTS.md

@@ -0,0 +1,65 @@
+---
+owner: @esengine
+backup: @SivanCola
+status: active
+reviewed: 2026-10-22
+---
+
+# loopmetrics inputs: from a session to the analyzer
+
+How to feed a real session, such as a user-reported loop, to
+`loopmetrics.py`. It covers what each source holds, which metrics it can
+answer, and what must be checked first. Nothing here contacts anyone: a
+session file is analysed only after the owner has it.
+
+## Sources the analyzer reads
+
+| source | where it comes from | shape |
+| --- | --- | --- |
+| trajectory | `reasonix run --trajectory FILE` | one JSON object per line, `{"event": {...}}` |
+| session wire log | `*.wire.jsonl` next to the session file under `~/.reasonix` (Studio and `serve` write it) | one flat event per line |
+
+## What each source can answer
+
+| metric | trajectory | wire log |
+| --- | --- | --- |
+| M1 redundant tool calls | yes | yes |
+| M2 reasoning repetition | per reasoning block | per message only, printed `source: message` |
+| M3 redo after continue | yes, with two files | not supported: no split at the stop |
+| M4 repeated short replies | yes | no value: the log has no `text` events |
+| M5 cost covariates | yes | yes |
+
+A wire log keeps only the event kinds its pane draws (the allow-list in
+`internal/frontend/serve/wirelog.go`). Streamed reasoning and text arrive in
+chunks and are not kept, so a wire log can never show a within-reply loop at
+block level. M4 prints `-` for it: that is "not measurable", not "no repeat".
+
+## Check the file is complete
+
+- A wire log stops growing at 8 MiB and drops later frames; it never truncates
+  earlier ones, so what is on disk is a prefix of the session.
+- The sidecar `<session>.wire.meta.json` says so with `"truncated": true`. The
+  analyzer prints `wire_truncated` (and lists truncated logs in a cell). `null`
+  means there is no sidecar: nothing is known, which is not "complete".
+- A session that resumed in a new process repeats attempt ids (`sa-1-1`).
+  Usage is therefore deduplicated on the whole usage payload, not on the id.
+  In the 65 local wire logs 575 of 1,039 usage events shared an id with
+  another event.
+
+## Steps
+
+1. Keep the file local. It holds prompts, tool output and paths; do not paste
+   it into an issue or commit it. The analyzer prints aggregates only.
+2. Run `python3 tools/loopmetrics/loopmetrics.py run FILE --json` and read
+   `wire_truncated`, `turns`, `m1.unresolved` and `m2.source` first.
+3. Read M1 and M2 against `experiments/phase1/results.md`. In 60 runs on
+   short read-only work no cell had a redundant call, a cross-block overlap
+   above 0.314 or a within-block repeat above 0.287.
+4. State what the file cannot show (the tables above) next to any number.
+
+## What one session is
+
+One session is an observation, not a result: a cell needs 5 runs of one
+condition. A reported loop that shows nothing by these definitions is also
+information. M1 cannot see a loop that varies its arguments each round, and M2
+is character overlap, so a paraphrased loop is under-counted.

tools/loopmetrics/loopmetrics.py

@@ -144,11 +144,14 @@ def reasoning(turn):
 
 
 def replies(events):
+    """Longest run of identical short replies; None when the stream has no text events."""
     best = run = 0
     prev = None
+    seen = False
     for _, ev in events:
         if ev.get("kind") != "text":
             continue
+        seen = True
         norm = normalise(ev.get("text") or "")
         if 0 < len(norm) <= REPLY_CAP and norm == prev:
             run += 1
@@ -158,7 +161,7 @@ def replies(events):
             run = 0
         prev = norm
         best = max(best, run)
-    return best
+    return best if seen else None
 
 
 def usage(events):
@@ -167,7 +170,7 @@ def usage(events):
         u = ev.get("usage")
         if ev.get("kind") != "usage" or not u:
             continue
-        key = u.get("attemptId") or ("seq", seq)
+        key = (u.get("attemptId"), hashlib.sha1(json.dumps(u, sort_keys=True).encode()).hexdigest())
         if key in seen:
             continue
         seen.add(key)
@@ -219,7 +222,8 @@ def merge(parts):
     m2["cross_median"], m2["cross_max"] = med(cross), max(cross, default=None)
     m2["within_flag_share"] = {str(f): share(within, f) for f in WITHIN_FLAGS}
     m2["cross_flag_share"] = {str(f): share(cross, f) for f in JACCARD_FLAGS}
-    out["m4"]["max_run"] = max(p["m4"]["max_run"] for p in parts)
+    runs4 = [p["m4"]["max_run"] for p in parts if p["m4"]["max_run"] is not None]
+    out["m4"]["max_run"] = max(runs4) if runs4 else None
     for k in ("requests", "prompt", "completion", "cache_hit", "cost"):
         out["m5"][k] = sum(p["m5"][k] for p in parts)
     if "redundant_without_ignore" in parts[0]["m1"]:
@@ -297,6 +301,24 @@ def load_run(spec):
     return legs if len(legs) > 1 else legs[0]
 
 
+def wire_truncated(spec):
+    """True when a session wire log's sidecar says the size cap dropped frames.
+
+    None when the file has no sidecar: nothing is known, which is not "complete".
+    """
+    for path in spec.split(","):
+        if not path.endswith(".wire.jsonl"):
+            continue
+        meta = path[: -len(".wire.jsonl")] + ".wire.meta.json"
+        try:
+            with open(meta, encoding="utf-8") as fh:
+                if json.load(fh).get("truncated"):
+                    return True
+        except (OSError, ValueError):
+            return None
+    return False if any(p.endswith(".wire.jsonl") for p in spec.split(",")) else None
+
+
 def cell_summary(name, paths, ignore=()):
     runs, failed = [], []
     for p in paths:
@@ -306,6 +328,7 @@ def cell_summary(name, paths, ignore=()):
             failed.append(f"{p}: {e}")
     out = {
         "cell": name, "runs": len(runs), "failed": failed,
+        "truncated_wire_logs": [p for p in paths if wire_truncated(p)],
         "verdict": "ok" if len(runs) >= MIN_RUNS else "not a result",
         "series": {},
     }
@@ -321,6 +344,8 @@ def fmt(v):
 
 def print_cell(s):
     print(f"cell {s['cell']}: {s['runs']} runs, {len(s['failed'])} failed -> {s['verdict']}")
+    for p in s["truncated_wire_logs"]:
+        print(f"  truncated wire log (a prefix of the session): {p}")
     for f in s["failed"]:
         print(f"  failed: {f}")
     for label, v in s["series"].items():
@@ -348,6 +373,7 @@ def main(argv=None):
             m = run_metrics(load_run(a.file), tuple(a.ignore_arg))
             for k in ("_calls", "_within", "_cross"):
                 m.pop(k)
+            m["wire_truncated"] = wire_truncated(a.file)
             print(json.dumps(m, indent=2, ensure_ascii=False) if a.json else json.dumps(m, ensure_ascii=False))
         elif a.cmd == "cell":
             summary = cell_summary(a.name, a.files, tuple(a.ignore_arg))

tools/loopmetrics/test_loopmetrics.py

@@ -217,6 +217,14 @@ class Replies(unittest.TestCase):
         self.assertEqual(metrics(s)["m4"]["max_run"], 2)
 
 
+class NoTextEvents(unittest.TestCase):
+    def test_a_stream_without_text_events_has_no_m4_value_not_a_zero(self):
+        self.assertIsNone(metrics(Stream().turn().call("c1"))["m4"]["max_run"])
+
+    def test_a_stream_with_text_events_and_no_repeat_is_zero_or_one(self):
+        self.assertEqual(metrics(Stream().turn().say("a long enough reply that is over the cap"))["m4"]["max_run"], 0)
+
+
 class Resume(unittest.TestCase):
     def first(self):
         s = Stream().turn()
@@ -250,6 +258,34 @@ class Cost(unittest.TestCase):
         self.assertEqual((m["requests"], m["completion"]), (2, 10))
 
 
+class Inputs(unittest.TestCase):
+    def test_two_requests_that_share_an_attempt_id_are_both_counted(self):
+        s = Stream().turn().usage("sa-1-1", completion=5).usage("sa-1-1", completion=9)
+        m = metrics(s)["m5"]
+        self.assertEqual((m["requests"], m["completion"]), (2, 14))
+
+    def test_an_identical_usage_event_emitted_twice_counts_once(self):
+        s = Stream().turn().usage("sa-1-1").usage("sa-1-1")
+        self.assertEqual(metrics(s)["m5"]["requests"], 1)
+
+    def test_a_wire_log_whose_sidecar_says_truncated_is_reported_as_a_prefix(self):
+        with tempfile.TemporaryDirectory() as d:
+            log = Stream().turn().write(d, "s.wire.jsonl")
+            self.assertIsNone(lm.wire_truncated(log))
+            with open(os.path.join(d, "s.wire.meta.json"), "w") as fh:
+                json.dump({"truncated": True}, fh)
+            self.assertTrue(lm.wire_truncated(log))
+            self.assertEqual(lm.cell_summary("x", [log])["truncated_wire_logs"], [log])
+
+    def test_a_complete_wire_log_and_a_trajectory_are_not_reported_truncated(self):
+        with tempfile.TemporaryDirectory() as d:
+            log = Stream().turn().write(d, "s.wire.jsonl")
+            with open(os.path.join(d, "s.wire.meta.json"), "w") as fh:
+                json.dump({"truncated": False}, fh)
+            self.assertFalse(lm.wire_truncated(log))
+            self.assertIsNone(lm.wire_truncated(Stream().turn().write(d, "t.jsonl")))
+
+
 class Legs(unittest.TestCase):
     def test_usage_in_two_processes_is_not_deduplicated_across_them(self):
         a = Stream().turn().usage("sa-1-1")