rune/scripts/check_drift_compression_parity.py

#!/usr/bin/env python3
"""check_drift_compression_parity.py — live parity for the drift_compression
gap-close port (dichotomy wave, #2479 tranche).

Three live layers over the FROZEN truth
(src/rust/memory_scoring/src/frozen_dcomp_truth.json, 65 cases, frozen via
scripts/gen_drift_compression_truth.py against the pure _py_* oracle):

  1. facade PINNED  — RUNE_MEMORY_CORE=python: the public fns run the
     load-bearing Python fallback against the frozen expectations.
  2. facade LIVE    — delegation on (the built memory_scoring .so through
     ms_loader): the same public fns against the same expectations.
  3. KERNEL         — ms_loader.* called directly (the FFI envelope path).

Plus two end-to-end seams the frozen file cannot carry (a sqlite
connection is not JSON-freezable):
  4. db-level get_compression_stats: an in-memory entries table seeded with
     active/inactive rows, SQL nulls and mixed rock statuses — pinned vs
     live vs the _py math, all three must agree (pins the status='active'
     filter and the None coercion of SQL nulls).
  5. FUZZ (seed 2479): deterministic random entries/rows/texts (unicode
     included) — pinned facade vs live facade vs kernel, plus the stats
     row-reduction, N rounds each.

Comparison law: structural == (dict/list equality). The one documented
representation quirk: integral FLOATS crossing the FFI render int-shaped
(savings_pct 50.0 from the oracle, 50 from the envelope) — == treats them
equal, which is exactly how the oracle's own consumers read the dict;
byte-shape parity for the stats dict is pinned by the Rust battery's
frozen replay instead.

Exit 0 iff every layer is 100%.

Run: PYTHONPATH=src python3 scripts/check_drift_compression_parity.py
"""
from __future__ import annotations

import json
import os
import random
import sqlite3
import sys
from pathlib import Path

REPO = Path(__file__).resolve().parents[1]
TRUTH_PATH = REPO / "src" / "rust" / "memory_scoring" / "src" / "frozen_dcomp_truth.json"
FUZZ_SEED = 2479

sys.path.insert(0, str(REPO / "src"))

# The delegation pin is read PER CALL (drift_core._ms_core reads the env
# every time), so one process can exercise both modes by toggling.
os.environ["RUNE_MEMORY_CORE"] = "python"

from rune import drift_compression as dc  # noqa: E402
from rune import ms_loader  # noqa: E402

STATUSES = ["washable", "established", "rock", "muddy", None]
FILLER = "alpha bravo charlie delta echo foxtrot golf hotel "


