444 lines
17 KiB
Python
444 lines
17 KiB
Python
#!/usr/bin/env python3
|
|
"""gen_experience_encoding_truth.py — freeze/replay parity truth for
|
|
experience_encoding (dichotomy wave, #2479 tranche).
|
|
|
|
Freeze mode (default): exercises the LIVE Python oracle
|
|
(src/rune/experience_encoding.py evaluate — the module's single public
|
|
function) across every branch and records (kernel_input,
|
|
expected_kernel_output) pairs into
|
|
src/rust/experience_encoding/src/frozen_ee_truth.json.
|
|
|
|
Replay mode (--replay): re-runs the SAME scenarios with the Rust core
|
|
delegation ACTIVE (RUNE_EE_CORE unpinned) and asserts both kernel-level
|
|
(loader direct) and facade-level (oracle evaluate delegates) parity
|
|
against the frozen file.
|
|
|
|
Poisoned-truth law: freeze pins RUNE_EE_CORE=python BEFORE any import.
|
|
Mutation law: case inputs are deep-copied BEFORE they reach the oracle.
|
|
Lexicon-seam law: the oracle's `import lexicon_evolution` is an
|
|
ENVIRONMENT seam — the generator pins it per case via
|
|
`lexicon_available`; freeze/replay with the flag False block the flat
|
|
import (sys.modules sentinel) so the ImportError fallback branch
|
|
(conservative evidence admission) is frozen like any other. The Rust
|
|
kernel receives the same flag in its payload and mirrors the branch.
|
|
"""
|
|
import copy
|
|
import json
|
|
import os
|
|
import sys
|
|
from datetime import datetime, timezone
|
|
|
|
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
|
TRUTH_PATH = os.path.join(REPO, "src", "rust", "experience_encoding",
|
|
"src", "frozen_ee_truth.json")
|
|
REPLAY = "--replay" in sys.argv
|
|
|
|
if not REPLAY:
|
|
os.environ["RUNE_EE_CORE"] = "python" # poisoned-truth law (freeze side)
|
|
else:
|
|
os.environ.pop("RUNE_EE_CORE", None) # live-parity law (facade must delegate)
|
|
|
|
sys.path.insert(0, os.path.join(REPO, "src"))
|
|
sys.path.insert(0, os.path.join(REPO, "src", "rune"))
|
|
|
|
import experience_encoding as oracle_mod # noqa: E402
|
|
|
|
if REPLAY:
|
|
from rune import ee_loader # noqa: E402 (kernel-level replay side)
|
|
if not ee_loader.available():
|
|
print("REPLAY ABORT: kernel not built —", ee_loader.LIB_PATH)
|
|
sys.exit(2)
|
|
|
|
|
|
class _Unset:
|
|
pass
|
|
|
|
|
|
_UNSET = _Unset()
|
|
|
|
|
|
def _input(event, existing_episodes=_UNSET, corpus_texts=_UNSET,
|
|
lexicon_available=True):
|
|
payload = {"event": event, "lexicon_available": lexicon_available}
|
|
if existing_episodes is not _UNSET:
|
|
payload["existing_episodes"] = list(existing_episodes)
|
|
if corpus_texts is not _UNSET:
|
|
payload["corpus_texts"] = list(corpus_texts)
|
|
return payload
|
|
|
|
|
|
class _LexiconBlocked:
|
|
"""Context manager: blocks the flat `import lexicon_evolution` the way
|
|
a missing module would (sys.modules None sentinel → ImportError)."""
|
|
|
|
def __enter__(self):
|
|
self._saved = sys.modules.get("lexicon_evolution", _UNSET)
|
|
sys.modules["lexicon_evolution"] = None
|
|
return self
|
|
|
|
def __exit__(self, *exc):
|
|
if self._saved is _UNSET:
|
|
sys.modules.pop("lexicon_evolution", None)
|
|
else:
|
|
sys.modules["lexicon_evolution"] = self._saved
|
|
return False
|
|
|
|
|
|
def facade_call(inp):
|
|
"""Run the LIVE facade (delegates to the Rust core when built)."""
|
|
if inp.get("lexicon_available", True):
|
|
return oracle_mod.evaluate(
|
|
copy.deepcopy(inp["event"]),
|
|
copy.deepcopy(inp.get("existing_episodes")),
|
|
copy.deepcopy(inp.get("corpus_texts", ())),
|
|
)
|
|
with _LexiconBlocked():
|
|
return oracle_mod.evaluate(
|
|
copy.deepcopy(inp["event"]),
|
|
copy.deepcopy(inp.get("existing_episodes")),
|
|
copy.deepcopy(inp.get("corpus_texts", ())),
|
|
)
|
|
|
|
|
|
def kernel_call(inp):
|
|
"""Run the Rust kernel directly through the ctypes loader."""
|
|
return ee_loader.evaluate(**copy.deepcopy(inp))
|
|
|
|
|
|
# ── evaluate ─────────────────────────────────────────────────────────────────
|
|
|
|
def ev_case(rec, name, event, existing=_UNSET, corpus=_UNSET,
|
|
lexicon_available=True):
|
|
inp = _input(event, existing, corpus, lexicon_available)
|
|
got = facade_call(copy.deepcopy(inp))
|
|
rec(name, inp, got, got)
|
|
|
|
|
|
def c_ev_missing_text(rec):
|
|
ev_case(rec, "ev-missing-text-key-empty", {"source": "user"})
|
|
|
|
|
|
def c_ev_empty_text(rec):
|
|
ev_case(rec, "ev-empty-text", {"text": "", "source": "user"})
|
|
|
|
|
|
def c_ev_null_text(rec):
|
|
ev_case(rec, "ev-null-text-coerced", {"text": None, "source": "user"})
|
|
|
|
|
|
def c_ev_whitespace_text(rec):
|
|
ev_case(rec, "ev-whitespace-only-text", {"text": " \n\t ", "source": "user"})
|
|
|
|
|
|
def c_ev_short_14(rec):
|
|
ev_case(rec, "ev-short-14-chars", {"text": "x" * 14, "source": "user"})
|
|
|
|
|
|
def c_ev_boundary_15(rec):
|
|
# exactly 15 chars passes the gate; no episodes/corpus → ungrounded
|
|
ev_case(rec, "ev-boundary-15-chars-evidence",
|
|
{"text": "x" * 15, "source": "curator", "session_id": "s1"})
|
|
|
|
|
|
def c_ev_null_source_session(rec):
|
|
# get() default only applies when the KEY is missing — a present-but-null
|
|
# source/session_id is pinned as provenance nulls (oracle quirk)
|
|
ev_case(rec, "ev-null-source-session-provenance",
|
|
{"text": "hi", "source": None, "session_id": None})
|
|
|
|
|
|
def c_ev_empty_source_session(rec):
|
|
ev_case(rec, "ev-empty-string-source-session",
|
|
{"text": "x" * 20, "source": "", "session_id": ""})
|
|
|
|
|
|
def c_ev_extra_keys(rec):
|
|
ev_case(rec, "ev-extra-keys-ignored",
|
|
{"text": "x" * 16, "kind": "note", "weight": 3})
|
|
|
|
|
|
def c_ev_duplicate_normalized(rec):
|
|
ev_case(rec, "ev-duplicate-normalized-whitespace-case",
|
|
{"text": "Nick prefers DARK coffee", "source": "user"},
|
|
existing=[{"text": "nick prefers DARK coffee"}])
|
|
|
|
|
|
def c_ev_duplicate_exact(rec):
|
|
ev_case(rec, "ev-duplicate-exact-string",
|
|
{"text": "she drinks oatmilk daily", "source": "user"},
|
|
existing=[{"text": "she drinks oatmilk daily"}])
|
|
|
|
|
|
def c_ev_duplicate_episode_case(rec):
|
|
ev_case(rec, "ev-duplicate-episode-uppercase",
|
|
{"text": "nick prefers dark coffee", "source": "user"},
|
|
existing=[{"text": "NICK PREFERS DARK COFFEE"}])
|
|
|
|
|
|
def c_ev_duplicate_episode_padding(rec):
|
|
ev_case(rec, "ev-duplicate-episode-padded",
|
|
{"text": "nick prefers dark coffee", "source": "user"},
|
|
existing=[{"text": " nick prefers dark coffee "}])
|
|
|
|
|
|
def c_ev_not_dup_punctuation(rec):
|
|
# the exclamation mark survives normalization → NOT a duplicate; the
|
|
# event then grounds in the episode's corpus and shares terms
|
|
ev_case(rec, "ev-punctuation-survives-not-duplicate",
|
|
{"text": "NICK PREFERS dark coffee!", "source": "user"},
|
|
existing=[{"text": "nick prefers dark coffee"}])
|
|
|
|
|
|
def c_ev_dup_null_episode_text(rec):
|
|
# episodes with None/missing text never duplicate; corpus fallback is
|
|
# all empty → ungrounded evidence
|
|
ev_case(rec, "ev-null-episode-texts-ungrounded",
|
|
{"text": "quantum flux patterns emerge here", "source": "user"},
|
|
existing=[{"text": None}, {}])
|
|
|
|
|
|
def c_ev_lexicon_blocked(rec):
|
|
# ImportError seam: long text, lexicon unavailable → conservative evidence
|
|
ev_case(rec, "ev-lexicon-blocked-conservative-evidence",
|
|
{"text": "she drinks oatmilk every single morning",
|
|
"source": "user"},
|
|
lexicon_available=False)
|
|
|
|
|
|
def c_ev_lexicon_blocked_short(rec):
|
|
# the empty/short gate runs BEFORE the lexicon import — still IGNORE
|
|
ev_case(rec, "ev-lexicon-blocked-short-still-ignored",
|
|
{"text": "hi", "source": "user"},
|
|
lexicon_available=False)
|
|
|
|
|
|
def c_ev_lexicon_blocked_with_episodes(rec):
|
|
# the duplicate check is INSIDE the try — blocked lexicon wins over a
|
|
# would-be duplicate (conservative evidence, not IGNORE)
|
|
ev_case(rec, "ev-lexicon-blocked-duplicate-path-conservative",
|
|
{"text": "nick prefers dark coffee", "source": "user"},
|
|
existing=[{"text": "nick prefers dark coffee"}],
|
|
lexicon_available=False)
|
|
|
|
|
|
def c_ev_framing_only(rec):
|
|
ev_case(rec, "ev-framing-only-no-lexical-signal",
|
|
{"text": "work thing way always", "source": "user"})
|
|
|
|
|
|
def c_ev_episode_shared_single(rec):
|
|
ev_case(rec, "ev-episode-shared-single-term",
|
|
{"text": "she never jokes about oatmilk at all", "source": "user"},
|
|
existing=[{"text": "i love oatmilk a lot"}],
|
|
corpus=["oatmilk is life", "tea later day"])
|
|
|
|
|
|
def c_ev_episode_shared_multi(rec):
|
|
ev_case(rec, "ev-episode-shared-multiple-sorted",
|
|
{"text": "oatmilk and celesta every morning", "source": "user"},
|
|
existing=[{"text": "oatmilk and celesta"}],
|
|
corpus=["oatmilk bar", "celesta hall"])
|
|
|
|
|
|
def c_ev_episode_shared_top3(rec):
|
|
ev_case(rec, "ev-episode-shared-sorted-top3",
|
|
{"text": "oatmilk celesta zephyr quartz ritual", "source": "user"},
|
|
existing=[{"text": "oatmilk celesta zephyr quartz"}],
|
|
corpus=["oatmilk celesta zephyr quartz notes"])
|
|
|
|
|
|
def c_ev_episode_novel_grounded(rec):
|
|
ev_case(rec, "ev-episode-novel-axis-grounded",
|
|
{"text": "celesta practice daily ritual", "source": "session"},
|
|
existing=[{"text": "i love oatmilk"}],
|
|
corpus=["she plays the celesta nightly"])
|
|
|
|
|
|
def c_ev_evidence_ungrounded(rec):
|
|
ev_case(rec, "ev-evidence-fact-shaped-ungrounded",
|
|
{"text": "quantum flux patterns emerge", "source": "curator"},
|
|
existing=[{"text": "i love oatmilk"}],
|
|
corpus=["oatmilk is life"])
|
|
|
|
|
|
def c_ev_evidence_empty_everything(rec):
|
|
ev_case(rec, "ev-evidence-no-episodes-no-corpus",
|
|
{"text": "quantum flux patterns emerge", "source": "curator"})
|
|
|
|
|
|
def c_ev_stem_default_corpus(rec):
|
|
# corpus omitted → corpus = episodes' texts; stem-aware grounding
|
|
# (migrated→migr matches migration→migr)
|
|
ev_case(rec, "ev-episode-stem-grounded-default-corpus",
|
|
{"text": "she migrated north yesterday", "source": "session"},
|
|
existing=[{"text": "the migration of songbirds"}])
|
|
|
|
|
|
def c_ev_corpus_override_shared_killed(rec):
|
|
# corpus override pushes oatmilk df over the threshold → share killed,
|
|
# but grounding survives (plain stem overlap) → novel-axis episode
|
|
ev_case(rec, "ev-corpus-override-share-killed-novel-grounded",
|
|
{"text": "she never jokes about oatmilk ok", "source": "user"},
|
|
existing=[{"text": "i love oatmilk"}],
|
|
corpus=["oatmilk"] * 30)
|
|
|
|
|
|
def c_ev_no_episodes_with_corpus(rec):
|
|
# shared_distinctive runs against the (empty) episodes while grounding
|
|
# uses the corpus → novel-axis episode
|
|
ev_case(rec, "ev-episode-no-episodes-corpus-grounded",
|
|
{"text": "she drinks oatmilk daily", "source": "user"},
|
|
corpus=["oatmilk is life"])
|
|
|
|
|
|
def c_ev_episodes_without_corpus(rec):
|
|
# corpus defaults to the episodes' texts
|
|
ev_case(rec, "ev-episode-default-corpus-is-episodes",
|
|
{"text": "she drinks oatmilk daily", "source": "user"},
|
|
existing=[{"text": "oatmilk rituals fill the morning"}])
|
|
|
|
|
|
# ── driver ───────────────────────────────────────────────────────────────────
|
|
|
|
EVALUATE = [
|
|
c_ev_missing_text, c_ev_empty_text, c_ev_null_text, c_ev_whitespace_text,
|
|
c_ev_short_14, c_ev_boundary_15, c_ev_null_source_session,
|
|
c_ev_empty_source_session, c_ev_extra_keys, c_ev_duplicate_normalized,
|
|
c_ev_duplicate_exact, c_ev_duplicate_episode_case,
|
|
c_ev_duplicate_episode_padding, c_ev_not_dup_punctuation,
|
|
c_ev_dup_null_episode_text, c_ev_lexicon_blocked,
|
|
c_ev_lexicon_blocked_short, c_ev_lexicon_blocked_with_episodes,
|
|
c_ev_framing_only, c_ev_episode_shared_single, c_ev_episode_shared_multi,
|
|
c_ev_episode_shared_top3, c_ev_episode_novel_grounded,
|
|
c_ev_evidence_ungrounded, c_ev_evidence_empty_everything,
|
|
c_ev_stem_default_corpus, c_ev_corpus_override_shared_killed,
|
|
c_ev_no_episodes_with_corpus, c_ev_episodes_without_corpus,
|
|
]
|
|
|
|
GROUPS = {
|
|
"evaluate": (EVALUATE, True, "ee_evaluate", lambda w: w),
|
|
}
|
|
|
|
|
|
# ── _ee_core (the loader-decision residue) ────────────────────────────────────
|
|
|
|
def _probe_once(rec, name, env):
|
|
"""Freeze the oracle's _ee_core() decision under one env state.
|
|
|
|
The env flip + restore brackets the probe so the poisoned-truth pin
|
|
(python) holds for every facade case captured around it. `env=None`
|
|
probes the unset state (os.environ.get default 'auto'). The auto/unset
|
|
rows record the ee_loader module name when the .so is built — the
|
|
replay env guarantees it is."""
|
|
old = os.environ.get("RUNE_EE_CORE")
|
|
if env is None:
|
|
os.environ.pop("RUNE_EE_CORE", None)
|
|
else:
|
|
os.environ["RUNE_EE_CORE"] = env
|
|
try:
|
|
core = oracle_mod._ee_core()
|
|
expected = {"core": None if core is None
|
|
else getattr(core, "__name__", str(core))}
|
|
rec(name, {"env": env}, dict(expected), expected)
|
|
finally:
|
|
if old is None:
|
|
os.environ.pop("RUNE_EE_CORE", None)
|
|
else:
|
|
os.environ["RUNE_EE_CORE"] = old
|
|
|
|
|
|
def p_python_pin(rec):
|
|
_probe_once(rec, "probe-python-pin", "python")
|
|
|
|
|
|
def p_auto(rec):
|
|
_probe_once(rec, "probe-auto", "auto")
|
|
|
|
|
|
def p_unset(rec):
|
|
_probe_once(rec, "probe-unset", None)
|
|
|
|
|
|
GROUPS["core_probe"] = ([p_python_pin, p_auto, p_unset], True, None, None)
|
|
|
|
|
|
def canon(x) -> str:
|
|
"""Content canonicalization (key-order independent)."""
|
|
return json.dumps(json.loads(json.dumps(x, default=str)), sort_keys=True)
|
|
|
|
|
|
def main():
|
|
frozen = None
|
|
if REPLAY:
|
|
with open(TRUTH_PATH) as f:
|
|
frozen = json.load(f)["cases"]
|
|
|
|
groups = {}
|
|
for group, (cases, facade_replayable, _sym, _unwrap) in GROUPS.items():
|
|
recs = []
|
|
|
|
def rec(name, inp, oracle_ret, expected, _r=recs, _g=group):
|
|
entry = {"name": name, "input": inp,
|
|
"return": oracle_ret, "expected": expected}
|
|
if REPLAY and facade_replayable:
|
|
old = next(c for c in frozen[_g] if c["name"] == name)
|
|
for key in ("input", "return", "expected"):
|
|
got, want = entry[key], old[key]
|
|
if canon(got) != canon(want):
|
|
print(f"FACADE DIVERGE [{_g}] {name} {key}")
|
|
print(" got: ", json.dumps(got, sort_keys=True, default=str)[:600])
|
|
print(" want:", json.dumps(want, sort_keys=True, default=str)[:600])
|
|
_r.append({"name": name, "diverged": key})
|
|
return
|
|
_r.append(entry)
|
|
|
|
for c in cases:
|
|
c(rec)
|
|
groups[group] = recs
|
|
|
|
total = sum(len(v) for v in groups.values())
|
|
if REPLAY:
|
|
facade_groups = [g for g, (_, r, _, _) in GROUPS.items() if r]
|
|
facade_total = sum(len(groups[g]) for g in facade_groups)
|
|
facade_failures = sum(1 for g in facade_groups for c in groups[g]
|
|
if "diverged" in c)
|
|
|
|
failures = 0
|
|
checked = 0
|
|
for group, cases in frozen.items():
|
|
if group not in GROUPS or GROUPS[group][2] is None:
|
|
continue # core_probe: metadata group, facade-replayed above
|
|
for case in cases:
|
|
checked += 1
|
|
got = kernel_call(case["input"])
|
|
want = case["expected"]
|
|
if canon(got) != canon(want):
|
|
failures += 1
|
|
print(f"KERNEL DIVERGE [{group}] {case['name']}")
|
|
print(" got: ", json.dumps(got, sort_keys=True, default=str)[:600])
|
|
print(" want:", json.dumps(want, sort_keys=True, default=str)[:600])
|
|
print(f"replay: facade {facade_total - facade_failures}/{facade_total}, "
|
|
f"kernel {checked - failures}/{checked}")
|
|
return 1 if (failures or facade_failures) else 0
|
|
|
|
doc = {
|
|
"frozen": datetime.now(timezone.utc).isoformat(),
|
|
"oracle": "src/rune/experience_encoding.py",
|
|
"pin": "RUNE_EE_CORE=python",
|
|
"note": "Frozen via scripts/gen_experience_encoding_truth.py against "
|
|
"the live Python oracle. Battery embeds this VERBATIM. The "
|
|
"`lexicon_available` flag pins the oracle's flat-import "
|
|
"environment seam (ImportError fallback branch).",
|
|
"cases": groups,
|
|
}
|
|
os.makedirs(os.path.dirname(TRUTH_PATH), exist_ok=True)
|
|
with open(TRUTH_PATH, "w") as f:
|
|
json.dump(doc, f, indent=1)
|
|
f.write("\n")
|
|
print(f"frozen {total} cases → {os.path.relpath(TRUTH_PATH, REPO)}")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|