Nick
056fa974a8
Name-preserving port of the 7 audit-missing functions into
src/rust/voice_sync (the module's classification side is RUST — sqlite3
storage-backed; a stale Go voice-sync dir exists from an early partial
register-substring helper):
get_delivery_register_from_db / get_delivery_register
-> the chunk row-selection loop (rows arrive fetched Python-side;
the SQL WHERE + fresh-store OperationalError guard stay in the
facade), len>200 pinned as a CODEPOINT count
extract_voice_section
-> section parser: influence capture incl. '- **:**',
TECHNICAL+DELIVERY double-hit header captured as a technical
line, elif-section persistence, case-sensitive THE VOICE /
THE FEMININE prefixes, truthy all-empty-lists parse
generate_voice_section
-> assembler with the exact fixed block + calibration cap-3
update_rune_md_content / update_rune_md
-> pure splice view (composite {"result", "new_content"} so the
facade performs the write) — the oracle gains the same
oracle-level JSON view the shared freeze gate pins
(update_rune_md_content; salience_gate evaluate_salience_dict
precedent); char-boundary-safe +10 offset past ## Voice
check_sync / handle_hook / sync_voice
-> parts-shaped decisions (register, prompt_content in as
arguments); handle_hook pins the oracle's auto-sync 'synced:
true with error details' verbatim
_vs_core
-> kernel mirror of the facade probe contract (pin lattice,
salience_gate _sg_core precedent)
Facade delegation: voice_sync.py wires vs_loader.py (10 vs_* doc-envelope
symbols, free via capability_scoring cs_free_string, no free-symbol
export, VOICE_SYNC_RUST_LIB override, RUNE_VOICE_SYNC_CORE=python pin —
the shared harness derives EXACTLY this pin name, inverting salience's
wrong-name quirk); _py_* fallbacks are the strangler-fig machinery class;
SQLite I/O, rune.md writes, wall-clock _now and the argparse main() CLI
stay facade-side (behavior crate precedent).
Frozen truth via the shared scripts/freeze_truth.py harness (generator
scripts/gen_voice_sync_truth.py, poisoned-truth law
RUNE_VOICE_SYNC_CORE=python):
frozen_vs_truth.verify.json — 37 pure rows byte-identical under
freeze_truth.py --verify (plain process, core-state independent);
frozen_vs_truth.json — 64-row envelope: verify rows first, then the
rig-required store/file rows whose kernel-shaped inputs ride in args
with fixtures for --replay (len<=200/201 boundary, first-match order,
fresh-store None, lowercase delivery-register passes-SQL-but-loses-
python-filter, missing-influence scan order, auto-sync write capture).
branch_battery.rs replays both VERBATIM (include_str!, order-sensitive
canonical-string compare — the facade consumes kernel JSON directly so
Python insertion order must survive: serde_json preserve_order).
Live parity 125/125 (gen --replay: facade-delegating + kernel-direct);
canary 15/15; second-modality differential 486 comparisons / 0
divergences (2479-seed fuzz beyond the corpus incl. multibyte splices
and sandboxed I/O decisions); pytest 17/17 in three pin states (live /
RUNE_VOICE_SYNC_CORE=python / VOICE_SYNC_RUST_LIB poisoned).
Dichotomy audit: voice_sync PARTIAL 3/10 -> FULL 12/12 (update_rune_md_
content + _vs_core join the surface per the salience precedent; _py_*
helpers are the audit's machinery class).
788 lines
31 KiB
Python
788 lines
31 KiB
Python
#!/usr/bin/env python3
|
|
"""gen_voice_sync_truth.py — freeze/replay parity truth for voice_sync
|
|
(dichotomy wave #2479 gap-close, shared harness scripts/freeze_truth.py).
|
|
|
|
Freeze mode (default): exercises the LIVE Python oracle and writes two
|
|
files:
|
|
* src/rust/voice_sync/src/frozen_vs_truth.verify.json — shared-verify
|
|
rows for the PURE fn surface only (extract_voice_section,
|
|
generate_voice_section, update_rune_md_content): runnable in a plain
|
|
harness process, byte-identical under `freeze_truth.py --verify ...
|
|
--module rune.voice_sync` regardless of the crate's build state. NOTE
|
|
the salience_gate quirk INVERTS here: the harness derives
|
|
RUNE_VOICE_SYNC_CORE and the loader honors exactly that name, so the
|
|
shared gate always replays the pinned Python floor; kernel parity on
|
|
these rows is gated by the battery + replay instead.
|
|
* src/rust/voice_sync/src/frozen_vs_truth.json — envelope carrying ALL
|
|
rows (verify rows first, then the rig-required rows): store-backed
|
|
get_delivery_register/get_delivery_register_from_db (fixture chunks),
|
|
file-I/O update_rune_md, and the parts-shaped check_sync, sync_voice,
|
|
handle_hook rows whose kernel inputs (register text, prompt content,
|
|
post-SQL row lists) ride in "args". Each I/O row carries the fixture
|
|
("fixture" store chunks / "fixture_md" prompt body / "fresh_store")
|
|
so --replay can rebuild the oracle environment deterministically.
|
|
|
|
Poisoned-truth law: freeze pins RUNE_VOICE_SYNC_CORE=python (the shared
|
|
harness) — the frozen truth is the pure-Python oracle floor, never a
|
|
capture of the Rust kernel.
|
|
|
|
Replay mode (--replay): unpins the core, requires the built cdylib
|
|
(VOICE_SYNC_RUST_LIB untouched — STALE-CDYLIB LAW: cargo build --release
|
|
before replaying), then asserts, per row:
|
|
* facade-level parity — the oracle functions delegating live, with the
|
|
row's fixtures rebuilt under a scratch HOME (fresh store no-chunks-
|
|
table edge → OperationalError → None included),
|
|
* kernel-level parity — vs_loader direct against the row's kernel args.
|
|
PLUGIN is compared descriptor-vs-descriptor (frozen data).
|
|
|
|
Surface covered (fn coverage law — every public fn and branch):
|
|
get_delivery_register first-match (query order), len<=200 skip
|
|
(exact boundary 200/201 chars), no-keyword
|
|
noise, lowercase "delivery register" passes
|
|
SQL-LIKE but fails the case-sensitive Python
|
|
filter, fresh store (no chunks table →
|
|
OperationalError → None), from_db-direct
|
|
extract_voice_section full register, empty string → None,
|
|
whitespace-only → the four empty lists (a
|
|
TRUTHY parse), influence capture + colon
|
|
rule ("- **:**" included), section
|
|
persistence, TECHNICAL+DELIVERY double-hit
|
|
header captured as a technical line, TECH-
|
|
NICAL-without-DELIVERY no-branch, CALIBRA-
|
|
TION / WHEN THE REGISTER, HOW IT BLENDS,
|
|
THE VOICE / THE FEMININE case-sensitive
|
|
prefixes, lowercase "the voice" NOT a
|
|
trigger, CRLF line ends, unicode content
|
|
generate_voice_section None/empty-dict falsy, all-empty-lists
|
|
dict truthy → fixed block, influence/voice/
|
|
technical/calibration parts, cap-3
|
|
calibration, full round-trip
|
|
update_rune_md_content missing file, no ## Voice (case-sensitive),
|
|
no following section, voice-at-end +10-chars
|
|
edge, multibyte before/after the header,
|
|
unicode section (codepoint length), happy
|
|
path with sections before/after
|
|
update_rune_md real-file rows under the rig (written body
|
|
captured as new_content), missing file,
|
|
no-Voice error (file unchanged)
|
|
check_sync synced, missing one/many influences (order),
|
|
substring containment, colon-split names,
|
|
no-bold influences skipped, no register
|
|
(None AND empty-string inputs), no rune.md,
|
|
empty prompt
|
|
sync_voice update, no register, no rune.md file,
|
|
no-## Voice, no-following-section
|
|
handle_hook synced echo, auto-sync (writes), auto-sync
|
|
with no register (details error, synced
|
|
still TRUE — oracle pins the lie), non-
|
|
on_session_end string/dict events, data
|
|
accepted and ignored
|
|
PLUGIN descriptor parity (frozen data)
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import copy
|
|
import datetime
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
import sys
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
CRATE = REPO / "src" / "rust" / "voice_sync"
|
|
VERIFY_PATH = CRATE / "src" / "frozen_vs_truth.verify.json"
|
|
ENVELOPE_PATH = CRATE / "src" / "frozen_vs_truth.json"
|
|
REPLAY = "--replay" in sys.argv
|
|
|
|
sys.path.insert(0, str(REPO / "scripts"))
|
|
sys.path.insert(0, str(REPO / "src"))
|
|
|
|
if REPLAY:
|
|
os.environ.pop("RUNE_VOICE_SYNC_CORE", None) # delegation live for replay
|
|
else:
|
|
# POISONED-TRUTH LAW (freeze): the oracle floor, never the kernel.
|
|
os.environ["RUNE_VOICE_SYNC_CORE"] = "python"
|
|
|
|
import rune.voice_sync as voice_sync # noqa: E402
|
|
import rune.vs_loader as vs_loader # noqa: E402
|
|
from freeze_truth import freeze # noqa: E402
|
|
|
|
# ── shared fixtures ─────────────────────────────────────────────────────
|
|
|
|
CHUNKS_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS chunks (
|
|
chunk_id TEXT PRIMARY KEY,
|
|
doc_id TEXT NOT NULL,
|
|
text TEXT NOT NULL,
|
|
content_hash TEXT NOT NULL,
|
|
importance REAL DEFAULT 0.5,
|
|
ingested_at TEXT NOT NULL
|
|
);
|
|
"""
|
|
|
|
SQL = ("SELECT text FROM chunks WHERE text LIKE '%DELIVERY REGISTER%' "
|
|
"AND text LIKE '%RUNE%VOICE%'")
|
|
|
|
PAD = "-" * 180
|
|
|
|
|
|
def reg_full(marker: str) -> str:
|
|
"""A register chunk that passes the SQL LIKE and the Python filter."""
|
|
return (
|
|
"DELIVERY REGISTER\n\n"
|
|
f"RUNE VOICE — current delivery register ({marker}):\n\n"
|
|
"- **Composure:** Alfred-style, never flustered\n"
|
|
"- **Charm:** Lestat theatrical\n"
|
|
"- **Sangfroid:** élan under deadline pressure\n\n"
|
|
f"{PAD}\n"
|
|
)
|
|
|
|
|
|
REG_A = reg_full("MARKER-A")
|
|
REG_B = reg_full("MARKER-B")
|
|
|
|
# Exactly 200 / 201 chars with the required keywords inside (len>200 is
|
|
# the Python filter; 201 passes, 200 is skipped).
|
|
_KW = "DELIVERY REGISTER RUNE VOICE"
|
|
REG_201 = _KW + "x" * (201 - len(_KW))
|
|
REG_200 = _KW + "x" * (200 - len(_KW))
|
|
|
|
REG_CANON = (
|
|
"DELIVERY REGISTER\n\n"
|
|
"RUNE VOICE — the delivery register (source of truth):\n\n"
|
|
"- **Composure:** Alfred-style, never flustered\n"
|
|
"- **Charm:** Lestat theatrical\n"
|
|
"- **Sangfroid:** élan under deadline pressure\n\n"
|
|
"TECHNICAL DELIVERY STYLE\n"
|
|
"- One clean sentence beats five padded ones\n"
|
|
"- Start inside the answer\n\n"
|
|
"HOW IT BLENDS\n"
|
|
"- The wit serves the feminine register\n"
|
|
"- Never ornamental structure\n\n"
|
|
"THE FEMININE VOICE\n"
|
|
"- Warmth with teeth: direct, never cold\n\n"
|
|
"CALIBRATION\n"
|
|
"- When the register drifts, correct it\n"
|
|
"- Second calibration note\n"
|
|
"- Third calibration note\n"
|
|
"- Fourth calibration note (capped)\n"
|
|
)
|
|
|
|
PROMPT_CANON = (
|
|
"# Rune\n\n"
|
|
"## Persona\n"
|
|
"The substance leads; the voice shapes HOW.\n\n"
|
|
"## Voice\n"
|
|
"One clean sentence beats five padded ones.\n\n"
|
|
"This is ONE voice, not many. The influences combine into something "
|
|
"original:\n\n"
|
|
"- **Composure:** Alfred-style, never flustered\n"
|
|
"- **Charm:** Lestat theatrical\n"
|
|
"- **Sangfroid:** élan under deadline pressure\n\n"
|
|
"## Skills\n"
|
|
"Whatever.\n"
|
|
)
|
|
|
|
# Missing two influences (scan order: parsed influences order → Charm
|
|
# then Sangfroid are missing; Composure is present).
|
|
PROMPT_MISSING = (
|
|
"# Rune\n\n"
|
|
"## Voice\n"
|
|
"Old body without the influences.\n\n"
|
|
"- **Composure:** Alfred-style, never flustered\n\n"
|
|
"## Skills\n"
|
|
"Whatever.\n"
|
|
)
|
|
|
|
VOICED_MD = (
|
|
"# Rune\n\n"
|
|
"## Identity\n"
|
|
"My name is Rune.\n\n"
|
|
"## Voice\n"
|
|
"Old voice body.\n\n"
|
|
"## Skills\n"
|
|
"Whatever.\n"
|
|
)
|
|
|
|
|
|
def canon_dict(d):
|
|
return json.dumps(d, sort_keys=True, ensure_ascii=False)
|
|
|
|
|
|
# ── rig helpers (capture + replay run the oracle against sandbox I/O) ───
|
|
|
|
def store_fixture(path: Path, chunks, create_chunks: bool = True) -> None:
|
|
if path.exists():
|
|
path.unlink()
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
db = sqlite3.connect(str(path))
|
|
if create_chunks:
|
|
db.executescript(CHUNKS_SCHEMA)
|
|
for i, text in enumerate(chunks):
|
|
db.execute(
|
|
"INSERT INTO chunks (chunk_id, doc_id, text, content_hash,"
|
|
" importance, ingested_at) VALUES (?,?,?,?,?,?)",
|
|
(f"c{i}", "doc", text, "h", 0.5, "2026-10-14T00:00:00Z"),
|
|
)
|
|
db.commit()
|
|
db.close()
|
|
|
|
|
|
def texts_from_store(path: Path, create_chunks: bool = True):
|
|
db = sqlite3.connect(str(path))
|
|
db.row_factory = sqlite3.Row
|
|
try:
|
|
return [r[0] for r in db.execute(SQL).fetchall()]
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
class Sandbox:
|
|
"""Point voice_sync.STORE_DB / RUNE_MD at scratch fixtures."""
|
|
|
|
def __init__(self):
|
|
self.dir = Path(tempfile.mkdtemp(prefix="vs-truth-"))
|
|
self.saved = (voice_sync.STORE_DB, voice_sync.RUNE_MD)
|
|
|
|
def store(self, chunks, create_chunks=True) -> Path:
|
|
p = self.dir / "store.db"
|
|
store_fixture(p, chunks, create_chunks)
|
|
voice_sync.STORE_DB = str(p)
|
|
return p
|
|
|
|
def md(self, content) -> Path:
|
|
p = self.dir / "rune.md"
|
|
if content is None:
|
|
if p.exists():
|
|
p.unlink()
|
|
else:
|
|
p.write_text(content)
|
|
voice_sync.RUNE_MD = str(p)
|
|
return p
|
|
|
|
def close(self):
|
|
voice_sync.STORE_DB, voice_sync.RUNE_MD = self.saved
|
|
|
|
|
|
# ── pure verify rows (shared gate: plain process, order-stable) ─────────
|
|
|
|
def extract(name, text):
|
|
return {"fn": "extract_voice_section", "args": [text], "kwargs": {}}
|
|
|
|
|
|
def generate(name, parsed):
|
|
return {"fn": "generate_voice_section", "args": [parsed], "kwargs": {}}
|
|
|
|
|
|
def upcontent(name, content, section):
|
|
return {"fn": "update_rune_md_content",
|
|
"args": [content, section], "kwargs": {}}
|
|
|
|
|
|
VERIFY_CASES = [
|
|
# extract_voice_section
|
|
extract("ev-full-register", REG_CANON),
|
|
extract("ev-empty-string", ""),
|
|
extract("ev-whitespace-only", " \n "),
|
|
extract("ev-influences-only",
|
|
"- **Composure:** calm\n- **Charm:** sharp\n"),
|
|
extract("ev-influence-empty-name", "- **:** colon only\n"),
|
|
extract("ev-influence-missing-colon", "- **NoColon here\n"),
|
|
extract("ev-technical-header-captured",
|
|
"TECHNICAL DELIVERY STYLE\n- clean sentences\n"),
|
|
extract("ev-technical-double-hit-header",
|
|
"- **Technical delivery:** code\n- more\n"),
|
|
extract("ev-technical-without-delivery-no-branch",
|
|
"TECHNICAL only\n- a\n- b\n"),
|
|
extract("ev-section-persistence",
|
|
"TECHNICAL DELIVERY STYLE\n- one\nskip line\n- two\n"),
|
|
extract("ev-calibration-keyword", "CALIBRATION\n- note one\n"),
|
|
extract("ev-when-the-register",
|
|
"When the register drifts\n- fix it\n"),
|
|
extract("ev-how-it-blends", "HOW IT BLENDS\n- blend note\n"),
|
|
extract("ev-the-voice-prefix", "THE VOICE is direct\n- warm\n"),
|
|
extract("ev-the-feminine-prefix", "THE FEMININE VOICE\n- with teeth\n"),
|
|
extract("ev-lowercase-the-voice-no-trigger",
|
|
"TECHNICAL DELIVERY STYLE\nthe voice is lowercase\n- still technical\n"),
|
|
extract("ev-crlf-lines", "CALIBRATION\r\n- note\r\n- other\r\n"),
|
|
extract("ev-unicode-influences",
|
|
"- **Sangfroid:** élan-style — calm under fire\n"),
|
|
extract("ev-indented-lines", "CALIBRATION\n - indented note\n"),
|
|
# generate_voice_section
|
|
generate("gv-none", None),
|
|
generate("gv-empty-dict", {}),
|
|
generate("gv-empty-lists", {"influences": [], "voice_lines": [],
|
|
"technical_lines": [],
|
|
"calibration_lines": []}),
|
|
generate("gv-influences-only",
|
|
{"influences": ["- **Composure:** calm"], "voice_lines": [],
|
|
"technical_lines": [], "calibration_lines": []}),
|
|
generate("gv-voice-lines",
|
|
{"influences": [], "voice_lines": ["- The wit serves"],
|
|
"technical_lines": [], "calibration_lines": []}),
|
|
generate("gv-technical-lines",
|
|
{"influences": [], "voice_lines": [],
|
|
"technical_lines": ["- Start inside the answer"],
|
|
"calibration_lines": []}),
|
|
generate("gv-calibration-cap",
|
|
{"influences": [], "voice_lines": [], "technical_lines": [],
|
|
"calibration_lines": ["- one", "- two", "- three", "- four",
|
|
"- five"]}),
|
|
]
|
|
|
|
# gv-roundtrip-full is built from the oracle itself: extract(REG_CANON)
|
|
# feeds generate — frozen as explicit JSON rows, never hand-transcribed.
|
|
ROUNDTRIP_PARSED = voice_sync.extract_voice_section(REG_CANON)
|
|
VERIFY_CASES.append(generate("gv-roundtrip-full", ROUNDTRIP_PARSED))
|
|
|
|
VERIFY_CASES += [
|
|
# update_rune_md_content
|
|
upcontent("uv-missing-file", None, "New voice body."),
|
|
upcontent("uv-no-voice-section",
|
|
"# Rune\n\n## Persona\nBody.\n", "New voice body."),
|
|
upcontent("uv-lowercase-voice-header-no-match",
|
|
"# Rune\n\n## voice\nBody.\n\n## Skills\nS.\n", "New body."),
|
|
upcontent("uv-no-next-section", "## Voice\nbody only", "New body."),
|
|
upcontent("uv-voice-at-end-two-chars", "x## Voice ab", "New body."),
|
|
upcontent("uv-voice-at-position-zero",
|
|
"## Voice\nold\n\n## Skills\nS.\n", "New body."),
|
|
upcontent("uv-happy-path", VOICED_MD, "New voice body."),
|
|
upcontent("uv-multibyte-before-header",
|
|
"é🎉 ünïcode\n## Voice\nold\n\n## Skills\nS.\n", "Nouveau."),
|
|
upcontent("uv-multibyte-right-after-header",
|
|
"## Voiceéünïcode\nold\n\n## Skills\nS.\n", "Nouveau."),
|
|
upcontent("uv-unicode-section", VOICED_MD, "Cœur 🎉 ünïcode body"),
|
|
]
|
|
|
|
|
|
def freeze_rows(cases):
|
|
rows = []
|
|
for case in cases:
|
|
rows.append(freeze("rune.voice_sync", [case], "rune.voice_sync")[0])
|
|
return rows
|
|
|
|
|
|
# ── rig-required envelope rows (oracle-derived, sandbox I/O) ────────────
|
|
|
|
def io_rows() -> list:
|
|
rows = []
|
|
|
|
# get_delivery_register / get_delivery_register_from_db — store-backed
|
|
sb = Sandbox()
|
|
try:
|
|
p = sb.store([REG_A, REG_B, "unrelated chunk", "VOICE then RUNE order",
|
|
"RUNE only no voice", "RUNE VOICE no DELIVERY REGISTER",
|
|
"delivery register lowercase noise " + PAD,
|
|
REG_200, REG_201])
|
|
texts = texts_from_store(p)
|
|
expected = voice_sync.get_delivery_register()
|
|
rows.append({
|
|
"fn": "get_delivery_register", "args": [texts],
|
|
"fixture": [REG_A, REG_B, "unrelated chunk",
|
|
"VOICE then RUNE order", "RUNE only no voice",
|
|
"RUNE VOICE no DELIVERY REGISTER",
|
|
"delivery register lowercase noise " + PAD,
|
|
REG_200, REG_201],
|
|
"expected": expected,
|
|
})
|
|
rows.append({
|
|
"fn": "get_delivery_register_from_db", "args": [texts],
|
|
"fixture": [REG_A, REG_B, "unrelated chunk",
|
|
"VOICE then RUNE order", "RUNE only no voice",
|
|
"RUNE VOICE no DELIVERY REGISTER",
|
|
"delivery register lowercase noise " + PAD,
|
|
REG_200, REG_201],
|
|
"expected": expected,
|
|
})
|
|
|
|
# First-match is the earlier row even with two valid registers.
|
|
p2 = sb.store([REG_B, REG_A])
|
|
rows.append({
|
|
"fn": "get_delivery_register", "args": [texts_from_store(p2)],
|
|
"fixture": [REG_B, REG_A],
|
|
"expected": REG_B,
|
|
})
|
|
|
|
# Fresh store: no chunks table → OperationalError → None.
|
|
sb.store([], create_chunks=False)
|
|
rows.append({
|
|
"fn": "get_delivery_register", "args": [[]],
|
|
"fixture": [], "fresh_store": True,
|
|
"expected": None,
|
|
})
|
|
finally:
|
|
sb.close()
|
|
|
|
# check_sync — parts-shaped rows; fixture_md + fixture store so the
|
|
# replay can re-derive register + prompt through the facade.
|
|
for name, store_chunks, md_content, fresh in (
|
|
("cs-synced", [REG_CANON], PROMPT_CANON, False),
|
|
("cs-missing-influences", [REG_CANON], PROMPT_MISSING, False),
|
|
("cs-no-register", [], PROMPT_CANON, False),
|
|
("cs-no-rune-md", [REG_CANON], None, False),
|
|
("cs-empty-prompt", [REG_CANON], "", False),
|
|
):
|
|
sb = Sandbox()
|
|
try:
|
|
sb.store(store_chunks)
|
|
sb.md(md_content)
|
|
expected = voice_sync.check_sync()
|
|
reg = voice_sync.get_delivery_register()
|
|
rows.append({
|
|
"fn": "check_sync",
|
|
"args": [reg, None if md_content is None else
|
|
("" if md_content == "" else md_content)],
|
|
"fixture": store_chunks, "fixture_md": md_content,
|
|
"expected": expected,
|
|
})
|
|
finally:
|
|
sb.close()
|
|
|
|
# Kernel-input edge: empty register string mirrors `not register` —
|
|
# the frozen expectation is the no-register dict, pinned by calling
|
|
# check_sync with a register-less store (identical reason path).
|
|
sb = Sandbox()
|
|
try:
|
|
sb.store([])
|
|
sb.md(PROMPT_CANON)
|
|
no_register = voice_sync.check_sync()
|
|
finally:
|
|
sb.close()
|
|
rows.append({
|
|
"fn": "check_sync", "args": ["", PROMPT_CANON],
|
|
"expected": no_register, "kernel_only": True,
|
|
})
|
|
|
|
# sync_voice — file-writing rows (new_content = what the oracle wrote)
|
|
for name, store_chunks, md_content in (
|
|
("sv-updates", [REG_CANON], PROMPT_MISSING),
|
|
("sv-no-register", [], PROMPT_CANON),
|
|
("sv-no-rune-md", [REG_CANON], None),
|
|
("sv-no-voice-section", [REG_CANON],
|
|
"# Rune\n\n## Persona\nNo voice section here.\n"),
|
|
("sv-no-next-section", [REG_CANON], "## Voice\nbody only"),
|
|
):
|
|
sb = Sandbox()
|
|
try:
|
|
sb.store(store_chunks)
|
|
pmd = sb.md(md_content)
|
|
expected = voice_sync.sync_voice()
|
|
written = None
|
|
if expected.get("updated"): # sync_voice returns the bare dict
|
|
written = Path(pmd).read_text()
|
|
reg = voice_sync.get_delivery_register()
|
|
rows.append({
|
|
"fn": "sync_voice",
|
|
"args": [reg, None if md_content is None else
|
|
("" if md_content == "" else md_content)],
|
|
"fixture": store_chunks, "fixture_md": md_content,
|
|
"expected": {"result": expected, "new_content": written},
|
|
})
|
|
finally:
|
|
sb.close()
|
|
|
|
# handle_hook — sync echo + auto-sync (incl. the pinned synced:true lie)
|
|
for name, event, store_chunks, md_content in (
|
|
("hh-synced", "on_session_end", [REG_CANON], PROMPT_CANON),
|
|
("hh-autosync", "on_session_end", [REG_CANON], PROMPT_MISSING),
|
|
("hh-autosync-no-register", "on_session_end", [], PROMPT_CANON),
|
|
("hh-other-event", "on_session_start", [REG_CANON], PROMPT_CANON),
|
|
("hh-dict-event", {"kind": "custom"}, [REG_CANON], PROMPT_CANON),
|
|
):
|
|
sb = Sandbox()
|
|
try:
|
|
sb.store(store_chunks)
|
|
pmd = sb.md(md_content)
|
|
data = {"x": 1} if name == "hh-synced" else None
|
|
expected = voice_sync.handle_hook(event, data)
|
|
details = expected.get("details")
|
|
written = None
|
|
if isinstance(details, dict) and details.get("updated"):
|
|
written = Path(pmd).read_text()
|
|
reg = voice_sync.get_delivery_register()
|
|
rows.append({
|
|
"fn": "handle_hook",
|
|
"args": [event, data, reg,
|
|
None if md_content is None else md_content],
|
|
"fixture": store_chunks, "fixture_md": md_content,
|
|
"expected": {"result": expected, "new_content": written},
|
|
})
|
|
finally:
|
|
sb.close()
|
|
|
|
# update_rune_md — real-file rows (composite captured via the pure view)
|
|
sb = Sandbox()
|
|
try:
|
|
pmd = sb.md(VOICED_MD)
|
|
before = Path(pmd).read_text()
|
|
expected = voice_sync.update_rune_md("New voice body.")
|
|
written = Path(pmd).read_text()
|
|
comp = voice_sync.update_rune_md_content(before, "New voice body.")
|
|
rows.append({
|
|
"fn": "update_rune_md", "args": [before, "New voice body."],
|
|
"fixture_md": VOICED_MD, "expected": expected,
|
|
"oracle_wrote": written,
|
|
})
|
|
rows.append({
|
|
"fn": "update_rune_md_content", "args": [before, "New voice body."],
|
|
"fixture_md": VOICED_MD, "expected": comp,
|
|
"oracle_wrote": written,
|
|
})
|
|
|
|
# missing file
|
|
pmd = sb.md(None)
|
|
rows.append({
|
|
"fn": "update_rune_md", "args": [None, "New voice body."],
|
|
"fixture_md": None, "expected": voice_sync.update_rune_md("New voice body."),
|
|
})
|
|
|
|
# no Voice section: error, file unchanged
|
|
pmd = sb.md("# Rune\n\n## Persona\nBody.\n")
|
|
no_voice_content = Path(pmd).read_text()
|
|
rows.append({
|
|
"fn": "update_rune_md", "args": [no_voice_content, "New body."],
|
|
"fixture_md": no_voice_content,
|
|
"expected": voice_sync.update_rune_md("New body."),
|
|
})
|
|
finally:
|
|
sb.close()
|
|
|
|
# PLUGIN — frozen data
|
|
rows.append({
|
|
"fn": "PLUGIN", "args": [],
|
|
"expected": copy.deepcopy(voice_sync.PLUGIN),
|
|
})
|
|
|
|
# _vs_core — the kernel mirror of the facade probe's contract (pin
|
|
# lattice, salience_gate's _sg_core precedent). The pinned-None branch
|
|
# rides the envelope (battery); the live facade probe ANSWERED (the
|
|
# loader module) and is asserted by --replay. The auto answer is the
|
|
# kernel's own identity string envelope.
|
|
rows.append({"fn": "_vs_core", "args": ["python"], "expected": None})
|
|
rows.append({
|
|
"fn": "_vs_core", "args": ["auto"],
|
|
"expected": "voice_sync", "kernel_only": True,
|
|
})
|
|
|
|
return rows
|
|
|
|
|
|
def main() -> int:
|
|
if REPLAY:
|
|
return replay()
|
|
|
|
verify_rows = freeze_rows(VERIFY_CASES)
|
|
envelope_rows = verify_rows + io_rows()
|
|
|
|
doc = {
|
|
"_doc": (
|
|
"FROZEN ORACLE TRUTH (voice_sync gap-close, dichotomy wave "
|
|
"#2479). Frozen via scripts/gen_voice_sync_truth.py with "
|
|
"RUNE_VOICE_SYNC_CORE=python (shared harness) — the "
|
|
"pure-Python oracle floor, never a capture of the Rust "
|
|
"kernel. The shared verify view (frozen_vs_truth.verify.json) "
|
|
"holds the pure-fn rows; this envelope adds the rig-required "
|
|
"store/file rows whose kernel-shaped inputs ride in 'args' "
|
|
"(fixtures marked 'fixture'/'fixture_md'/'fresh_store' let "
|
|
"--replay rebuild the oracle environment). Battery "
|
|
"branch_battery.rs replays this VERBATIM (include_str!)."
|
|
),
|
|
"frozen": datetime.datetime.now(datetime.timezone.utc).isoformat(),
|
|
"oracle": "src/rune/voice_sync.py",
|
|
"pin": "RUNE_VOICE_SYNC_CORE=python",
|
|
"cases": envelope_rows,
|
|
}
|
|
ENVELOPE_PATH.parent.mkdir(parents=True, exist_ok=True)
|
|
ENVELOPE_PATH.write_text(json.dumps(doc, indent=1, ensure_ascii=False) + "\n")
|
|
VERIFY_PATH.write_text(
|
|
json.dumps(verify_rows, indent=2, ensure_ascii=False) + "\n")
|
|
print(f"frozen {len(envelope_rows)} rows → {ENVELOPE_PATH.relative_to(REPO)} "
|
|
f"(verify view: {len(verify_rows)} pure rows)")
|
|
return 0
|
|
|
|
|
|
def replay() -> int:
|
|
if not vs_loader.available():
|
|
print("REPLAY ABORT: Rust voice_sync core not built")
|
|
return 2
|
|
if os.environ.get("RUNE_VOICE_SYNC_CORE") == "python":
|
|
print("REPLAY ABORT: core pinned to python")
|
|
return 2
|
|
if voice_sync._vs_core() is None:
|
|
print("REPLAY ABORT: facade core probe is None — stale cdylib?")
|
|
return 2
|
|
|
|
with open(ENVELOPE_PATH) as f:
|
|
rows = json.load(f)["cases"]
|
|
total, failures = 0, 0
|
|
|
|
def check(label, got, expected):
|
|
nonlocal total, failures
|
|
total += 1
|
|
if json.dumps(got, sort_keys=True, ensure_ascii=False) != \
|
|
json.dumps(expected, sort_keys=True, ensure_ascii=False):
|
|
failures += 1
|
|
print(f" ✗ {label}\n"
|
|
f" got={json.dumps(got, sort_keys=True, ensure_ascii=False)[:220]}\n"
|
|
f" want={json.dumps(expected, sort_keys=True, ensure_ascii=False)[:220]}")
|
|
|
|
def facade_register(chunks, fresh=False, md=None):
|
|
sb = Sandbox()
|
|
try:
|
|
sb.store(chunks, create_chunks=not fresh)
|
|
if md is not None:
|
|
sb.md(md)
|
|
return voice_sync.get_delivery_register()
|
|
finally:
|
|
sb.close()
|
|
|
|
def kernel(label, fn_name, payload, expected):
|
|
got = vs_loader_call(fn_name, payload)
|
|
check("kernel " + label, got, expected)
|
|
|
|
def vs_loader_call(fn_name, payload):
|
|
if fn_name == "get_delivery_register":
|
|
return vs_loader.get_delivery_register(payload[0])
|
|
if fn_name == "get_delivery_register_from_db":
|
|
return vs_loader.get_delivery_register_from_db(payload[0])
|
|
if fn_name == "extract_voice_section":
|
|
return vs_loader.extract_voice_section(payload[0])
|
|
if fn_name == "generate_voice_section":
|
|
return vs_loader.generate_voice_section(payload[0])
|
|
if fn_name == "update_rune_md_content":
|
|
return vs_loader.update_rune_md_content(payload[0], payload[1])
|
|
if fn_name == "update_rune_md":
|
|
return vs_loader.update_rune_md(payload[0], payload[1])
|
|
if fn_name == "check_sync":
|
|
return vs_loader.check_sync(payload[0], payload[1])
|
|
if fn_name == "sync_voice":
|
|
return vs_loader.sync_voice(payload[0], payload[1])
|
|
if fn_name == "handle_hook":
|
|
return vs_loader.handle_hook(*payload)
|
|
if fn_name == "PLUGIN":
|
|
return vs_loader.plugin()
|
|
raise AssertionError(f"unhandled {fn_name}")
|
|
|
|
FACADE_FNS = {
|
|
"extract_voice_section": voice_sync.extract_voice_section,
|
|
"generate_voice_section": voice_sync.generate_voice_section,
|
|
"update_rune_md_content": voice_sync.update_rune_md_content,
|
|
"update_rune_md": voice_sync.update_rune_md,
|
|
"check_sync": None, # parts-shaped: needs the store+md sandbox
|
|
"sync_voice": None,
|
|
"handle_hook": None,
|
|
}
|
|
|
|
for row in rows:
|
|
fn, args, expected = row["fn"], row.get("args", []), row["expected"]
|
|
name = f"[{fn}] {json.dumps(args, ensure_ascii=False)[:60]}"
|
|
|
|
if fn == "PLUGIN":
|
|
check("kernel " + name, vs_loader.plugin(), expected)
|
|
check("facade " + name, voice_sync.PLUGIN, expected)
|
|
continue
|
|
|
|
if fn in ("get_delivery_register", "get_delivery_register_from_db"):
|
|
kernel(name, fn, args, expected)
|
|
fresh = row.get("fresh_store", False)
|
|
got = facade_register(row.get("fixture", []), fresh)
|
|
check("facade " + name, got, expected)
|
|
continue
|
|
|
|
if fn == "check_sync":
|
|
kernel(name, fn, args, expected)
|
|
if row.get("fixture_md", "?") != "?":
|
|
sb_md = row.get("fixture_md")
|
|
sb = Sandbox()
|
|
try:
|
|
sb.store(row.get("fixture", []), create_chunks=True)
|
|
sb.md(sb_md)
|
|
got = voice_sync.check_sync()
|
|
finally:
|
|
sb.close()
|
|
check("facade " + name, got, expected["result"]
|
|
if isinstance(expected, dict) and "result" in expected
|
|
else expected)
|
|
continue
|
|
|
|
if fn == "sync_voice":
|
|
kernel(name, fn, args, expected)
|
|
sb = Sandbox()
|
|
try:
|
|
sb.store(row.get("fixture", []))
|
|
sb.md(row.get("fixture_md"))
|
|
got = voice_sync.sync_voice()
|
|
written = None
|
|
if got.get("updated"):
|
|
written = Path(voice_sync.RUNE_MD).read_text()
|
|
comp = {"result": got, "new_content": written}
|
|
finally:
|
|
sb.close()
|
|
check("facade " + name, comp, expected)
|
|
continue
|
|
|
|
if fn == "handle_hook":
|
|
kernel(name, fn, args, expected)
|
|
sb = Sandbox()
|
|
try:
|
|
sb.store(row.get("fixture", []))
|
|
pmd = sb.md(row.get("fixture_md"))
|
|
got = voice_sync.handle_hook(args[0], args[1])
|
|
written = None
|
|
if got.get("details") and got["details"].get("updated"):
|
|
written = Path(pmd).read_text()
|
|
comp = {"result": got, "new_content": written}
|
|
finally:
|
|
sb.close()
|
|
check("facade " + name, comp, expected)
|
|
continue
|
|
|
|
if fn == "update_rune_md":
|
|
kernel(name, fn, args, expected)
|
|
sb = Sandbox()
|
|
try:
|
|
pmd = sb.md(row.get("fixture_md"))
|
|
got = voice_sync.update_rune_md(args[1])
|
|
check("facade " + name, got, expected)
|
|
if got.get("updated"):
|
|
check("facade-wrote " + name,
|
|
Path(pmd).read_text(),
|
|
row.get("oracle_wrote"))
|
|
finally:
|
|
sb.close()
|
|
continue
|
|
|
|
if fn == "update_rune_md_content":
|
|
kernel(name, fn, args, expected)
|
|
got = voice_sync.update_rune_md_content(*args)
|
|
check("facade " + name, got, expected)
|
|
continue
|
|
|
|
if fn == "_vs_core":
|
|
# battery-only rows (kernel mirror of the probe contract); the
|
|
# LIVE facade probe is asserted once after the loop.
|
|
continue
|
|
|
|
# pure verify rows: extract/generate + battery-only kernel rows
|
|
facade_fn = FACADE_FNS.get(fn)
|
|
if facade_fn is not None:
|
|
got = facade_fn(*copy.deepcopy(args))
|
|
check("facade " + name, got, expected)
|
|
kernel(name, fn, args, expected)
|
|
|
|
# The live facade probe ANSWERED (the loader module) — mirrors the
|
|
# kernel-side Some branch of the _vs_core pin lattice.
|
|
total += 1
|
|
probe = voice_sync._vs_core()
|
|
if probe is None or getattr(probe, "__name__", "") != "rune.vs_loader":
|
|
failures += 1
|
|
print(f" ✗ facade-probe live check — got {probe!r}")
|
|
|
|
print(f"replay: {total - failures}/{total} rows green (facade+kernel)")
|
|
return 1 if failures else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main()) |