feat(warm_router): close STATE-family gaps — dichotomy wave

DB/STATE/PERSISTENCE family lane (contract factory-heavy-wr-state):
name-preserving port of the 11 unmatched family fns into
src/go/warm-router/gaps_state.go — _ensure_use_history_table (verbatim
DDL trio + the latched created-flag no-op), record_outcome,
record_neuron_fire, _db (read-only URI), _aaak (compress seam, topic
pin, ''-on-failure), write_neuron (upsert plan: UPDATE vs the full
18-column INSERT tuple, store://neuron URI), _touch (default entry
shape in literal key order), _session_state, load_state (any-shape
pass-through + default degrade), save_state (json.dumps indent=2
byte contract), _derive_triggers (first-occurrence tie order, [:8]
cap), list_neurons (or-falsiness defaults, RUNE-safe 80-char summary
truncation + token cost per the str[:n] quirk law), promote_active
(keep-resident ≤90 CHARS, dry-run count), handle_hook (message
or-chain, >2-char gate, 120-CHAR error truncation, AttributeError
crash-class marker). SQLite stays facade-side per the landed crate
seams: SQL text + params + decisions are the contract, connections
injected SQLExecers. main gains the state-scenario parity subcommand.

Frozen truth: 46-case envelope via the shared freeze harness +
scripts/gen_warm_router_state_truth.py wrapper shims (recording
sqlite3.connect, patched wr._db/STATE, clock pinned from row args —
_touch binds datetime at module level, pinned directly;
write_neuron's function-local import reached by the rig). --verify
byte-identical, fresh regen sha-stable, no absolute paths
(STATE_STORE_DB placeholder). Battery truth_state_test.go replays
every row through the SAME runStateOp the CLI rides (one execution
path for battery + live parity). Live parity 46/46 via
scripts/check_warm_router_state_parity.py (order-sensitive rendered
JSON). go build/gofmt/vet/test clean; module pytest green delegation-
live (MS_RUST_LIB); pinned-python divergence identical to the main
checkout (memory_scoring fallback retired — control evidence
committed). dichotomy audit warm_router 17/33 -> 28/33 fns (5
remaining are routing-family, landed separately as 9102fc2b).

scripts/check_warm_router_state_parity.py

@@ -0,0 +1,105 @@
+#!/usr/bin/env python3
+"""check_warm_router_state_parity.py — LIVE parity for the warm_router
+DB/STATE/PERSISTENCE family GO port (#2479 dichotomy gap-close, contract
+factory-heavy-wr-state): drives the Python oracle (src/rune/warm_router.py
+through the generator's wrapper shims — SQLite recorded, never real) and
+the GO binary's `state-scenario` subcommand through IDENTICAL ops, then
+diffs the results as ORDER-SENSITIVE rendered JSON (Python json.dumps
+compact separators on both sides; key order and number shape are part of
+the contract — the battery replays the same path via runStateOp).
+
+Usage: python3 scripts/check_warm_router_state_parity.py [go-binary]
+  (default builds: cd src/go/warm-router && go build -o /tmp/warm-router-parity .)
+"""
+import json
+import subprocess
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / "scripts"))
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "src" / "rune"))
+
+import gen_warm_router_state_truth as G  # noqa: E402  (rig + shims)
+
+
+def case_to_op(fn, args):
+    """The argToOp mirror (positional case args → scenario-op fields)."""
+    def get(i):
+        return args[i] if i < len(args) else None
+    if fn == "_wr_state_ensure":
+        return {"op": "ensure_use_history_table", "twice": get(0)}
+    if fn == "_wr_state_record_outcome":
+        return {"op": "record_outcome", "session_id": get(0), "outcome": get(1), "fresh": get(2)}
+    if fn == "_wr_state_record_neuron_fire":
+        return {"op": "record_neuron_fire", "entry_id": get(0), "score": get(1),
+                "session_id": get(2), "outcome": get(3), "fresh": get(4)}
+    if fn == "_wr_state_db":
+        return {"op": "db", "store_db": get(0)}
+    if fn == "_wr_state_aaak":
+        return {"op": "aaak", "text": get(0), "mode": get(1)}
+    if fn == "_wr_state_write_neuron":
+        return {"op": "write_neuron", "slug": get(0), "triggers": get(1), "body": get(2),
+                "importance": get(3), "existing_id": get(4), "aaak": get(5), "now": get(6)}
+    if fn == "_wr_state_touch":
+        return {"op": "touch", "state": get(0), "slug": get(1), "now": get(2)}
+    if fn == "_wr_state_session_state":
+        return {"op": "session_state", "content": get(0)}
+    if fn == "_wr_state_load_state":
+        return {"op": "load_state", "content": get(0)}
+    if fn == "_wr_state_save_state":
+        return {"op": "save_state", "state": get(0)}
+    if fn == "_wr_state_derive_triggers":
+        return {"op": "derive_triggers", "text": get(0), "existing": get(1)}
+    if fn == "_wr_state_list_neurons":
+        return {"op": "list_neurons", "rows": get(0)}
+    if fn == "_wr_state_promote_active":
+        return {"op": "promote_active", "rows": get(0), "dry_run": get(1)}
+    if fn == "_wr_state_handle_hook":
+        return {"op": "handle_hook", "event": get(0), "data": get(1), "route_mode": get(2),
+                "fired": get(3), "total": get(4), "route_error": get(5)}
+    raise ValueError(f"unmapped fn {fn}")
+
+
+def render(v):
+    return json.dumps(v, ensure_ascii=True)
+
+
+def main():
+    binary = sys.argv[1] if len(sys.argv) > 1 else "/tmp/warm-router-parity"
+    if len(sys.argv) < 2:
+        subprocess.run(["go", "build", "-o", binary, "."],
+                       cwd=REPO / "src" / "go" / "warm-router", check=True)
+
+    cases = G._cases()
+    ops = [case_to_op(fn, list(args)) for fn, args, _kw in cases]
+    proc = subprocess.run([binary, "state-scenario"], input=json.dumps({"ops": ops}),
+                          capture_output=True, text=True, timeout=60)
+    if proc.returncode != 0:
+        print("GO BINARY FAILED:", proc.returncode)
+        print(proc.stderr[:2000])
+        return 1
+    go_results = json.loads(proc.stdout)["results"]
+
+    fails = 0
+    for i, (fn, args, kwargs) in enumerate(cases):
+        py_out = getattr(G, fn)(*args, **kwargs)
+        go_out = go_results[i]
+        if isinstance(go_out, dict) and "__error__" in go_out:
+            print(f"case {i} {fn}: GO ERROR {go_out['__error__']}")
+            fails += 1
+            continue
+        if render(go_out) != render(py_out):
+            print(f"case {i} {fn} DIVERGE:\n  go: {render(go_out)[:300]}\n  py: {render(py_out)[:300]}")
+            fails += 1
+    total = len(cases)
+    if fails:
+        print(f"PARITY FAILED: {fails}/{total} divergences")
+        return 1
+    print(f"PARITY OK: {total}/{total} live state-family cases byte-identical (go vs python oracle)")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/gen_warm_router_state_truth.py

