""" Rule-9 tests for the Mooncake ingest — scope derived from THIS artifact's input space (jsonl schema + block semantics + the replay recipe), fixtures only, no dataset needed. Includes the required hand-derivable LRU fixture. Run: python3 test_mooncake_ingest.py B59: granularity is a keyword-required parameter (bt) with NO default, and load() accepts both the mooncake and lgw-trace/1 schemas — the new checks pin both. """ from __future__ import annotations import hashlib import json import pathlib import sys import tempfile sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent)) import mooncake_ingest as mi # AUDIT-16 convention: this script's output encoding is its own, never the host's # (P4; added B64 when the derived entry-point set first covered this file). for _s in (sys.stdout, sys.stderr): if hasattr(_s, "reconfigure"): _s.reconfigure(encoding="utf-8") passed = failed = 0 def check(name, cond, detail=""): global passed, failed if cond: passed += 1; print(f" ok {name}") else: failed += 1; print(f" FAIL {name} {detail}") def row(ts, inl, outl, hids): return {"timestamp": ts, "input_length": inl, "output_length": outl, "hash_ids": hids} # --- both reuse definitions, hand-derivable ------------------------------------- # r1: blocks [1,2], 1024 tokens (512+512). r2: blocks [1,3], 700 tokens (512+188). # Occurrences: 4 total; reused = {second 1} = 1 -> A = 1/4 = 0.25. # Tokens: total 1724; reused tokens = first block of r2 (512) -> B = 512/1724. rows = [row(0, 1024, 10, [1, 2]), row(5, 700, 10, [1, 3])] r = mi.reuse_both_definitions(rows, bt=512) check("definition A hand-derived", abs(r["A_block_occurrence_share"] - 0.25) < 1e-12) check("definition B hand-derived", abs(r["B_token_weighted_share"] - 512 / 1724) < 1e-12) check("B > A when reuse sits in full blocks", r["B_token_weighted_share"] > r["A_block_occurrence_share"]) # --- LRU replay, hand-derivable (the required fixture) --------------------------- # capacity 2 blocks; requests: [1,2] -> misses (cache 1,2); [1,3] -> 1 HIT (touch), # 3 miss evicts 2 (LRU) -> cache {1,3}; [1,2] -> 1 HIT, 2 miss evicts 3. # occurrence hits = 2/6; all blocks full 512 tokens -> token rate identical. seq = [row(0, 1024, 0, [1, 2]), row(1, 1024, 0, [1, 3]), row(2, 1024, 0, [1, 2])] out = mi.lru_replay(seq, 2, bt=512) check("LRU hit rate = 2/6 by hand", abs(out["hit_rate_occurrences"] - 2 / 6) < 1e-12, str(out)) check("token-weighted equals occurrence when all blocks full", abs(out["hit_rate_tokens"] - out["hit_rate_occurrences"]) < 1e-12) # unbounded cache ceiling equals definition A/B on the same stream unb = mi.lru_replay(seq, None, bt=512) check("unbounded LRU = reuse ceiling", abs(unb["hit_rate_occurrences"] - mi.reuse_both_definitions(seq, bt=512)["A_block_occurrence_share"]) < 1e-12) # LRU monotone: bigger capacity never hits less check("capacity monotonicity", mi.lru_replay(seq, 3, bt=512)["hit_rate_occurrences"] >= out["hit_rate_occurrences"]) # --- duplicates (AG-03) ----------------------------------------------------------- dupfix = [row(0, 1024, 100, [1, 2]), row(1, 1024, 100, [1, 2]), row(2, 700, 50, [1, 3])] d = mi.duplicates(dupfix) check("duplicate call share 1/3", abs(d["call_share"] - 1 / 3) < 1e-12) check("duplicate token share = repeat tokens / all", abs(d["token_share"] - 1124 / (1124 + 1124 + 750)) < 1e-12) # AUDIT-06 item 6: the CA-01 overlap is an INPUT-side claim (f_prefix is input-only), # so the disclosed magnitude must be an input-token share, not input+output. check("duplicate INPUT-token share reported separately", abs(d["input_token_share"] - 1024 / (1024 + 1024 + 700)) < 1e-12, str(d)) # --- schema/validation aborts (input-space rejections) ---------------------------- def write_fixture(lines): d = pathlib.Path(tempfile.mkdtemp()) for fname in mi.FILES: p = d / fname with p.open("w", encoding="utf-8", newline="\n") as _f: _f.write("\n".join(json.dumps(x) for x in lines) + "\n") return d def run_expect_abort(lines, frag, name): d = write_fixture(lines) old = dict(mi.FILES) mi.FILES.update({f: hashlib.sha256((d / f).read_bytes()).hexdigest() for f in mi.FILES}) try: mi.ingest(d) check(name, False, "did not abort") except SystemExit as e: check(name, frag in str(e), str(e)[:80]) finally: mi.FILES.clear(); mi.FILES.update(old) # B59: a line-1 row missing hash_ids satisfies NEITHER schema, so it now gets the # dispatch refusal naming both (tested below); the ROW-level schema abort is owned # by lines >= 2 of a dispatched mooncake file, so it is pinned there. run_expect_abort([row(0, 10, 1, [1]), {"timestamp": 1, "input_length": 10, "output_length": 1}], "SCHEMA MISMATCH", "missing hash_ids aborts (row level, line 2)") run_expect_abort([row(0, -5, 1, [1])], "BAD TOKEN COUNTS", "negative tokens abort") run_expect_abort([row(0, 10, 1, ["x"])], "BAD hash_ids", "non-int hash aborts") run_expect_abort([row(0, 2000, 1, [1, 2])], "BLOCK ACCOUNTING", "tokens exceeding 512*blocks abort") run_expect_abort([row(5, 10, 1, [1]), row(0, 10, 1, [2])], "NON-MONOTONIC", "timestamp regression aborts") # --- hash pinning ------------------------------------------------------------------ d = write_fixture([row(0, 512, 1, [1])]) try: mi.ingest(d) # real pinned hashes won't match the fixture check("hash mismatch aborts", False, "did not abort") except SystemExit as e: check("hash mismatch aborts", "HASH MISMATCH" in str(e)) # ===================== AUDIT-06 item 1: TTL (RED FIRST) ===================== # Hand-derivable fixture. All blocks full 512 tokens so token- and occurrence- # weighted rates coincide and the arithmetic is checkable by hand. # t=0s blocks [1,2] -> both MISS (cache 1,2 expire at 300s) # t=200s blocks [1,3] -> 1 is INSIDE window -> HIT ; 3 miss # t=400s blocks [1,4] -> refresh-on-hit: 1 was refreshed at 200s to 500s -> HIT # hard expiry: 1 still expires at 300s -> MISS # occurrences = 6. refresh: 2 hits = 2/6. hard: 1 hit = 1/6. TTLFIX = [row(0, 1024, 0, [1, 2]), row(200_000, 1024, 0, [1, 3]), row(400_000, 1024, 0, [1, 4])] check("TTL replay accepts ttl_seconds/refresh_on_hit", "ttl_seconds" in mi.lru_replay.__code__.co_varnames and "refresh_on_hit" in mi.lru_replay.__code__.co_varnames) r_ref = mi.lru_replay(TTLFIX, None, ttl_seconds=300, refresh_on_hit=True, bt=512) r_hard = mi.lru_replay(TTLFIX, None, ttl_seconds=300, refresh_on_hit=False, bt=512) check("5-min TTL refresh-on-hit = 2/6 by hand", abs(r_ref["hit_rate_occurrences"] - 2/6) < 1e-12, str(r_ref)) check("5-min TTL hard expiry = 1/6 by hand", abs(r_hard["hit_rate_occurrences"] - 1/6) < 1e-12, str(r_hard)) # just inside vs just outside the window, no intervening touch inside = [row(0, 512, 0, [7]), row(299_000, 512, 0, [7])] outside = [row(0, 512, 0, [7]), row(301_000, 512, 0, [7])] check("reuse just INSIDE the window hits", mi.lru_replay(inside, None, ttl_seconds=300, bt=512)["hit_rate_occurrences"] == 0.5) check("reuse just OUTSIDE the window misses", mi.lru_replay(outside, None, ttl_seconds=300, bt=512)["hit_rate_occurrences"] == 0.0) # TTL monotonicity: a longer window can never produce fewer hits h300 = mi.lru_replay(TTLFIX, None, ttl_seconds=300, bt=512)["hit_rate_tokens"] h3600 = mi.lru_replay(TTLFIX, None, ttl_seconds=3600, bt=512)["hit_rate_tokens"] check("longer TTL never hits less", h3600 >= h300 - 1e-12) # RULE 10 — properties the replay ALREADY carried must survive the change check("ttl_seconds=None reproduces the pre-TTL behaviour", mi.lru_replay(seq, 2, ttl_seconds=None, bt=512)["hit_rate_occurrences"] == 2/6) _unb = mi.lru_replay(seq, None, ttl_seconds=None, bt=512)["hit_rate_occurrences"] check("RULE10 unbounded LRU still == definition A", abs(_unb - mi.reuse_both_definitions(seq, bt=512)["A_block_occurrence_share"]) < 1e-12) check("RULE10 capacity monotonicity holds with TTL on", mi.lru_replay(TTLFIX, 3, ttl_seconds=300, bt=512)["hit_rate_occurrences"] >= mi.lru_replay(TTLFIX, 1, ttl_seconds=300, bt=512)["hit_rate_occurrences"] - 1e-12) # the shipped engine input must be priced on the tier it was replayed at check("TTL tiers declare ttl_seconds AND the write multiple they are priced at", all({"ttl_seconds", "w"} <= set(v) for v in mi.TTL_TIERS.values()), str(getattr(mi, "TTL_TIERS", None))) check("default operating point is the 5-minute tier", mi.DEFAULT_TIER == "5min") check("5-min tier priced at the 5-min write multiple", mi.TTL_TIERS["5min"]["w"] == 1.25) check("1-hour tier priced at the 1-hour write multiple", mi.TTL_TIERS["1hour"]["w"] == 2.00) # ---------------------------------------------------------------- AUDIT-08 (red first) # A retention WINDOW is a vendor product decision; the PRICE attached to that window # differs per vendor. TTL_TIERS binds Anthropic's purchasable pairings. The retention # probe generalises AUDIT-06 C1: measure the achievable hit rate at arbitrary windows # so any vendor's documented window can be paired with THAT vendor's documented # multiples, instead of borrowing a hit rate measured at someone else's window. check("retention probe declares the windows the published cards actually document", {300, 600, 1800, 3600} <= set(getattr(mi, "RETENTION_PROBE_SECONDS", ())), str(getattr(mi, "RETENTION_PROBE_SECONDS", None))) check("the probe is a bare window list and binds NO write multiple " "(a window is not a price)", isinstance(getattr(mi, "RETENTION_PROBE_SECONDS", None), (tuple, list)) and all(isinstance(s, int) for s in getattr(mi, "RETENTION_PROBE_SECONDS", ())), str(getattr(mi, "RETENTION_PROBE_SECONDS", None))) _probe = {s: mi.lru_replay(TTLFIX, None, ttl_seconds=s, bt=512)["hit_rate_tokens"] for s in sorted(getattr(mi, "RETENTION_PROBE_SECONDS", (300, 3600)))} check("probe is monotone non-decreasing in window length", all(b >= a - 1e-12 for a, b in zip(list(_probe.values()), list(_probe.values())[1:])), str(_probe)) check("the 300 s probe equals the 5-minute tier's own measured hit rate " "(one replay, two callers)", abs(_probe.get(300, -1) - mi.lru_replay(TTLFIX, None, ttl_seconds=mi.TTL_TIERS["5min"]["ttl_seconds"], bt=512)["hit_rate_tokens"]) < 1e-12, str(_probe)) # ===================== B59: dual-schema load + granularity as a parameter ========= def expect_abort(fn, frag, name): try: fn() check(name, False, "did not abort") except SystemExit as e: check(name, frag in str(e), str(e)[:100]) def write_lgw(meta, lrows): p = pathlib.Path(tempfile.mkdtemp()) / "trace.lgw.jsonl" with p.open("w", encoding="utf-8", newline="\n") as f: f.write(json.dumps(meta) + "\n") for x in lrows: f.write(json.dumps(x) + "\n") return p LGW_META = {"format": "lgw-trace/1", "hasher_version": "1.0.0", "hash_granularity": 2048, "granularity_unit": "chars", "lengths_unit": "chars", "timestamp_unit": "ms", "chunk_mode": "cumulative-prefix-sha256-128bit", "salt_fingerprint": "deadbeef", "rows": 2, "content_fields_present": False} def lrow(ts, inl, outl, hashes): return {"timestamp": ts, "input_length": inl, "output_length": outl, "prefix_hashes": hashes} # granularity has NO default: a mooncake trace (no metadata) with no caller value # is refused BY NAME — a replay at the wrong block size is a plausible wrong number _moon = write_fixture([row(0, 512, 1, [1])]) / "conversation_trace.jsonl" expect_abort(lambda: mi.load(_moon), "MISSING GRANULARITY", "mooncake trace without block_tokens is refused by name") _r, _m = mi.load(_moon, block_tokens=512) check("mooncake load returns rows verbatim and meta with the supplied granularity", _r == [row(0, 512, 1, [1])] and _m["schema"] == "mooncake" and _m["block_tokens"] == 512 and _m["timestamp_unit"] == "ms", str(_m)) # lgw-trace/1: dispatch on the DECLARED discriminator; granularity from metadata _p = write_lgw(LGW_META, [lrow(0, 4096, 10, ["a" * 32, "b" * 32]), lrow(5, 2048, 10, ["a" * 32])]) _r, _m = mi.load(_p) check("lgw-trace/1 dispatches on format and takes granularity from metadata", _m == {"schema": "lgw-trace/1", "block_tokens": 2048, "timestamp_unit": "ms", "lengths_unit": "chars"}, str(_m)) check("lgw rows normalise prefix_hashes to the internal hash_ids shape", _r[0]["hash_ids"] == ["a" * 32, "b" * 32] and "prefix_hashes" not in _r[0]) _r2, _ = mi.load(_p, block_tokens=2048) check("agreeing block_tokens with lgw metadata is accepted", _r2 == _r) expect_abort(lambda: mi.load(_p, block_tokens=512), "GRANULARITY CONFLICT", "conflicting caller granularity vs lgw metadata is refused by name") _bad = dict(LGW_META); del _bad["hash_granularity"] expect_abort(lambda: mi.load(write_lgw(_bad, [])), "MISSING GRANULARITY", "lgw metadata without hash_granularity is refused by name") _bad = dict(LGW_META); _bad["timestamp_unit"] = "s" expect_abort(lambda: mi.load(write_lgw(_bad, [])), "UNSUPPORTED timestamp_unit", "non-ms timestamp_unit is refused, not converted") # a trace satisfying NEITHER schema is refused naming BOTH — never coerced, # never a try/except fallback that swallows malformed input as the other type def _names_both(e): return "lgw-trace/1" in str(e) and "mooncake" in str(e) try: mi.load(write_lgw({"foo": 1}, [])) check("neither-schema input refused naming both schemas", False, "did not abort") except SystemExit as e: check("neither-schema input refused naming both schemas", _names_both(e), str(e)[:100]) _notjson = pathlib.Path(tempfile.mkdtemp()) / "x.jsonl" with _notjson.open("w", encoding="utf-8", newline="\n") as _fh: _fh.write("timestamp,input_length\n0,5\n") try: mi.load(_notjson) check("non-JSON line 1 refused naming both schemas", False, "did not abort") except SystemExit as e: check("non-JSON line 1 refused naming both schemas", _names_both(e), str(e)[:100]) # the accounting refusal prints the trace's OWN granularity, not a hardcoded 512 _bad = dict(LGW_META); _bad["hash_granularity"] = 1000 expect_abort(lambda: mi.load(write_lgw(_bad, [lrow(0, 2001, 0, ["a" * 32, "b" * 32])])), "1000*2", "accounting refusal prints the trace's own granularity") # granularity is a REAL parameter: pinned non-512 fixtures (B58 ran 1024 in memory; # the suite holds it now) seq1024 = [row(0, 2048, 0, [1, 2]), row(1, 2048, 0, [1, 3]), row(2, 2048, 0, [1, 2])] check("LRU 2/6 by hand at bt=1024 (pinned non-512 granularity)", abs(mi.lru_replay(seq1024, 2, bt=1024)["hit_rate_occurrences"] - 2 / 6) < 1e-12) _r1024 = mi.reuse_both_definitions([row(0, 2048, 0, [1, 2]), row(5, 1400, 0, [1, 3])], bt=1024) check("definition B by hand at bt=1024 with a partial final block", abs(_r1024["B_token_weighted_share"] - 1024 / 3448) < 1e-12, str(_r1024)) check("capacity_tokens echoes the parameter, not 512", mi.lru_replay(seq1024, 2, bt=1024)["capacity_tokens"] == 2048) # string block ids (lgw hashes) replay identically to int ids seq_str = [row(0, 1024, 0, ["aa", "bb"]), row(1, 1024, 0, ["aa", "cc"]), row(2, 1024, 0, ["aa", "bb"])] check("string block ids replay identically (the replay is id-type-agnostic)", abs(mi.lru_replay(seq_str, 2, bt=512)["hit_rate_occurrences"] - 2 / 6) < 1e-12) # and no default exists at the function boundary either try: mi.lru_replay(seq, 2) check("bt is keyword-required with no default", False, "call without bt succeeded") except TypeError: check("bt is keyword-required with no default", True) # end-to-end: an lgw trace loads and replays through to an h_target _pipe = write_lgw(dict(LGW_META), [lrow(0, 4096, 10, ["a" * 32, "b" * 32]), lrow(60_000, 4096, 10, ["a" * 32, "b" * 32])]) _rows, _meta = mi.load(_pipe) _h = mi.lru_replay(_rows, None, ttl_seconds=300, bt=_meta["block_tokens"])["hit_rate_tokens"] check("lgw trace end-to-end: load -> replay -> h_target (both blocks re-served)", abs(_h - 0.5) < 1e-12, str(_h)) print(f"\nmooncake ingest tests: {passed} passed, {failed} failed") sys.exit(1 if failed else 0)