def long_text(n: int) -> str:
    return (FILLER * (n // len(FILLER) + 1))[:n]


def pinned() -> bool:
    return os.environ.get("RUNE_MEMORY_CORE") == "python"


def facade_call(fn_name: str, args: list, kwargs: dict):
    """Public-facade dispatch for one frozen case (stats → the rows kernel)."""
    if fn_name == "_py_get_compression_stats":
        fn = dc._py_get_compression_stats if pinned() else ms_loader.get_compression_stats
        return fn(*args, **kwargs)
    public = fn_name.removeprefix("_py_")
    fn = getattr(dc, public)
    return fn(*args, **kwargs)


def kernel_call(fn_name: str, args: list, kwargs: dict):
    if fn_name == "_py_get_compression_ratio":
        return ms_loader.get_compression_ratio(*args, **kwargs)
    if fn_name == "_py_compress_text":
        return ms_loader.compress_text(*args, **kwargs)
    if fn_name == "_py_compress_entry":
        return ms_loader.compress_entry(*args, **kwargs)
    if fn_name == "_py_should_warn":
        return ms_loader.should_warn(*args, **kwargs)
    if fn_name == "_py_warn_before_compress":
        return ms_loader.warn_before_compress(*args, **kwargs)
    if fn_name == "_py_get_compression_stats":
        return ms_loader.get_compression_stats(*args, **kwargs)
    raise KeyError(fn_name)


def run_layer(label: str, cases: list, call) -> tuple[int, int]:
    ok = 0
    fails = []
    for case in cases:
        try:
            got = call(case["fn"], case["args"], case.get("kwargs", {}))
        except Exception as e:  # noqa: BLE001 — a crash IS the divergence
            fails.append((case["fn"], case["args"], f"raised {type(e).__name__}: {e}"))
            continue
        if got == case["expected"]:
            ok += 1
        else:
            fails.append((case["fn"], case["args"], f"got {got!r} want {case['expected']!r}"))
    print(f"{label}: {ok}/{len(cases)} match")
    for fn, args, detail in fails[:5]:
        print(f"  DIVERGE {fn} args={json.dumps(args, default=str)[:120]}: {detail[:200]}")
    return ok, len(cases)


def make_db() -> sqlite3.Connection:
    db = sqlite3.connect(":memory:")
    db.execute(
        "CREATE TABLE entries (id INTEGER PRIMARY KEY, text TEXT, wash_count INTEGER,"
        " rock_status TEXT, status TEXT)"
    )
    db.executemany(
        "INSERT INTO entries VALUES (?,?,?,?,?)",
        [
            (1, long_text(120), 0, "washable", "active"),
            (2, long_text(80), 5, "rock", "active"),
            (3, long_text(60), 3, "established", "active"),
            (4, long_text(90), 15, "washable", "active"),
            (5, None, None, None, "active"),       # SQL nulls (LENGTH(NULL) → None)
            (6, long_text(500), 25, "washable", "active"),
            (7, long_text(77), 7, "washable", "retired"),   # filtered: status != active
            (8, long_text(33), 1, "muddy", "active"),       # unknown status
        ],
    )
    return db


def db_level_check() -> tuple[int, int]:
    db = make_db()
    os.environ["RUNE_MEMORY_CORE"] = "python"
    pinned_stats = dc.get_compression_stats(db)
    del os.environ["RUNE_MEMORY_CORE"]
    live_stats = dc.get_compression_stats(db)
    db.close()

    rows = [
        ["washable", 0, 120], ["rock", 5, 80], ["established", 3, 60],
        ["washable", 15, 90], [None, None, None], ["washable", 25, 500],
        ["muddy", 1, 33],
    ]
    math_stats = dc._py_get_compression_stats(rows)
    checks = [pinned_stats == live_stats, live_stats == math_stats]
    print(f"db-level get_compression_stats: pinned==live {checks[0]}, live==math {checks[1]}")
    if not all(checks):
        print(f"  pinned: {pinned_stats}")
        print(f"  live:   {live_stats}")
        print(f"  math:   {math_stats}")
    return sum(checks), len(checks)


def fuzz_check(rounds: int = 400) -> tuple[int, int]:
    rng = random.Random(FUZZ_SEED)
    ok = total = 0

    def agree(label: str, values: list) -> None:
        nonlocal ok, total
        total += 1
        if all(v == values[0] for v in values[1:]):
            ok += 1
        else:
            print(f"  FUZZ DIVERGE [{label}]: {json.dumps(values, default=str)[:400]}")

    os.environ["RUNE_MEMORY_CORE"] = "python"
    for i in range(rounds):
        text = long_text(rng.choice([0, 5, 19, 20, 21, 100, 199, 200, 201, 600, 1337]))
        if rng.random() < 0.2:
            text = text + "汉字" * rng.randint(1, 4) + " " * rng.randint(0, 3)
        entry = {"text": text}
        if rng.random() < 0.8:
            entry["wash_count"] = rng.choice([0, 1, 5, 9, 10, 19, 20, 25, 100, None])
        if rng.random() < 0.8:
            entry["rock_status"] = rng.choice(STATUSES)
        if rng.random() < 0.5:
            entry["id"] = rng.choice([i, f"slug-{i}", None])
        ratio = rng.choice([0.0, 0.1, 0.2, 0.25, 0.5, 0.9, 0.999, 1.0, rng.random()])

        # pinned facade (pure Python)
        ce_p = dc.compress_entry(entry)
        sw_p = dc.should_warn(entry)
        wb_p = dc.warn_before_compress(entry)
        ct_p = dc.compress_text(text, ratio)
        # live facade (delegation)
        del os.environ["RUNE_MEMORY_CORE"]
        ce_l = dc.compress_entry(entry)
        sw_l = dc.should_warn(entry)
        wb_l = dc.warn_before_compress(entry)
        ct_l = dc.compress_text(text, ratio)
        # kernel direct
        ce_k = ms_loader.compress_entry(entry)
        sw_k = ms_loader.should_warn(entry)
        wb_k = ms_loader.warn_before_compress(entry)
        ct_k = ms_loader.compress_text(text, ratio)
        os.environ["RUNE_MEMORY_CORE"] = "python"

        agree("compress_entry", [ce_p, ce_l, ce_k])
        agree("should_warn", [sw_p, sw_l, sw_k])
        agree("warn_before_compress", [wb_p, wb_l, wb_k])
        agree("compress_text", [ct_p, ct_l, ct_k])

        if rng.random() < 0.35:
            rows = [
                [
                    rng.choice(STATUSES),
                    rng.choice([0, 3, 10, 15, 20, 40, None]),
                    rng.choice([0, 1, 17, 33, 100, 250, 1000, None]),
                ]
                for _ in range(rng.randint(0, 6))
            ]
            st_p = dc._py_get_compression_stats(rows)
            st_k = ms_loader.get_compression_stats(rows)
            agree("get_compression_stats", [st_p, st_k])

    print(f"fuzz (seed {FUZZ_SEED}): {ok}/{total} agree across pinned/live/kernel")
    return ok, total


def main() -> int:
    if not ms_loader.available():
        print("memory_scoring .so not built — run cargo build --release first")
        return 1
    cases = json.loads(TRUTH_PATH.read_text())

    total_ok = total_n = 0

    os.environ["RUNE_MEMORY_CORE"] = "python"
    ok, n = run_layer("facade pinned (RUNE_MEMORY_CORE=python)", cases, facade_call)
    total_ok, total_n = total_ok + ok, total_n + n

    del os.environ["RUNE_MEMORY_CORE"]
    ok, n = run_layer("facade live (Rust core through ms_loader)", cases, facade_call)
    total_ok, total_n = total_ok + ok, total_n + n

    ok, n = run_layer("kernel direct (FFI envelope)", cases, kernel_call)
    total_ok, total_n = total_ok + ok, total_n + n

    ok, n = db_level_check()
    total_ok, total_n = total_ok + ok, total_n + n

    ok, n = fuzz_check()
    total_ok, total_n = total_ok + ok, total_n + n

    print(f"parity: {total_ok}/{total_n} match")
    return 0 if total_ok == total_n else 1


if __name__ == "__main__":
    sys.exit(main())