@@ -0,0 +1,489 @@
+#!/usr/bin/env python3
+"""gen_warm_router_state_truth.py — warm_router DB/STATE/PERSISTENCE
+family truth battery (#2479 dichotomy wave, GO side, contract
+factory-heavy-wr-state).
+
+Covers the unmatched family functions: _ensure_use_history_table,
+record_outcome, record_neuron_fire, _db, _aaak, write_neuron, _touch,
+_session_state, load_state, save_state, _derive_triggers, list_neurons,
+promote_active, handle_hook.
+
+Uses the SHARED harness (scripts/freeze_truth.py) with the wrapper-shim
+law (persona_definition pattern): SQLite stays facade-side, so every
+capture runs the LIVE oracle function against a recording sqlite3.connect
+(the verbatim statement sequence + parameter tuples + commit points are
+the contract) or a hermetic wr.STATE / patched wr._db — never the real
+store. `now` is always an argument (no-clock law): write_neuron/touch pin
+the clock from the row's now value via freeze_rigs.pin_clock (the
+function-local `from datetime import datetime` quirk).
+
+No absolute paths in truth (REPO_ROOT/placeholder law): the _db row rides
+the placeholder store path "STATE_STORE_DB".
+
+Verify (ships its own --verify — same byte-compare law as the harness,
+plus the shim-module registration the stock CLI cannot do):
+
+    python3 scripts/gen_warm_router_state_truth.py --verify
+
+Usage: python3 scripts/gen_warm_router_state_truth.py   (from repo root)
+"""
+
+import contextlib
+import copy
+import json
+import os
+import sys
+import tempfile
+import types
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parent.parent
+sys.path.insert(0, str(REPO / "scripts"))
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "src" / "rune"))
+
+from freeze_truth import freeze, write_envelope  # noqa: E402
+from freeze_rigs import pin_clock  # noqa: E402
+
+FROZEN_NOW = "2026-10-14T21:30:00+00:00"
+
+# Hermetic env BEFORE the first oracle import (memory_env derives the
+# memory dir at import time — pin_home law).
+_BOOT_TMP = tempfile.mkdtemp(prefix="wr-state-boot-")
+os.environ.setdefault("RUNE_MEMORY_DIR", _BOOT_TMP)
+
+import warm_router as wr  # noqa: E402
+
+import sqlite3 as _sq  # noqa: E402
+
+
+# ── recording rig (facade-side SQLite seam) ───────────────────────
+
+class RecCur:
+    def __init__(self, conn):
+        self._conn = conn
+
+    def fetchone(self):
+        return self._conn._fetchone
+
+    def fetchall(self):
+        return self._conn._fetchall
+
+    @property
+    def lastrowid(self):
+        return self._conn._lastrowid
+
+
+class RecDB:
+    def __init__(self, events, fetchone=None, fetchall=None, lastrowid=0):
+        self._events = events
+        self._fetchone = fetchone
+        self._fetchall = fetchall
+        self._lastrowid = lastrowid
+        self.row_factory = None
+
+    def execute(self, sql, params=()):
+        self._events.append({"sql": sql, "params": list(params)})
+        return RecCur(self)
+
+    def commit(self):
+        self._events.append({"commit": True})
+
+    def close(self):
+        pass
+
+
+@contextlib.contextmanager
+def _patch_sqlite(events, fetchone=None, fetchall=None, lastrowid=0):
+    saved = _sq.connect
+
+    def connect(*_a, **_k):
+        return RecDB(events, fetchone, fetchall, lastrowid)
+
+    _sq.connect = connect
+    try:
+        yield
+    finally:
+        _sq.connect = saved
+
+
+@contextlib.contextmanager
+def _fake_module(name, mod):
+    saved = sys.modules.get(name)
+    sys.modules[name] = mod
+    try:
+        yield
+    finally:
+        if saved is None:
+            sys.modules.pop(name, None)
+        else:
+            sys.modules[name] = saved
+
+
+def _state_path(content):
+    """A hermetic wr.STATE: absent file for content=None, else written."""
+    tmp = tempfile.mkdtemp(prefix="wr-state-")
+    path = os.path.join(tmp, "warm_state.json")
+    if content is not None:
+        with open(path, "w", encoding="utf-8") as fh:
+            fh.write(content)
+    return path
+
+
+# ── wrappers (the frozen fn names) ────────────────────────────────
+
+def _wr_state_ensure(twice):
+    events = []
+    wr._use_history_table_created = False
+    with _patch_sqlite(events):
+        wr._ensure_use_history_table()
+        if twice:
+            wr._ensure_use_history_table()
+    return events
+
+
+def _wr_state_record_outcome(session_id, outcome, fresh):
+    events = []
+    wr._use_history_table_created = not fresh
+    with _patch_sqlite(events):
+        wr.record_outcome(session_id, outcome=outcome)
+    return events
+
+
+def _wr_state_record_neuron_fire(entry_id, score, session_id, outcome, fresh):
+    events = []
+    wr._use_history_table_created = not fresh
+    with _patch_sqlite(events):
+        wr.record_neuron_fire(entry_id, score=score, session_id=session_id,
+                              outcome=outcome)
+    return events
+
+
+def _wr_state_db(store_db):
+    captured = {}
+    events = []
+    wr.STORE_DB = store_db
+
+    def connect(*a, **k):
+        captured["args"] = list(a)
+        captured["kwargs"] = dict(k)
+        return RecDB(events)
+
+    saved = _sq.connect
+    _sq.connect = connect
+    try:
+        wr._db()
+    finally:
+        _sq.connect = saved
+    return {"uri": captured["args"][0]}
+
+
+def _wr_state_aaak(text, mode):
+    fake = types.ModuleType("aaak")
+    if mode == "ok":
+        fake.compress = lambda t, topic: "AAAK:" + t
+    else:
+        def _boom(t, topic):
+            raise RuntimeError("compress failed")
+        fake.compress = _boom
+    with _fake_module("aaak", fake):
+        out = wr._aaak(text)
+    return {"out": out}
+
+
+def _wr_state_write_neuron(slug, triggers, body, importance, existing_id, aaak, now):
+    pin_clock(now)
+    events = []
+    auto = []
+    saved_aaak = wr._aaak
+    wr._aaak = lambda _t: aaak
+    fake_ms = types.ModuleType("memory_store")
+    fake_ms._auto_associate = lambda dbw, eid, b, topic="warm", source="warm_router": \
+        auto.append([eid, b, topic, source])
+    fetch = (existing_id,) if existing_id is not None else None
+    try:
+        with _fake_module("memory_store", fake_ms):
+            with _patch_sqlite(events, fetchone=fetch, lastrowid=4242):
+                uri = wr.write_neuron(slug, triggers, body, importance=importance)
+    finally:
+        wr._aaak = saved_aaak
+    return {"uri": uri, "events": events, "auto_associate": auto}
+
+
+def _wr_state_touch(state, slug, now):
+    # _touch binds datetime at MODULE level (unlike write_neuron's
+    # function-local import), so the rig's datetime-module patch never
+    # reaches it — pin the module attr directly.
+    frozen_cls = pin_clock(now)
+    saved_dt = wr.datetime
+    wr.datetime = frozen_cls
+    try:
+        st = copy.deepcopy(state)
+        wr._touch({"slug": slug}, state=st)
+    finally:
+        wr.datetime = saved_dt
+    return st
+
+
+def _wr_state_session_state(content):
+    wr.STATE = _state_path(content)
+    return wr._session_state()
+
+
+def _wr_state_load_state(content):
+    wr.STATE = _state_path(content)
+    return wr.load_state()
+
+
+def _wr_state_save_state(state):
+    tmp = tempfile.mkdtemp(prefix="wr-save-")
+    wr.STATE = os.path.join(tmp, "warm_state.json")
+    wr.save_state(copy.deepcopy(state))
+    with open(wr.STATE, encoding="utf-8") as fh:
+        return fh.read()
+
+
+def _wr_state_derive_triggers(text, existing):
+    return wr._derive_triggers(text, existing=existing)
+
+
+def _wr_state_list_neurons(rows):
+    class FakeDB:
+        def execute(self, *_a, **_k):
+            return self
+
+        def fetchall(self):
+            return [tuple(r) for r in rows]
+
+        def close(self):
+            pass
+
+    saved = wr._db
+    wr._db = lambda: FakeDB()
+    try:
+        neurons = wr.list_neurons()
+    finally:
+        wr._db = saved
+    return neurons
+
+
+def _wr_state_promote_active(rows, dry_run):
+    events = []
+    preset = [dict(zip(("id", "text", "topic"), r)) for r in rows]
+    with _patch_sqlite(events, fetchall=preset):
+        n = wr.promote_active(dry_run=dry_run)
+    return {"promoted": n, "events": events}
+
+
+def _wr_state_handle_hook(event, data, route_mode, fired, total, route_error):
+    calls = []
+
+    def fake_route(message, max_tokens=wr.DEFAULT_MAX_TOKENS, session_id=""):
+        calls.append(message)
+        if route_mode == "error":
+            raise RuntimeError(route_error)
+        return ([{"slug": s} for s in fired], total, [])
+
+    saved = wr.route
+    wr.route = fake_route
+    try:
+        out = wr.handle_hook(event, copy.deepcopy(data))
+    except AttributeError:  # the non-string-message crash class
+        out = {"__crash__": "AttributeError"}
+    finally:
+        wr.route = saved
+    if calls:
+        out = dict(out)
+        out["__calls__"] = calls
+    return out
+
+
+# ── case matrix ───────────────────────────────────────────────────
+
+NOW = FROZEN_NOW
+
+
+def _cases():
+    C = []
+
+    # _ensure_use_history_table — DDL trio + commit; the latched second
+    # call issues NOTHING (not even a connect).
+    C.append(("_wr_state_ensure", [True], {}))
+    C.append(("_wr_state_ensure", [False], {}))
+
+    # record_outcome — fresh (DDL + UPDATE) vs warm flag (UPDATE only).
+    C.append(("_wr_state_record_outcome", ["sess-1", "helpful", True], {}))
+    C.append(("_wr_state_record_outcome", ["sess-1", "ignored", False], {}))
+
+    # record_neuron_fire — fresh insert with score; warm default args.
+    C.append(("_wr_state_record_neuron_fire", [7, 0.75, "sess-2", "unknown", True], {}))
+    C.append(("_wr_state_record_neuron_fire", [7, 0.0, "sess-2", "helpful", False], {}))
+
+    # _db — the read-only URI (placeholder store path, no absolute paths).
+    C.append(("_wr_state_db", ["STATE_STORE_DB"], {}))
+
+    # _aaak — topic pin is inside the seam; ok + failure ("").
+    C.append(("_wr_state_aaak", ["the warm neuron body", "ok"], {}))
+    C.append(("_wr_state_aaak", ["the warm neuron body", "fail"], {}))
+
+    # write_neuron — update path (stripped body, joined triggers, ts).
+    C.append(("_wr_state_write_neuron",
+              ["operating-gates", ["gate", "exit code"], "  Never pipe gates through tail.  ",
+               0.9, 7, "gates rule summary", NOW], {}))
+    # insert path (full 18-column tuple, lastrowid entry).
+    C.append(("_wr_state_write_neuron",
+              ["human-prefs", ["coffee"], "Prefer coconut milk.",
+               0.5, None, "", NOW], {}))
+    # empty triggers join; unicode body strips per Unicode whitespace.
+    C.append(("_wr_state_write_neuron",
+              ["project-café", [], "café au lait notes",
+               0.75, 11, "café", NOW], {}))
+
+    # _touch — default entry shape + fires bump + pinned last_fired.
+    C.append(("_wr_state_touch", [{}, "operating-gates", NOW], {}))
+    # existing entry (fires 2 -> 3), key order preserved.
+    C.append(("_wr_state_touch",
+              [{"neurons": {"human-prefs": {"fires": 2, "last_fired": "2026-10-13T00:00:00+00:00",
+                                            "promotion_streak": 1, "demotion_streak": 0}}},
+               "human-prefs", NOW], {}))
+    # neurons setdefault lands LAST when absent (insertion-order pin).
+    C.append(("_wr_state_touch",
+              [{"divides": [], "sessions": {"s1": {"primed": {}, "turns": 3}}},
+               "new-slug", NOW], {}))
+
+    # _session_state — missing file default + sessions setdefault;
+    # existing state (sessions appended last).
+    C.append(("_wr_state_session_state", [None], {}))
+    C.append(("_wr_state_session_state",
+              ['{"neurons": {"a": {"fires": 1, "last_fired": "", "promotion_streak": 0, "demotion_streak": 0}}, "divides": []}'],
+              {}))
+
+    # load_state — valid, missing, bad JSON (default), non-dict top level
+    # passes through, empty object.
+    C.append(("_wr_state_load_state",
+              ['{"neurons": {"a": {"fires": 1, "last_fired": "x", "promotion_streak": 0, "demotion_streak": 0}}, "divides": ["d"]}'],
+              {}))
+    C.append(("_wr_state_load_state", [None], {}))
+    C.append(("_wr_state_load_state", ["{nope"], {}))
+    C.append(("_wr_state_load_state", ["[1, 2]"], {}))
+    C.append(("_wr_state_load_state", ["{}"], {}))
+
+    # save_state — indent=2, insertion order, ensure_ascii (\u00e9),
+    # floats keep repr, no trailing newline.
+    C.append(("_wr_state_save_state", [{"neurons": {}, "divides": []}], {}))
+    C.append(("_wr_state_save_state",
+              [{"zebra": "last", "alpha": {"nested": [1, 2.5, True, None]},
+                "mid": "café"}], {}))
+    C.append(("_wr_state_save_state", [{}], {}))
+    C.append(("_wr_state_save_state",
+              [{"neurons": {"a": {"fires": 3, "last_fired": NOW,
+                                  "promotion_streak": 0, "demotion_streak": 0}},
+                "sessions": {"s1": {"primed": {"a": 0.36}, "turns": 1,
+                                     "last_active": NOW}}}], {}))
+
+    # _derive_triggers — ties keep first occurrence; stopwords dropped;
+    # existing hits first; [:8] cap; 4-char floor; empty text.
+    C.append(("_wr_state_derive_triggers",
+              ["beta alpha beta gamma alpha delta alpha", []], {}))
+    C.append(("_wr_state_derive_triggers",
+              ["the parser and the pipeline, the parser with the memory", []], {}))
+    C.append(("_wr_state_derive_triggers",
+              ["router parity keeps parity with parity checks",
+               ["parity", "absent-trigger", ""]], {}))
+    C.append(("_wr_state_derive_triggers",
+              ["one two three four five six seven eight nine ten eleven twelve",
+               ["eleven", "twelve", "three", "absent-trigger", ""]], {}))
+    C.append(("_wr_state_derive_triggers", ["abc abcd xyzw", []], {}))
+    C.append(("_wr_state_derive_triggers", ["", []], {}))
+
+    # list_neurons — or-falsiness defaults (None/0 importance -> 0.5,
+    # None slug -> "warm"), wash ladder, unicode summary CHAR truncation,
+    # unicode body token cost (rune count / 4).
+    C.append(("_wr_state_list_neurons",
+              [[["Never pipe gate commands", "operating-gates", "gate, exit", 0.9,
+                 "gate summary", 20],
+                ["Established rule", "operating-exit", "exit", 0, "", 10],
+                ["Testing rule", None, None, None, None, 1],
+                [None, "empty-body", "", 0.5, None, 0],
+                ["café " + "é" * 100, "unicode-body", "café", 0.75,
+                 "knows " + "ü" * 90, None]]], {}))
+    C.append(("_wr_state_list_neurons", [[]], {}))
+
+    # promote_active — short universal operating rules stay resident;
+    # long operating + persona/project/human promote; dry_run counts only.
+    PROMOTE_ROWS = [
+        [3, "Check the exit code before continuing any work in the tree.", "operating"],
+        [4, "Long operating rule: " + "detail " * 20, "operating"],
+        [5, "The persona prefers coconut milk in coffee at breakfast.", "persona"],
+        [6, "Project state: the spine rebuild is mid-flight.", "project"],
+        [7, "The human drinks espresso after lunch, never before.", "human"],
+    ]
+    C.append(("_wr_state_promote_active", [PROMOTE_ROWS, False], {}))
+    C.append(("_wr_state_promote_active", [PROMOTE_ROWS, True], {}))
+    C.append(("_wr_state_promote_active", [[], False], {}))
+
+    # handle_hook — full matrix: ok route (fired > 5 truncates), text
+    # fallback, both-empty -> str(data) routes, short message skips,
+    # other events skip, None/empty data skip, route error truncates at
+    # 120 CHARS, non-string message crashes (AttributeError parity).
+    C.append(("_wr_state_handle_hook",
+              ["on_message", {"message": "check the gate parity"}, "ok",
+               ["a", "b", "c", "d", "e", "f", "g"], 9, ""], {}))
+    C.append(("_wr_state_handle_hook",
+              ["on_message", {"message": "", "text": "router parity question"}, "ok",
+               ["x"], 3, ""], {}))
+    C.append(("_wr_state_handle_hook",
+              ["on_message", {"message": "", "text": ""}, "ok", ["str-fallback"], 2, ""], {}))
+    C.append(("_wr_state_handle_hook",
+              ["on_message", {"message": "ab"}, "ok", [], 0, ""], {}))
+    C.append(("_wr_state_handle_hook",
+              ["on_startup", {"message": "check the gate parity"}, "ok", [], 0, ""], {}))
+    C.append(("_wr_state_handle_hook", ["on_message", None, "ok", [], 0, ""], {}))
+    C.append(("_wr_state_handle_hook", ["on_message", {}, "ok", [], 0, ""], {}))
+    LONG_ERR = "route exploded badly: " + "x" * 200
+    C.append(("_wr_state_handle_hook",
+              ["on_message", {"message": "check the gate parity"}, "error",
+               [], 0, LONG_ERR], {}))
+    C.append(("_wr_state_handle_hook",
+              ["on_message", {"message": 123}, "ok", [], 0, ""], {}))
+
+    return C
+
+
+SHIMS = "gen_warm_router_state_truth_shims"
+
+
+def main():
+    sys.modules[SHIMS] = sys.modules["__main__"]
+    truth = freeze(SHIMS, _cases(), oracle_module=SHIMS)
+    target = REPO / "src" / "go" / "warm-router" / "truth_state.json"
+    write_envelope(FROZEN_NOW, truth, target)
+    print(f"wrote {target} ({len(truth)} cases)")
+
+
+def verify(path):
+    raw = json.loads(Path(path).read_text())
+    frozen = raw["cases"] if isinstance(raw, dict) else raw
+    if isinstance(raw, dict) and raw.get("frozen_now"):
+        os.environ["RUNE_FROZEN_NOW"] = raw["frozen_now"]
+    sys.modules[SHIMS] = sys.modules["__main__"]
+    recased = [{"fn": c["fn"], "args": c.get("args", []), "kwargs": c.get("kwargs", {})}
+               for c in frozen]
+    fresh = freeze(SHIMS, recased, oracle_module=SHIMS)
+    if json.dumps(fresh, sort_keys=True) != json.dumps(frozen, sort_keys=True):
+        print(f"MISMATCH: regenerated truth differs from {path}")
+        for f, g in zip(frozen, fresh):
+            if json.dumps(f, sort_keys=True) != json.dumps(g, sort_keys=True):
+                print(f"  first divergence at fn={f['fn']}")
+                print(f"  frozen:   {json.dumps(f['expected'])[:300]}")
+                print(f"  oracle:   {json.dumps(g['expected'])[:300]}")
+                break
+        return 1
+    print(f"OK: {len(frozen)} cases byte-identical against live oracle")
+    return 0
+
+
+if __name__ == "__main__":
+    if len(sys.argv) > 1 and sys.argv[1] == "--verify":
+        sys.exit(verify(REPO / "src" / "go" / "warm-router" / "truth_state.json"))
+    main()

