headroom v2.0.8: simulate the shipped MessageDeduper against a log tree and classify what survives it (replay + partial-snapshot classes)

headroom_v208_residual_check.py

"""Verify headroom v2.0.8's MessageDeduper against a Windows hub log tree.

Simulates the shipped O(1) rule (drop a usage record whose message.id equals the
PREVIOUS record's message.id) and measures what inflation survives it, then
classifies the survivors:
  A REPLAY  - the identical record (same uuid) is written again later in the file
  B PARTIAL - one message.id carries DIFFERENT usage payloads (streaming snapshots)
Ground truth for "how many real messages" = distinct uuid, since uuid is unique
per written record and a replay repeats it verbatim.
0 LLM tokens, read-only.
updated: 2026-08-26
"""
import json, os, sys, time, hashlib
from collections import defaultdict

root = sys.argv[1]
files = []
for dp, _, fns in os.walk(root):
    for fn in fns:
        if fn.endswith(".jsonl"):
            files.append(os.path.join(dp, fn))
files.sort()

rec_total = 0
after_fix = 0            # records surviving his O(1) deduper
uuid_seen_global = 0
replay_records = 0       # records whose uuid already appeared in the same file
partial_msgs = 0         # message.ids with >1 distinct usage payload
partial_lost = 0         # output_tokens lost by keeping the FIRST snapshot
distinct_mid = set()
distinct_uuid_pairs = set()
files_with_replay = 0
files_with_partial = 0
tok_total = 0
tok_after_fix = 0
tok_truth = 0
t0 = time.time()

for fi, path in enumerate(files):
    prev_id = None
    uuids = set()
    mid_payloads = defaultdict(dict)   # mid -> sig -> out_tokens
    local_replay = 0
    try:
        fh = open(path, "r", encoding="utf-8", errors="replace")
    except Exception:
        continue
    with fh:
        for line in fh:
            if '"usage"' not in line:
                continue
            try:
                o = json.loads(line)
            except Exception:
                continue
            msg = o.get("message")
            if not isinstance(msg, dict):
                continue
            u = msg.get("usage")
            if not isinstance(u, dict):
                continue
            mid = msg.get("id")
            uu = o.get("uuid")
            out = u.get("output_tokens") or 0
            rec_total += 1
            tok_total += out
            # his shipped rule
            if mid is not None and mid == prev_id:
                pass                       # dropped by v2.0.8
            else:
                after_fix += 1
                tok_after_fix += out
            prev_id = mid
            if uu is not None:
                if uu in uuids:
                    replay_records += 1
                    local_replay += 1
                else:
                    uuids.add(uu)
                    tok_truth += 0         # truth accumulated below per message
            if mid is not None:
                distinct_mid.add((path, mid))
                sig = hashlib.md5(json.dumps(u, sort_keys=True).encode()).hexdigest()
                mid_payloads[mid][sig] = out
    if local_replay:
        files_with_replay += 1
    fp = 0
    for mid, sigs in mid_payloads.items():
        if len(sigs) > 1:
            partial_msgs += 1
            fp += 1
            outs = sorted(sigs.values())
            partial_lost += (outs[-1] - outs[0])
        # truth: one message contributes its LARGEST snapshot once
        tok_truth += max(sigs.values()) if sigs else 0
    if fp:
        files_with_partial += 1
    if fi and fi % 2000 == 0:
        print("  ...%d/%d  %.0fs" % (fi, len(files), time.time() - t0), file=sys.stderr)

nmid = len(distinct_mid)
print("files                      : %d" % len(files))
print("usage records (raw)        : %d" % rec_total)
print("distinct (file,message.id) : %d" % nmid)
print("raw inflation              : x%.2f" % (rec_total / nmid if nmid else 0))
print("records surviving v2.0.8   : %d" % after_fix)
print("inflation AFTER v2.0.8     : x%.2f" % (after_fix / nmid if nmid else 0))
print("")
print("A replayed records (uuid seen twice in one file) : %d  in %d files" % (replay_records, files_with_replay))
print("B message.ids with differing payloads            : %d  in %d files" % (partial_msgs, files_with_partial))
print("")
print("output_tokens raw          : %d" % tok_total)
print("output_tokens after v2.0.8 : %d" % tok_after_fix)
print("output_tokens ground truth : %d" % tok_truth)
if tok_truth:
    print("residual overcount         : %+.1f%%" % (100.0 * (tok_after_fix - tok_truth) / tok_truth))
print("B undercount if first kept : %d output_tokens" % partial_lost)
print("elapsed %.0fs" % (time.time() - t0))