rune/scripts/check_model_profiles_parity.py

#!/usr/bin/env python3
"""check_model_profiles_parity.py — live oracle-vs-Go parity sweep for the
model_profiles gap closure (#2479, contract factory-gap-mp).

The frozen envelope (frozen_mp_truth.json, 36 cases) pins the seam
surface byte-identically. THIS script drives the LIVE Python oracle
(src/rune/model_profiles.py) and the Go binary across the broader
surface the frozen corpus samples: every model family + unknown,
parameter/quantization edge shapes, the sanitization matrix, the
to_dict/from_dict lens (incl. unknown-key filtering and defaults),
build assembly across families (uncensored tier, display/size
overrides), record sequences (dedup, empty-topic, success mix),
prompt-context shapes, the scan sweep against a canned Ollama server
(fresh + existing-preservation + banker's rounding + nameless skip +
dead-port degrade), and save/load round-trips (byte-identical raw
text). Timestamps are pinned via the binary's "now" seam and the
oracle's module-global datetime patch; stores are per-case hermetic
temp dirs on both sides.

Usage: PYTHONPATH=src python3 scripts/check_model_profiles_parity.py
  [--bin src/go/model-profiles/model-profiles]
Exit 0 on 100% parity, 1 with the first divergence.
"""

import argparse
import dataclasses
import json
import os
import shutil
import subprocess
import sys
import tempfile
import threading
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO / "src"))

import rune.model_profiles as mp  # noqa: E402

FROZEN_NOW = "2026-10-13T09:41:23.123456"

NAMES = [
    "qwen2.5:7b", "Qwen3:14b", "llama3.1:8b", "mistral-7b", "gemma2:9b",
    "phi3:mini", "deepseek-r1:32b", "DavidAU-Qwen-14b", "some-new-model",
    "library/qwen3:14b", "a?b:c/d", "model-70B-GGUF", "Qwen2.5-0.5b",
    "dolphin-uncensored:7b", "qwen2.5:14b-instruct-q5_K_M",
    "llama-3-8b-Q4_K_M", "deepseek-r1:70b-IQ2_M", "phi-4-14b-int4",
    "model-3B-FP16", "tiny-INT8", "x-GGUF-fp32",
]

QUANT_NAMES = [
    "m-Q4_K_M", "m-IQ3_XS", "m-q5_k_m", "m-iq2_m", "m-GGUF", "m-gguf",
    "m-FP16", "m-FP32", "m-INT8", "m-INT4", "m-none",
]

BUILD_CASES = [
    ("qwen3:14b", {"display_name": "Qwen3 14B", "size_gb": 9.0}),
    ("mistral-7b", None),
    ("llama3.1:8b", {"display_name": "L", "size_gb": 4.5}),
    ("gemma2:9b", {"display_name": "G"}),
    ("phi3:mini", {"size_gb": 2.1}),
    ("deepseek-r1:32b", {"display_name": "DS", "size_gb": 19.2}),
    ("DavidAU-X:7b", None),
    ("dolphin-uncensored:7b", None),
    ("mystery-model", None),
    ("mystery-model", {"display_name": "M", "size_gb": 1.5}),
]

RECORD_SEQS = [
    [[True, ""]],
    [[False, "refuses weapons"]],
    [[False, "t1"], [False, "t1"]],
    [[False, ""]],
    [[True, ""], [False, "a"], [True, ""], [False, "b"], [False, "a"]],
    [[False, "x"], [False, "y"], [False, "z"]],
]

CTX_SEEDS = [
    {"name": "full:1b", "strengths": ["s1", "s2", "s3", "s4"],
     "prompting_tips": ["tip one", "tip two"],
     "known_refusal_topics": ["r1", "r2", "r3", "r4"], "safety_tier": 2},
    {"name": "tips:1b", "prompting_tips": ["only tip"]},
    {"name": "tier:1b", "safety_tier": 1},
    {"name": "refusals:1b", "known_refusal_topics": ["r1"]},
]

SCAN_MODELS = [
    {"name": "qwen2.5:7b", "size": 7516192768},
    {"name": "banker-lo:2b", "size": 2415919104},
    {"name": "banker-hi:2b", "size": 2952790016},
    {"name": "llama3.1:8b", "size": 4831838208},
    {"size": 999},
    {"name": "sizeless:1b"},
]


class _TagsHandler(BaseHTTPRequestHandler):
    payload = b"{}"

    def do_GET(self):
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(self.payload)))
        self.end_headers()
        self.wfile.write(self.payload)

    def log_message(self, *a):
        pass


def canned_server(fixture):
    handler = type("H", (_TagsHandler,),
                   {"payload": json.dumps(fixture).encode()})
    srv = ThreadingHTTPServer(("127.0.0.1", 0), handler)
    threading.Thread(target=srv.serve_forever, daemon=True).start()
    return srv