src/go/warm-router/main.go

@@ -116,6 +116,11 @@ func main() {
 	case "gap-cli":
 		// gap family parity surface (gaps_routing.go)
 		gapRoutingCLI(args)
+	case "state-scenario":
+		// DB/STATE/PERSISTENCE family parity driver (gaps_state.go):
+		// {"ops": [...]} on stdin (or a file arg) → {"results": [...]}.
+		runStateScenario(args)
+
 	case "base-level":
 		if len(args) < 2 {
 			fmt.Fprintln(os.Stderr, "usage: warm-router base-level <timestamps-csv> <now-iso>")

src/go/warm-router/truth_state_test.go

@@ -0,0 +1,168 @@
+package main
+
+// truth_state_test.go — DB/STATE/PERSISTENCE family parity gate: replays
+// truth_state.json (frozen against the live Python oracle through
+// scripts/gen_warm_router_state_truth.py wrapper shims — byte-identical
+// via its --verify) through the same runStateOp the state-scenario CLI
+// rides, so battery and live parity share ONE execution path.
+
+import (
+	"os"
+	"testing"
+)
+
+// argToOp — the frozen wrapper name → scenario-op mapping (positional
+// args → op keys).
+func argToOp(fn string, args []any) (*ODict, bool) {
+	get := func(i int) any {
+		if i < len(args) {
+			return args[i]
+		}
+		return nil
+	}
+	mk := func(name string, pairs ...OKV) *ODict {
+		op := NewODict(OKV{"op", name})
+		for _, kv := range pairs {
+			op.Set(kv.Key, kv.Value)
+		}
+		return op
+	}
+	switch fn {
+	case "_wr_state_ensure":
+		return mk("ensure_use_history_table", OKV{"twice", get(0)}), true
+	case "_wr_state_record_outcome":
+		return mk("record_outcome",
+			OKV{"session_id", get(0)}, OKV{"outcome", get(1)}, OKV{"fresh", get(2)}), true
+	case "_wr_state_record_neuron_fire":
+		return mk("record_neuron_fire",
+			OKV{"entry_id", get(0)}, OKV{"score", get(1)}, OKV{"session_id", get(2)},
+			OKV{"outcome", get(3)}, OKV{"fresh", get(4)}), true
+	case "_wr_state_db":
+		return mk("db", OKV{"store_db", get(0)}), true
+	case "_wr_state_aaak":
+		return mk("aaak", OKV{"text", get(0)}, OKV{"mode", get(1)}), true
+	case "_wr_state_write_neuron":
+		return mk("write_neuron",
+			OKV{"slug", get(0)}, OKV{"triggers", get(1)}, OKV{"body", get(2)},
+			OKV{"importance", get(3)}, OKV{"existing_id", get(4)}, OKV{"aaak", get(5)},
+			OKV{"now", get(6)}), true
+	case "_wr_state_touch":
+		return mk("touch", OKV{"state", get(0)}, OKV{"slug", get(1)}, OKV{"now", get(2)}), true
+	case "_wr_state_session_state":
+		return mk("session_state", OKV{"content", get(0)}), true
+	case "_wr_state_load_state":
+		return mk("load_state", OKV{"content", get(0)}), true
+	case "_wr_state_save_state":
+		return mk("save_state", OKV{"state", get(0)}), true
+	case "_wr_state_derive_triggers":
+		return mk("derive_triggers", OKV{"text", get(0)}, OKV{"existing", get(1)}), true
+	case "_wr_state_list_neurons":
+		return mk("list_neurons", OKV{"rows", get(0)}), true
+	case "_wr_state_promote_active":
+		return mk("promote_active", OKV{"rows", get(0)}, OKV{"dry_run", get(1)}), true
+	case "_wr_state_handle_hook":
+		return mk("handle_hook",
+			OKV{"event", get(0)}, OKV{"data", get(1)}, OKV{"route_mode", get(2)},
+			OKV{"fired", get(3)}, OKV{"total", get(4)}, OKV{"route_error", get(5)}), true
+	}
+	return nil, false
+}
+
+// TestStateFamilyParity — every frozen row through runStateOp, compared
+// as ORDER-SENSITIVE py-rendered JSON (key order and number shape are
+// part of the contract).
+func TestStateFamilyParity(t *testing.T) {
+	raw, err := os.ReadFile("truth_state.json")
+	if err != nil {
+		t.Fatalf("truth_state.json missing: %v", err)
+	}
+	decoded, err := oDecode(raw)
+	if err != nil {
+		t.Fatalf("unparseable: %v", err)
+	}
+	doc, _ := decoded.(*ODict)
+	if doc == nil {
+		t.Fatalf("truth_state.json: top level must be an object")
+	}
+	casesAny, _ := doc.Get("cases")
+	cases, _ := casesAny.([]any)
+	if len(cases) == 0 {
+		t.Fatalf("no cases")
+	}
+	fails := 0
+	for i, rawCase := range cases {
+		cs, _ := rawCase.(*ODict)
+		if cs == nil {
+			t.Errorf("case %d: not an object", i)
+			fails++
+			continue
+		}
+		fn, _ := cs.Get("fn")
+		fnStr, _ := fn.(string)
+		argsAny, _ := cs.Get("args")
+		args, _ := argsAny.([]any)
+		op, ok := argToOp(fnStr, args)
+		if !ok {
+			t.Errorf("case %d: no op mapping for %s", i, fnStr)
+			fails++
+			continue
+		}
+		got, err := runStateOp(op)
+		if err != nil {
+			t.Errorf("case %d %s: ERROR %v", i, fnStr, err)
+			fails++
+			continue
+		}
+		want, _ := cs.Get("expected")
+		gotRendered := pyRenderCompact(got)
+		wantRendered := pyRenderCompact(want)
+		if gotRendered != wantRendered {
+			t.Errorf("case %d %s DIVERGE:\n got %s\nwant %s", i, fnStr, gotRendered, wantRendered)
+			fails++
+		}
+	}
+	if fails > 0 {
+		t.Fatalf("PARITY FAILED: %d/%d divergences", fails, len(cases))
+	}
+	t.Logf("PARITY OK: %d state-family cases exact-equal vs the Python oracle", len(cases))
+}
+
+// TestUseHistoryFlagLatch — the created-flag latch: the second
+// EnsureUseHistoryTable issues NOTHING (the oracle's module-global
+// optimization, frozen in the twice-row).
+func TestUseHistoryFlagLatch(t *testing.T) {
+	ResetUseHistoryTableFlag()
+	defer ResetUseHistoryTableFlag()
+	events := []any{}
+	exec := recExec{&events}
+	if err := EnsureUseHistoryTable(exec); err != nil {
+		t.Fatalf("first ensure: %v", err)
+	}
+	if err := EnsureUseHistoryTable(exec); err != nil {
+		t.Fatalf("second ensure: %v", err)
+	}
+	if len(events) != 4 { // 3 DDL + 1 commit — nothing from the latched call
+		t.Errorf("latched call must be silent: %d events", len(events))
+	}
+}
+
+// TestTruncateRunes — summary[:80] and error[:120] slice CHARACTERS
+// (str[:n] quirk law), never bytes.
+func TestTruncateRunes(t *testing.T) {
+	sum := repeatRunes("é", 100)
+	got := truncateRunesW(sum, 80)
+	if want := repeatRunes("é", 80); got != want {
+		t.Errorf("truncateRunesW: got %d runes want 80", len([]rune(got)))
+	}
+	if truncateRunesW("short", 80) != "short" {
+		t.Error("under-boundary string must pass through")
+	}
+}
+
+func repeatRunes(s string, n int) string {
+	out := ""
+	for i := 0; i < n; i++ {
+		out += s
+	}
+	return out
+}

verification/wrs-freeze-regen.exit

@@ -0,0 +1 @@
+0

verification/wrs-freeze-verify.exit

@@ -0,0 +1 @@
+0

verification/wrs-gobuild.exit

@@ -0,0 +1 @@
+0

verification/wrs-gofmt.exit

@@ -0,0 +1 @@
+0

verification/wrs-gotest.exit

@@ -0,0 +1 @@
+0

verification/wrs-govet.exit

@@ -0,0 +1 @@
+0

verification/wrs-parity.exit

@@ -0,0 +1 @@
+0

verification/wrs-pytest-live.exit

@@ -0,0 +1 @@
+0

verification/wrs-pytest-pinned-control-main.exit

@@ -0,0 +1 @@
+1

verification/wrs-pytest-pinned.exit

@@ -0,0 +1 @@
+1

verification/wrs-summary.md

@@ -0,0 +1,67 @@
+# warm_router DB/STATE/PERSISTENCE family gap-close — verification summary
+
+Module: warm_router (GO side, dichotomy wave). Contract builder:
+`factory-heavy-wr-state`. Worktree: /mnt/workdrive/rune-wave-wr.
+
+## Scope (family lane)
+
+Ported into src/go/warm-router/gaps_state.go (new family file):
+_ensure_use_history_table, record_outcome, record_neuron_fire, _db,
+_aaak, write_neuron, _touch, _session_state (ensureSessionState),
+load_state, save_state, _derive_triggers, list_neurons, promote_active,
+handle_hook. main carries the `state-scenario` parity subcommand (one
+switch case). SQLite stays facade-side per the landed crate seams: the
+verbatim SQL text, parameter tuples and decisions are the Go contract;
+connections are injected SQLExecer seams.
+
+## Audit delta
+
+dichotomy_audit.py: warm_router 17/33 -> 28/33 fns matched
+(remaining 5 are sibling families: base_level_activation,
+_associated_neuron_bodies, graph_context, _primed_associations,
+render_fired). Repo: FULL 127 -> 128 modules, fn-level 1729 -> 1740.
+
+## Gates (all evidence in verification/, exit codes beside logs)
+
+| gate | log | exit |
+|---|---|---|
+| go test -count=1 ./... (warm-router, incl. 46-case battery) | wrs-gotest.log | 0 |
+| go build ./... | wrs-gobuild.log | 0 |
+| gofmt -l (clean) | wrs-gofmt.log | 0 (empty) |
+| go vet ./... | wrs-govet.log | 0 |
+| freeze --verify (byte-identical vs live oracle) | wrs-freeze-verify.log | 0 |
+| fresh regen (byte-stable, sha 9e47e487…) | wrs-freeze-regen.log | 0 |
+| live parity go binary vs python oracle | wrs-parity.log | 0 (46/46) |
+| pytest module suite, delegation live (MS_RUST_LIB -> main .so) | wrs-pytest-live.log | 0 (5 passed) |
+| pytest module suite, pinned python | wrs-pytest-pinned.log | 1 (2 failed) |
+
+## Divergences (documented, not bugs)
+
+- Pinned-python pytest (RUNE_MEMORY_CORE=python) fails the 2
+  kernel-dependent tests — IDENTICALLY in the main checkout
+  (control: wrs-pytest-pinned-control-main.log, same 2 failed / 3
+  passed). memory_scoring's Python fallback is RETIRED (strangler
+  stage 3 — the stub raises; the outer catch degrades to -2.0), so the
+  suite is delegation-dependent by design. My diff contains zero
+  Python source changes.
+- The battery/parity compare the executed statement sequence +
+  parameter tuples + commit points, not connection lifecycles (the
+  oracle opens one connection per call; the Go seam reuses the injected
+  execer — facade adapter's choice, statements are the contract).
+- handle_hook non-string-message crash: the oracle raises an uncaught
+  AttributeError; the Go port returns the {"__crash__": "AttributeError"}
+  marker (crash-class parity; frozen row pins the class).
+- neurons.go's landed BodyKnowledge truncates summaries by BYTES
+  (s[:80]); ListNeurons uses a rune-safe copy per the str[:n] quirk law.
+  The two implementations coexist; bodies with non-ASCII summaries
+  diverge in the landed helper — frozen truth pins the rune-safe shape.
+
+## Frozen truth
+
+src/go/warm-router/truth_state.json — 46 cases, envelope
+frozen_now=2026-10-14T21:30:00+00:00, generated + --verified by
+scripts/gen_warm_router_state_truth.py (wrapper-shim law: recording
+sqlite3.connect, patched wr._db/STATE, pinned clock from row args).
+No absolute paths (STATE_STORE_DB placeholder), no wall-clock taint
+(`_touch` binds datetime at module level — shim pins wr.datetime
+directly; write_neuron's function-local import is reached by the rig).