"""C2a: does every fact survive compression? Answers computed by code from the ORIGINAL payload, then checked against what Headroom 0.41.0 produced with the default compress() call (same messages as c1b). CCR markers (<>) are resolved through Headroom's own local compression store, as headroom_retrieve would do. Also a full row-level check.""" import json, csv, io, re, sys, os from headroom import compress from headroom.cache.compression_store import get_compression_store P = "payloads/" FILES = {"github_issues": "gh_issues.json", "pypi_info": "pypi_info.json", "synthetic_logs": "synthetic_logs.json", "wiki_search": "wiki_search.json"} store = get_compression_store() def run(raw): msgs = [{"role": "system", "content": "You are a helpful assistant."}, {"role": "user", "content": "Here is the tool output. Answer questions about it."}, {"role": "assistant", "content": None, "tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "fetch", "arguments": "{}"}}]}, {"role": "tool", "tool_call_id": "call_1", "content": raw}] return compress(msgs, model="gpt-4o").messages[-1]["content"] CCR = re.compile(r"<]*>>") def resolve(s): stats = {"markers": 0, "resolved": 0} def f(m): stats["markers"] += 1 e = store.retrieve(m.group(1)) if e is None: return m.group(0) stats["resolved"] += 1 return e.original_content return CCR.sub(f, s), stats def decode_table(s): m = re.match(r"\[(\d+)\]\{([^}]*)\}\n", s) if not m: return None spec = [f.split(":") for f in m.group(2).split(",")] rows = [r for r in csv.reader(io.StringIO(s[m.end():])) if r] out = [] for r in rows: d = {} for (k, t), v in zip(spec, r): d[k] = int(v) if t == "int" else v out.append(d) return int(m.group(1)), [k for k, _ in spec], out data = {n: json.load(open(P + f)) for n, f in FILES.items()} gh, py, lg, wk = data["github_issues"], data["pypi_info"], data["synthetic_logs"], data["wiki_search"] Q = { "github_issues": [("title of issue #%d" % gh[0]["number"], gh[0]["title"]), ("state of issue #%d" % gh[37]["number"], gh[37]["state"]), ("who opened issue #%d" % gh[61]["number"], gh[61]["user"]["login"]), ("comment count of issue #%d" % gh[88]["number"], str(gh[88]["comments"])), ("how many of the 100 items are pull requests (aggregate)", str(sum(1 for x in gh if "pull_request" in x)))], "pypi_info": [("version of %s" % py[0]["name"], py[0]["version"]), ("requires_python of %s" % py[20]["name"], str(py[20]["requires_python"])), ("license of %s" % py[45]["name"], str(py[45]["license"])), ("summary of %s" % py[70]["name"], str(py[70]["summary"])), ("how many packages have requires_python set (aggregate)", str(sum(1 for x in py if x["requires_python"])))], "synthetic_logs": [("message of the FATAL entry", next(x["msg"] for x in lg if x["level"] == "FATAL")), ("request_id of the FATAL entry", next(x["request_id"] for x in lg if x["level"] == "FATAL")), ("latency of %s" % lg[500]["request_id"], str(lg[500]["latency_ms"])), ("service at %s" % lg[1999]["ts"], lg[1999]["service"]), ("how many ERROR entries (aggregate)", str(sum(1 for x in lg if x["level"] == "ERROR")))], "wiki_search": [("title of result 1", wk[0]["title"]), ("pageid of '%s'" % wk[49]["title"], str(wk[49]["pageid"])), ("wordcount of '%s'" % wk[149]["title"], str(wk[149]["wordcount"])), ("snippet of '%s'" % wk[3]["title"], wk[3]["snippet"]), ("how many results have wordcount > 5000 (aggregate)", str(sum(1 for x in wk if x["wordcount"] > 5000)))], } res = {}; total = kept_direct = kept_ccr = 0 for name, qs in Q.items(): raw = open(P + FILES[name]).read() comp = run(raw) same_as_c1b = comp == open(P + name + ".compressed.txt").read() text = json.loads(comp) if comp.startswith('"') else comp unchanged = text == raw full, st = resolve(text) marks = {} def ph(m): k = "@@CCR%d@@" % len(marks); marks[k] = m.group(0); return k tab = decode_table(CCR.sub(ph, text)) tab_raw = decode_table(CCR.sub(ph, text)) if tab: for r in tab[2]: for k, v in r.items(): if isinstance(v, str) and v in marks: r[k] = resolve(marks[v])[0] rowcheck = None if tab: n, fields, rows = tab orig = [{k: x.get(k) for k in fields} for x in data[name]] ident = sum(1 for a, b in zip(orig, rows) if a == b) rowcheck = {"declared_rows": n, "decoded_rows": len(rows), "original_rows": len(orig), "rows_identical_after_ccr": ident, "fields_dropped": sorted(set(data[name][0]) - set(fields))} all_rows_ok = unchanged or (rowcheck and rowcheck["rows_identical_after_ccr"] == rowcheck["original_rows"] and not rowcheck["fields_dropped"]) out = [] for q, a in qs: agg = q.endswith("(aggregate)") # aggregate: answerable straight from the compressed table if the column it needs has no CCR marker agg_field = {"synthetic_logs": "level", "wiki_search": "wordcount"}.get(name) col_clean = bool(tab) and agg_field is not None and all(not (isinstance(r0.get(agg_field), str) and "@@CCR" in r0.get(agg_field)) for r0 in tab_raw[2]) and all_rows_ok direct = unchanged or (agg and col_clean) or (not agg and (a in text or a.replace('"', '""') in text)) via = direct or (bool(all_rows_ok) if agg else (a in full or a.replace('"', '""') in CCR.sub(lambda m: resolve(m.group(0))[0].replace('"', '""'), text))) total += 1; kept_direct += direct; kept_ccr += via out.append({"q": q, "answer": a[:200], "in_compressed_text": direct, "after_ccr_retrieve": via}) res[name] = {"identical_to_c1b_output": same_as_c1b, "unchanged_passthrough": unchanged, "ccr": st, "row_check": rowcheck, "questions": out} res["summary"] = {"questions": total, "retained_in_compressed_text": kept_direct, "retained_after_ccr": kept_ccr, "retention_pct_direct": round(100 * kept_direct / total, 1), "retention_pct_with_ccr": round(100 * kept_ccr / total, 1)} json.dump(res, open(sys.argv[1], "w"), indent=1, ensure_ascii=False) print(json.dumps(res["summary"])) for n in Q: print(n, res[n]["identical_to_c1b_output"], res[n]["unchanged_passthrough"], res[n]["ccr"], res[n]["row_check"], [(q["in_compressed_text"], q["after_ccr_retrieve"]) for q in res[n]["questions"]])