"""eval.py — does the model still get the answer after compression? Compression ratios prove the log got smaller, not that the signal survived. This is the test that matters: eight synthetic-but-realistic failure logs, each with ONE planted root cause buried in machine noise (plus red herrings), asked to a model twice — raw vs `compress(level=2)` — and graded by a deterministic keyword check on the answer. If compression eats the diagnosis, this catches it. python3 eval.py --dry # build scenarios, show token counts, no model python3 eval.py # ask the model (needs a logged-in `claude` CLI) LM_EVAL_CMD="llm -m gpt-5-mini" python3 eval.py # any prompt-on-stdin CLI Scenario generation is seeded — same logs every run. Model answers vary run to run; the binary keyword grading absorbs phrasing differences. """ import argparse import os import random import re import shlex import subprocess import sys from lessismore import compress, count_tokens QUESTION = ("You are debugging a production incident. Above is the captured " "log output. In 1-2 sentences: what is the ROOT CAUSE of the failure?") MODEL_CMD = os.environ.get("LM_EVAL_CMD", "claude -p --model claude-haiku-4-5-20251001") # ---------------------------------------------------------------- scenarios # Each returns (log_text, expect_regex). The regex is what a correct root-cause # answer must mention; red herrings are planted so a wrong answer fails it. def s_buried_oom(): """Kernel OOM-kills postgres mid-log; the tail is 400 lines of red-herring connection errors that a lost-in-the-middle model blames instead.""" r, out = random.Random(11), [] for i in range(700): out.append(f"2026-07-07T09:{i // 60:02d}:{i % 60:02d}Z INFO [api] " f"GET /v1/orders/{r.randint(1000, 9999)} 200 in {r.randint(8, 90)}ms") out.append("2026-07-07T09:11:40Z kernel: Out of memory: Killed process 2211 (postgres) " "total-vm:8123456kB, anon-rss:7901234kB") out.append("2026-07-07T09:11:40Z postgres[2211]: FATAL: terminating connection due to " "unexpected postmaster exit") for i in range(400): out.append(f"2026-07-07T09:{12 + i // 60:02d}:{i % 60:02d}Z ERROR [api] " f"psycopg2.OperationalError: connection to server at 10.0.0.5 refused") return "\n".join(out), r"out of memory|oom|killed process|memory" def s_status_codes(): """Rate limiting (429) from the payments API triggers cascading 500s. The codes differ only in digits — exactly what a careless similar-line collapse would eat.""" r, out = random.Random(12), [] for i in range(1400): if i < 600: c, ms = 200, r.randint(40, 200) elif i < 700: c, ms = r.choice([429, 429, 429, 200]), r.randint(5, 30) else: c, ms = r.choice([500, 502, 500]), r.randint(2000, 3100) out.append(f"2026-07-07T14:{i // 60 % 60:02d}:{i % 60:02d}Z gateway: upstream " f"payments-api returned status {c} for POST /api/checkout in {ms}ms") return "\n".join(out), r"429|rate.?limit|too many requests" def s_interleaved_disk(): """Three services interleave (zero consecutive repeats — dedupe-proof); the db's 'No space left on device' is scattered 1-in-40.""" r, out = random.Random(13), [] for i in range(1500): ts = f"2026-07-07T16:{i // 60 % 60:02d}:{i % 60:02d}Z" which = i % 3 if which == 0: out.append(f"{ts} INFO [api] request {r.randint(10000, 99999)} completed") elif which == 1: out.append(f"{ts} WARN [worker] job retry {r.randint(1, 5)} scheduled, backing off") elif i % 40 == 2: out.append(f"{ts} ERROR [db] could not extend file base/16384/2619: " f"No space left on device") else: out.append(f"{ts} INFO [db] checkpoint complete: wrote {r.randint(100, 999)} buffers") return "\n".join(out), r"space|disk|storage|full" def s_ansi_ci(): """1400 green PASSED lines in full ANSI dress; one red FAILED assertion names the offending function.""" r, out = random.Random(14), [] mods = ["auth", "cart", "checkout", "billing", "search", "profile"] for i in range(1400): m = r.choice(mods) out.append(f"\x1b[32mPASSED\x1b[0m tests/test_{m}.py::test_{m}_{r.randint(1, 99):02d} " f"\x1b[90m({r.randint(1, 40)}ms)\x1b[0m") if i == 981: out.append("\x1b[31mFAILED\x1b[0m tests/test_billing.py::test_invoice_total") out.append("\x1b[31mE AssertionError: round_half(2.675) == 2.68, got 2.67 — " "float truncation in round_half()\x1b[0m") out.append("\x1b[31m1 failed\x1b[0m, \x1b[32m1401 passed\x1b[0m in 42.31s") return "\n".join(out), r"round_half|rounding|truncat|2\.6[78]" def s_pip_conflict(): """Progress-bar walls (\\r frames) drown a one-line dependency conflict.""" r, out = random.Random(15), [] for pkg in ("numpy", "pandas", "scipy", "matplotlib", "scikit_learn", "torch"): out.append(f"Collecting {pkg}") out.append(f" Downloading {pkg}-2.1.0-cp312-cp312-macosx_11_0_arm64.whl " f"({r.randint(1, 80)}.{r.randint(0, 9)} MB)") out.append("".join(f"\r |{'█' * (i // 3)}{' ' * (34 - i // 3)}| {i}% " f"{r.randint(100, 999)}.{r.randint(0, 9)} kB/s eta 0:00:{99 - i:02d}" for i in range(1, 101))) out.append("ERROR: Cannot install app 1.0 because requests 2.32.0 requires urllib3<3, " "but you have urllib3 3.0.1 which is incompatible.") return "\n".join(out), r"urllib3" def s_cert_expired(): """An nginx access-log wall; the error-log lines that matter say the upstream's TLS certificate expired.""" r, out = random.Random(16), [] for i in range(1300): ts = f"[07/Jul/2026:18:{i // 60 % 60:02d}:{i % 60:02d} +0000]" if i % 60 == 30: out.append(f"2026/07/07 18:{i // 60 % 60:02d}:{i % 60:02d} [error] 812#0: SSL_do_handshake() " f"failed (SSL: certificate verify failed: certificate has expired) " f"while connecting to upstream auth-service:8443") out.append(f'10.0.3.{r.randint(2, 250)} - - {ts} "GET /login HTTP/1.1" 502 552') else: out.append(f'10.0.3.{r.randint(2, 250)} - - {ts} "GET /{r.choice(["", "static/app.js", "api/health"])} ' f'HTTP/1.1" 200 {r.randint(200, 9000)}') return "\n".join(out), r"expir|certificate" def s_env_missing(): """A crash-looping pod re-prints the same traceback 60 times; the KeyError names the missing variable. Probe-failure noise is the red herring.""" out = [] for i in range(60): out.append(f"2026-07-07T20:{i:02d}:01Z k8s: Readiness probe failed: connect: " f"connection refused") out.append(f"2026-07-07T20:{i:02d}:03Z k8s: Back-off restarting failed container " f"app in pod shop-6d8f9/app") out += ["Traceback (most recent call last):", ' File "/app/.venv/lib/python3.12/site-packages/myapp/config.py", line 44, ' "in load", " dsn = os.environ['DATABASE_URL']", ' File "", line 685, in __getitem__', "KeyError: 'DATABASE_URL'"] return "\n".join(out), r"database_url|environment variable|env var" def s_deadlock(): """A 40-thread Java dump, hundreds of near-identical frames; one section declares the deadlock.""" r, out = random.Random(18), [] for t in range(40): out.append(f'"pool-1-thread-{t}" #{t + 20} prio=5 tid=0x{r.getrandbits(48):012x} ' f"waiting on condition") for _ in range(12): cls = r.choice(["QueueWorker", "BatchLoader", "HttpDispatch", "CacheSync"]) out.append(f"\tat com.shop.core.{cls}.run({cls}.java:{r.randint(40, 400)})") out += ["Found one Java-level deadlock:", '"OrderWriter" waiting to lock monitor 0x00007f2c (an InventoryLock),', ' which is held by "StockUpdater", which is waiting to lock 0x00007f2d,', ' which is held by "OrderWriter"'] for t in range(20): out.append(f'"GC-thread-{t}" os_prio=31 tid=0x{r.getrandbits(48):012x} runnable') return "\n".join(out), r"deadlock" SCENARIOS = [("buried-oom", s_buried_oom), ("status-codes", s_status_codes), ("interleaved-disk", s_interleaved_disk), ("ansi-ci", s_ansi_ci), ("pip-conflict", s_pip_conflict), ("cert-expired", s_cert_expired), ("env-missing", s_env_missing), ("deadlock", s_deadlock)] # ---------------------------------------------------------------- harness def ask(prompt: str, cmd: str = MODEL_CMD) -> str: r = subprocess.run(shlex.split(cmd), input=prompt, capture_output=True, text=True, timeout=600) if r.returncode != 0: sys.exit(f"model command failed: {cmd}\n{r.stderr}") return r.stdout def main(): p = argparse.ArgumentParser(description=__doc__.splitlines()[0]) p.add_argument("--dry", action="store_true", help="build scenarios and report token counts; no model calls") p.add_argument("--dump", metavar="DIR", help="also write .raw.txt / .small.txt prompts to DIR") a = p.parse_args() from lessismore import _encoder if _encoder() is None: print("note: token counts are chars/4 estimates — pip install tiktoken " "for o200k counts (grading is unaffected)", file=sys.stderr) rows, ok_raw, ok_small = [], 0, 0 print(f"{'scenario':18} {'raw':>7} {'small':>7} {'ratio':>6} raw small") for name, gen in SCENARIOS: log, expect = gen() small = compress(log, 2) tr, ts = count_tokens(log), count_tokens(small) if a.dump: import pathlib d = pathlib.Path(a.dump) d.mkdir(exist_ok=True) (d / f"{name}.raw.txt").write_text(log + "\n\n" + QUESTION) (d / f"{name}.small.txt").write_text(small + "\n\n" + QUESTION) if a.dry: print(f"{name:18} {tr:>7,} {ts:>7,} {tr / ts:>5.1f}x") continue graded = [] for text in (log, small): ans = ask(text + "\n\n" + QUESTION) graded.append(bool(re.search(expect, ans, re.I))) ok_raw += graded[0] ok_small += graded[1] mark = {True: "PASS", False: "FAIL"} print(f"{name:18} {tr:>7,} {ts:>7,} {tr / ts:>5.1f}x {mark[graded[0]]} {mark[graded[1]]}") rows.append((name, tr, ts, graded)) if not a.dry: n = len(SCENARIOS) print(f"\nroot cause found: raw {ok_raw}/{n}, compressed {ok_small}/{n} " f"(model: {MODEL_CMD})") if __name__ == "__main__": main()