diff --git a/tools/loopmetrics/INPUTS.md b/tools/loopmetrics/INPUTS.md new file mode 100644 index 0000000000..4d9443f22a --- /dev/null +++ b/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 `.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. diff --git a/tools/loopmetrics/loopmetrics.py b/tools/loopmetrics/loopmetrics.py index ddb35e24ce..ccc77f5a63 100644 --- a/tools/loopmetrics/loopmetrics.py +++ b/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)) diff --git a/tools/loopmetrics/test_loopmetrics.py b/tools/loopmetrics/test_loopmetrics.py index 039df307d3..4133c585cb 100644 --- a/tools/loopmetrics/test_loopmetrics.py +++ b/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")