def run_go(binpath, req):
    p = subprocess.run([str(binpath)], input=json.dumps(req),
                       capture_output=True, text=True)
    if p.returncode != 0:
        raise RuntimeError(
            f"go op {req.get('op')} exited {p.returncode}: {p.stdout} {p.stderr}")
    return json.loads(p.stdout)["result"]


def fresh_store():
    d = tempfile.mkdtemp(prefix="mp_parity_")
    return os.path.join(d, ".rune", "model_profiles")


def oracle_in_store(store):
    """Point the oracle at a hermetic store (PROFILES_DIR is module-global)."""
    mp.PROFILES_DIR = store


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--bin", default=str(
        REPO / "src/go/model-profiles/model-profiles"))
    args = ap.parse_args()
    binpath = Path(args.bin)
    if not binpath.exists():
        binpath = Path("/tmp/mp-go")
    fails = total = 0

    def canon(v):
        # Integral-float law (the context_broker precedent): the oracle's
        # build path assigns int 0 where the dataclass declares float
        # (size_gb=0 vs 0.0) — numerically identical, textually distinct.
        # Canonicalize integral floats to int on BOTH sides before compare.
        if isinstance(v, float) and v.is_integer():
            return int(v)
        if isinstance(v, dict):
            return {k: canon(x) for k, x in v.items()}
        if isinstance(v, list):
            return [canon(x) for x in v]
        return v

    def check(name, oracle_val, go_val):
        nonlocal fails, total
        total += 1
        if json.dumps(canon(oracle_val), sort_keys=True) != json.dumps(canon(go_val), sort_keys=True):
            fails += 1
            print(f"DIVERGE {name}:\n  oracle: {json.dumps(oracle_val)[:300]}\n"
                  f"  go:     {json.dumps(go_val)[:300]}")

    # pure detection surface (no store, no clock)
    for n in NAMES:
        check(f"family:{n}", mp.detect_family(n),
              run_go(binpath, {"op": "detect_family", "name": n}))
    for n in NAMES + QUANT_NAMES:
        check(f"params:{n}", mp.detect_parameters(n),
              run_go(binpath, {"op": "detect_parameters", "name": n}))
        check(f"quant:{n}", mp.detect_quantization(n),
              run_go(binpath, {"op": "detect_quantization", "name": n}))
    for inter, succ in [(0, 0), (1, 1), (10, 7), (4, 0), (3, 2)]:
        p = mp.ModelProfile(name="x", interactions=inter, successes=succ)
        check(f"success_rate:{inter}/{succ}", p.success_rate,
              run_go(binpath, {"op": "success_rate",
                               "interactions": inter, "successes": succ}))

    # sanitization matrix (compare file NAMES — store roots differ)
    for n in NAMES:
        store = fresh_store()
        oracle_in_store(store)
        oracle_name = os.path.basename(mp.get_profile_path(n))
        go_name = run_go(binpath, {"op": "get_profile_path",
                                   "name": n, "store": store})
        check(f"path:{n}", oracle_name, go_name)

    # to_dict / from_dict lens
    full = mp.ModelProfile(
        name="qwen2.5:7b", display_name="Q ✓", family="qwen",
        parameters="7B", quantization="Q4_K_M", size_gb=4.7,
        strengths=["a", "b"], weaknesses=["c"], prompting_tips=["t1"],
        safety_tier=2, known_refusal_topics=["r1"],
        thinking_format="<think>", response_style="concise",
        context_window=32768, optimal_temperature=0.7,
        interactions=10, successes=7, failures=3,
        last_used="2026-10-12T00:00:00+00:00",
        created="2026-10-01T00:00:00+00:00")
    check("to_dict:full", full.to_dict(),
          run_go(binpath, {"op": "to_dict", **full.to_dict()}))
    for doc in ({"name": "n", "bogus": 1, "context_window": 8192},
                {"name": "n2"},
                dataclasses.asdict(full)):
        check(f"from_dict:{doc.get('name')}",
              dataclasses.asdict(mp.ModelProfile.from_dict(doc)),
              run_go(binpath, {"op": "from_dict", **doc}))

    # build assembly (pin the oracle clock; binary takes "now")
    import datetime as dtmod
    real_dt = dtmod.datetime

    class FrozenDT(real_dt):
        @classmethod
        def now(cls, tz=None):
            base = real_dt(2026, 10, 13, 9, 41, 23, 123456)
            return base.replace(tzinfo=tz) if tz is not None else base

    mp.datetime = FrozenDT
    try:
        for name, info in BUILD_CASES:
            oracle = dataclasses.asdict(
                mp.build_profile_from_ollama(name, info))
            go = run_go(binpath, {"op": "build_profile", "name": name,
                                  "model_info": info, "now": FROZEN_NOW + "+00:00"})
            check(f"build:{name}", oracle, go)
    finally:
        mp.datetime = real_dt

    # record sequences + prompt context (hermetic store each, both
    # sides replay the same sequence from empty, then compare docs)
    mp.datetime = FrozenDT
    try:
        for i, seq in enumerate(RECORD_SEQS):
            ostore, gstore = fresh_store(), fresh_store()
            oracle_in_store(ostore)
            for success, topic in seq:
                mp.record_interaction(f"m{i}:1b", success, topic)
            for success, topic in seq:
                run_go(binpath, {"op": "record", "name": f"m{i}:1b",
                                 "success": success, "topic": topic,
                                 "store": gstore, "now": FROZEN_NOW + "+00:00"})
            oracle_in_store(ostore)
            o_doc = dataclasses.asdict(mp.load_profile(f"m{i}:1b"))
            g_doc = run_go(binpath, {"op": "load_profile",
                                     "name": f"m{i}:1b", "store": gstore})
            check(f"record-x[{i}]", o_doc, g_doc)
    finally:
        mp.datetime = real_dt

    for seed in CTX_SEEDS:
        name = seed["name"]
        ostore, gstore = fresh_store(), fresh_store()
        oracle_in_store(ostore)
        mp.save_profile(mp.ModelProfile.from_dict(seed))
        go_state = dict(seed)
        run_go(binpath, {"op": "save_profile", "store": gstore, **go_state})
        oracle_in_store(ostore)
        o_ctx = mp.get_prompt_context(name)
        g_ctx = run_go(binpath, {"op": "prompt_context",
                                 "name": name, "store": gstore})
        check(f"ctx:{name}", o_ctx, g_ctx)

    # save/load round-trip byte parity (raw file text)
    for doc in (dataclasses.asdict(full), {"name": "plain-model"},
                {"name": "d:1b", "family": "deepseek", "size_gb": 0.0,
                 "safety_tier": 3, "optimal_temperature": 0.6}):
        ostore, gstore = fresh_store(), fresh_store()
        oracle_in_store(ostore)
        mp.save_profile(mp.ModelProfile.from_dict(doc))
        with open(mp.get_profile_path(doc["name"])) as f:
            o_raw = f.read()
        g_raw = run_go(binpath, {"op": "save_profile",
                                 "store": gstore, **doc})
        check(f"save-raw:{doc['name']}", o_raw, g_raw)

    # scan sweep (canned server; existing-preservation; dead degrade)
    for label, fixture, seed in (
            ("fresh", {"models": SCAN_MODELS}, None),
            ("existing", {"models": SCAN_MODELS[:3]}, {"name": "qwen2.5:7b",
             "family": "qwen", "interactions": 5, "successes": 4,
             "known_refusal_topics": ["old"], "size_gb": 1.0}),
            ("empty", {"models": []}, None),
            ("nameless", {"models": [{"size": 1}, {}]}, None)):
        srv = canned_server(fixture)
        url = f"http://127.0.0.1:{srv.server_address[1]}"
        try:
            ostore, gstore = fresh_store(), fresh_store()
            oracle_in_store(ostore)
            if seed:
                mp.save_profile(mp.ModelProfile.from_dict(seed))
                # seed the GO store with the same file the oracle has
                os.makedirs(gstore, exist_ok=True)
                for f in os.listdir(ostore):
                    shutil.copy(os.path.join(ostore, f),
                                os.path.join(gstore, f))
            saved_url = mp.OLLAMA_URL
            mp.OLLAMA_URL = url
            mp.datetime = FrozenDT
            try:
                o_profiles = [dataclasses.asdict(p)
                              for p in mp.scan_all_models()]
            finally:
                mp.OLLAMA_URL = saved_url
                mp.datetime = real_dt
            g_profiles = run_go(binpath, {"op": "scan", "store": gstore,
                                          "ollama_url": url,
                                          "now": FROZEN_NOW + "+00:00"})
            check(f"scan:{label}", o_profiles, g_profiles)
        finally:
            srv.shutdown()

    # dead-port degrade
    ostore, gstore = fresh_store(), fresh_store()
    oracle_in_store(ostore)
    saved_url = mp.OLLAMA_URL
    mp.OLLAMA_URL = "http://127.0.0.1:1"
    try:
        o_dead = mp.scan_all_models()
    finally:
        mp.OLLAMA_URL = saved_url
    g_dead = run_go(binpath, {"op": "scan", "store": gstore,
                              "ollama_url": "http://127.0.0.1:1"})
    check("scan:dead", o_dead, g_dead)

    print(f"parity: {total - fails}/{total} match")
    return 1 if fails else 0


if __name__ == "__main__":
    raise SystemExit(main())