rune/scripts/gen_memory_store_gate_truth.py

#!/usr/bin/env python3
"""gen_memory_store_gate_truth.py — memory-store batch 4 truth battery
(contract BUILDER-2026-10-07T19:47:37Z-1cf1dc).

The LIVE oracle: the dense-side meta-judgement extracted from recall()
(lines 655-676) exercised against a FIXTURE STORE (sqlite entries + df
counts), plus the entries-side fusion vs the chunk-side one (identity
check), plus the abstain constants.

Usage: python3 scripts/gen_memory_store_gate_truth.py   (from repo root)
"""

import json
import sqlite3
import sys
import tempfile
from pathlib import Path

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

import memory_store as ms  # noqa: E402


def make_fixture_store(db_path):
    """Entries table + df data mirroring recall()'s df-of-word query."""
    db = sqlite3.connect(db_path)
    db.row_factory = sqlite3.Row
    db.execute("CREATE TABLE entries (id INTEGER PRIMARY KEY, text TEXT, status TEXT)")
    rows = [
        (1, "the curator gates memory writes through worthiness"),
        (2, "black holes bend light near the event horizon"),
        (3, "black coffee in the morning keeps nick running"),
        (4, "the spine rebuild parity proofs live in store"),
        (5, "memory writes require the operator approval gate"),
        (6, "unrelated filler text about weather patterns"),
        (7, "curator provenance recovery from odysseus"),
        (8, "another filler row to pad the document frequency"),
    ]
    db.executemany("INSERT INTO entries VALUES (?,?,?)",
                   [(i, t, "active") for (i, t) in rows])
    db.commit()
    return db


def main():
    tmp = Path(tempfile.mkdtemp(prefix="ms-gate-truth-"))
    db = make_fixture_store(str(tmp / "store.db"))
    total = db.execute("SELECT COUNT(*) AS n FROM entries WHERE status='active'").fetchone()["n"]

    def df_of(word):
        return db.execute(
            "SELECT COUNT(*) AS n FROM entries WHERE status='active' AND text LIKE ?",
            (f"%{word}%",),
        ).fetchone()["n"]

    # replicate recall()'s dense-gate block verbatim
    def oracle_gate(query, hits):
        q_content = {t for t in query.lower().split() if len(t) > 3 and t not in ms._STOPWORDS}
        dense = []
        for rowid, cos in hits:
            etxt = (db.execute("SELECT text FROM entries WHERE id=?", (rowid,)).fetchone()["text"] or "").lower()
            distinctive = [w for w in q_content if w in etxt and df_of(w) / total < 0.05]
            if distinctive:
                dense.append((rowid, cos))
            elif not q_content or cos >= 0.80:
                dense.append((rowid, cos))
        return dense

    cases = []

    def add(fn, name, inp, expected):
        cases.append({"fn": fn, "name": name, "input": inp, "expected": expected})

    # gate: shared DISTINCTIVE word (curator: df=2/8=0.25? no — check)
    # df(curator) = rows 1,7 = 2 → 0.25 ≥ 0.05 → NOT distinctive!
    # The oracle's threshold means common-store words fail; only truly
    # rare words pass. Fixture exercise:
    for name, query, hits in [
        ("distinctive_shared", "worthiness gates", [(1, 0.9), (2, 0.85)]),
        ("common_word_falls_back", "black", [(2, 0.86), (3, 0.79)]),
        ("strong_cos_no_share", "unrelated query terms", [(6, 0.91), (4, 0.70)]),
        ("no_content_words", "the was and", [(1, 0.60), (6, 0.95)]),
    ]:
        add("dense_gate", name,
            {"query": query, "hits": [[r, c] for r, c in hits], "total": total},
            {"kept": [[r, c] for r, c in oracle_gate(query, hits)],
             "q_content": sorted({t for t in query.lower().split() if len(t) > 3 and t not in ms._STOPWORDS}),
             "df": {w: df_of(w) for w in query.lower().split() if len(w) > 3}})

    # entries-side fusion identity with the chunk-side one
    dense = [[1, 0.9], [2, 0.8]]
    bm25 = [[2, 10.0], [3, 4.0]]
    add("fusion_identity", "entries_eq_chunks",
        {"dense": dense, "bm25": bm25},
        {"fusion": ms._py_rank_fusion(dense, bm25)})

    add("constants", "abstain",
        {},
        {"budget": ms.RECALL_BUDGET, "min_score": ms.RECALL_MIN_SCORE})

    import shutil
    shutil.rmtree(tmp, ignore_errors=True)

    out = {
        "oracle": "src/rune/memory_store.py recall() dense gate + _py_rank_fusion",
        "generated": "2026-10-07",
        "fixture_df_note": "df via LIKE-substring over active entries; threshold 0.05",
        "cases": cases,
    }
    target = REPO / "src/go/memory-store-deep" / "truth_gate.json"
    target.write_text(json.dumps(out, indent=1, ensure_ascii=False) + "\n")
    print(f"wrote {target} ({len(cases)} cases)")


if __name__ == "__main__":
    main()