feat(voicebox): literal-100% residue — dichotomy wave

Port the six audit-missing fns name-preserving (exact snake names,
tool_protocol gaps.go law) into src/go/voicebox/gaps.go: _run (argv law,
VENV head), start_recording / stop_recording (the push-to-talk control
law incl. the active-flag ordering, >4000 gate, elif-wav unlink swallow),
_transcribe_and_type (join law + xdotool check=True except-arm + finally
unlink), on_press / on_release (key sentinel twins, recover-based
suppression). Pure logic 1:1; every device boundary (arecord Popen,
faster-whisper, xdotool, temp file, unlink, stderr) behind the
voiceboxSeams injectable; the pynput listener loop stays facade (its one
observable line byte-pinned as the talk_banner op).

Truth 6 -> 21 cases, now generator-owned (scripts/gen_voicebox_truth.py:
pynput/tempfile/subprocess/faster_whisper rig capturing the oracle's REAL
nested closures over its REAL recording dict; real temp files so
os.path/unlink run unpatched; TMP/ + REPO_ROOT normalization; venv
re-exec guard). Regen x2 byte-identical + --replay 21/21 stable. Battery
(TestTruthReplay via VB_TRUTH_PATH + 2 unit twins) green; gofmt/vet/
build clean; LIVE PARITY 100%: fresh oracle regen byte-identical to
frozen and battery green against it. DEFECT CAUGHT BY THE BATTERY:
calledProcessError dropped the quotes around the python list repr —
fixed to Command '[…]' returned non-zero exit status N (case 19 exposed
it). Python surface: canary imports voicebox BOTH pins (adjacent pytest
has no voicebox importer — plugin surface, task_gate precedent);
test_p1_03 67 passed/1 skipped BOTH pins after building memory_scoring +
memory_retrieval .so in this worktree (retired-oracle stubs recurse
without them — worktree artifact, not a regression). voicebox 5/11 ->
11/11 fn-level. Evidence: verification/vb-*.{log,exit,txt,md}.
This commit is contained in:
Nick committed 2026-10-15 09:17:38 +01:00
1 parent d268b365ac
commit 626f848397
16 files changed
+1815 -43

No files matched your search

+405
View File
@@ -0,0 +1,405 @@
#!/usr/bin/env python3
"""gen_voicebox_truth.py — freeze/replay truth for the voicebox literal-100%
residue (dichotomy wave #2479, final lane).
Freeze mode (default): drives the LIVE Python oracle (src/rune/voicebox.py)
and records effect rows into src/go/voicebox/truth.json:
* legacy surface (handle_hook, PLUGIN manifest) — regenerated from the
live module attributes so the whole file is generator-owned.
* _run — argv law ([VENV] + args) + the default kwargs ({}) via a
subprocess.run capture.
* push-to-talk — the nested talk() closures (start_recording,
stop_recording, _transcribe_and_type, on_press, on_release) driven
through a pynput rig: fake keyboard.Key sentinels + a Listener that
captures the real closures, so every case exercises the oracle's own
nested functions over the oracle's own `recording` dict. Device IO
(NamedTemporaryFile, arecord Popen, faster_whisper, xdotool, stderr)
is patched at the oracle's module globals; temp files are REAL files
(fixed per-case roots) so os.path.exists/getsize/unlink run unpatched
— only the absolute root is normalized to TMP/ in the frozen rows.
Replay mode (--replay <path>): re-runs the same scenarios against the
frozen file (regen byte-stability gate; the Go battery replays the same
rows via VB_TRUTH_PATH for live parity).
Determinism laws: fixed temp roots (no mkstemp randomness), per-case
counters reset, no clocks (the module is clock-free), stderr captured per
case. The oracle's import-time venv re-exec is guarded (the dedicated
venv must not exist — freeze halts loudly if it does).
"""
import io
import json
import os
import shutil
import subprocess as real_subprocess
import sys
import threading
import types
from pathlib import Path
REPO = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
TRUTH_PATH = os.path.join(REPO, "src", "go", "voicebox", "truth.json")
TMP_ROOT = "/tmp/vb_truth_fix"
VENV = "/tmp/opencode/voice-venv/bin/python3"
args = sys.argv[1:]
REPLAY_PATH = None
if "--replay" in args:
REPLAY_PATH = args[args.index("--replay") + 1]
if os.path.exists(VENV):
sys.stderr.write(f"REFUSING: {VENV} exists — oracle import would re-exec\n")
sys.exit(2)
sys.path.insert(0, os.path.join(REPO, "src", "rune"))
import voicebox as oracle # noqa: E402
# ── the rig ─────────────────────────────────────────────────────────────────
CTRL_R = "ctrl_r"
OTHER_KEY = "f9"
class FakePopen:
def __init__(self, argv, log):
self.argv, self.log = argv, log
def terminate(self):
self.log["terminated"] += 1
def wait(self):
self.log["waited"] += 1
def make_case_rig(wav_bytes, transcriber, xdotool_fail):
"""One hermetic per-case rig: fake subprocess/tempfile/sys + captured
pynput closures. Returns (case_state, handler-driver)."""
state = {
"counter": 0,
"popen": [],
"proc_log": {"terminated": 0, "waited": 0},
"runs": [],
"temp_calls": [],
"stderr": io.StringIO(),
"whisper": None,
"files": {}, # normalized name -> real path
}
root = os.path.join(TMP_ROOT, f"case{state['counter']}")
shutil.rmtree(TMP_ROOT, ignore_errors=True)
os.makedirs(TMP_ROOT, exist_ok=True)
state["root"] = root
def fake_run(argv, **kw):
state["runs"].append({"argv": argv, "kwargs": kw})
if xdotool_fail and argv and argv[0] == "xdotool":
raise real_subprocess.CalledProcessError(xdotool_fail, argv)
return real_subprocess.CompletedProcess(argv, 0)
def fake_popen(argv, **kw):
state["popen"].append({"argv": argv, "kwargs": kw})
return FakePopen(argv, state["proc_log"])
fake_subprocess = types.SimpleNamespace(run=fake_run, Popen=fake_popen)
def fake_ntf(*, suffix, prefix, delete):
name = f"{prefix}{state['counter']:03d}{suffix}"
state["counter"] += 1
real = os.path.join(state["root"], name)
os.makedirs(state["root"], exist_ok=True)
with open(real, "wb") as f:
f.write(b"\0" * wav_bytes)
norm = f"TMP/{name}"
state["files"][norm] = real
state["temp_calls"].append(
{"suffix": suffix, "prefix": prefix, "delete": delete, "name": norm})
return types.SimpleNamespace(name=real, close=lambda: None)
fake_tempfile = types.SimpleNamespace(NamedTemporaryFile=fake_ntf)
class SysShim:
stderr = state["stderr"]
def __getattr__(self, item):
return getattr(sys, item)
# faster_whisper fake (built per case; installed before the closure runs)
def make_whisper_module():
calls = {}
class Model:
def __init__(self, size, device=None, compute_type=None):
calls["model"] = {"size": size, "device": device,
"compute_type": compute_type}
def transcribe(self, wav, language=None):
calls["transcribe"] = {"wav": os.path.basename(wav),
"language": language}
if transcriber["kind"] == "raise":
raise ValueError(transcriber["message"])
segs = [types.SimpleNamespace(text=t) for t in transcriber["texts"]]
return segs, None
mod = types.ModuleType("faster_whisper")
mod.WhisperModel = Model
mod._calls = calls
return mod
whisper_mod = make_whisper_module()
# pynput fake: capture the closures talk() registers
captured = {}
class FakeListener:
def __init__(self, on_press=None, on_release=None):
captured["on_press"] = on_press
captured["on_release"] = on_release
def __enter__(self):
return self
def __exit__(self, *a):
return False
def join(self):
return None
keyboard = types.ModuleType("keyboard")
keyboard.Key = types.SimpleNamespace(ctrl_r=CTRL_R, f9=OTHER_KEY)
keyboard.Listener = FakeListener
pynput = types.ModuleType("pynput")
pynput.keyboard = keyboard
real_modules = {}
for name, mod in [("pynput", pynput), ("faster_whisper", whisper_mod)]:
real_modules[name] = sys.modules.get(name)
sys.modules[name] = mod
saved = {k: getattr(oracle, k) for k in ("subprocess", "tempfile", "sys")}
oracle.subprocess = fake_subprocess
oracle.tempfile = fake_tempfile
oracle.sys = SysShim()
class Key:
"""The key object handed to on_press/on_release: equality against
keyboard.Key.ctrl_r is identity against the rig's sentinel."""
def __init__(self, name):
self.name = name
def __eq__(self, other):
return other is getattr(keyboard.Key, self.name, None) if self.name else False
def __hash__(self):
return hash(self.name)
def run_steps(steps):
# talk() in a thread: defines the closures, prints the banner,
# enters+exits the fake listener, returns.
t = threading.Thread(target=oracle.talk, daemon=True)
t.start()
t.join()
for step in steps:
if "press" in step:
captured["on_press"](Key(step["press"]))
elif "release" in step:
captured["on_release"](Key(step["release"]))
elif step.get("delete_wav"):
for real in state["files"].values():
try:
os.unlink(real)
except OSError:
pass
def effects():
popen = [
{"argv": [norm_argv(a) for a in p["argv"]], "kwargs": p["kwargs"]}
for p in state["popen"]]
runs = [
{"argv": [norm_argv(a) for a in r["argv"]], "kwargs": r["kwargs"]}
for r in state["runs"]]
wav_after = {n: os.path.exists(r) for n, r in state["files"].items()}
whisper = None
if whisper_mod._calls.get("model"):
whisper = dict(whisper_mod._calls)
# the talk() banner is facade surface (pinned by the talk_banner
# op) — strip it so effect rows carry closure law only
lines = [ln for ln in state["stderr"].getvalue().splitlines()
if ln != TALK_BANNER]
return {
"stderr_lines": lines,
"popen": popen,
"terminated": state["proc_log"]["terminated"],
"waited": state["proc_log"]["waited"],
"runs": runs,
"temp_calls": state["temp_calls"],
"wav_exists_after": wav_after,
"whisper": whisper,
}
def teardown():
for k, v in saved.items():
setattr(oracle, k, v)
for name, mod in real_modules.items():
if mod is None:
sys.modules.pop(name, None)
else:
sys.modules[name] = mod
return run_steps, effects, teardown
def norm_argv(a):
return f"TMP/{os.path.basename(a)}" if isinstance(a, str) and a.startswith(TMP_ROOT) else a
# ── _run cases (module-level fn; subprocess.run capture only) ──────────────
def run_cases():
here = os.path.dirname(os.path.abspath(oracle.__file__))
def repo_norm(a):
if isinstance(a, str) and a.startswith(here):
return "REPO_ROOT/src/rune" + a[len(here):]
return a
cases = []
for name, argv in [
("run-say", [oracle.SAY, "hello world"]),
("run-listen", [oracle.LISTEN]),
("run-multi-args", ["a", "b", "c"]),
]:
seen = {}
def fake_run(a, **kw):
seen["argv"] = list(a)
seen["kwargs"] = kw
return real_subprocess.CompletedProcess(list(a), 0)
fake_subprocess = types.SimpleNamespace(run=fake_run, Popen=None)
saved = oracle.subprocess
oracle.subprocess = fake_subprocess
try:
ret = oracle._run(*argv)
cases.append({
"op": "run", "case": name,
"input": {"args": [repo_norm(a) for a in argv]},
"want": {"argv": [repo_norm(a) for a in seen["argv"]],
"kwargs": seen["kwargs"],
"returncode": ret.returncode},
})
finally:
oracle.subprocess = saved
return cases
# ── push-to-talk cases (the nested closures via the pynput rig) ────────────
OK_TEXTS = {"kind": "texts", "texts": [" hello ", " world "]}
UNICODE_TEXTS = {"kind": "texts", "texts": [" Grüße ", "世界"]}
EMPTY_TEXTS = {"kind": "texts", "texts": [" ", "", "\t"]}
RAISE = {"kind": "raise", "message": "boom"}
def p2t_cases():
cases = []
def one(name, steps, wav_bytes, transcriber=None, xdotool_fail=0):
run_steps, effects, teardown = make_case_rig(
wav_bytes, transcriber or OK_TEXTS, xdotool_fail)
try:
run_steps(steps)
eff = effects()
finally:
teardown()
cases.append({
"op": "push2talk", "case": name,
"input": {"steps": steps, "wav_bytes": wav_bytes,
"transcriber": transcriber or OK_TEXTS,
"xdotool_fail": xdotool_fail},
"want": eff,
})
one("p2t-press-release-small",
[{"press": CTRL_R}, {"release": CTRL_R}], wav_bytes=10)
one("p2t-press-release-big",
[{"press": CTRL_R}, {"release": CTRL_R}], wav_bytes=4001)
one("p2t-double-press-guard",
[{"press": CTRL_R}, {"press": CTRL_R}, {"release": CTRL_R}],
wav_bytes=10)
one("p2t-release-cold-noop", [{"release": CTRL_R}], wav_bytes=10)
one("p2t-other-key-ignored",
[{"press": OTHER_KEY}, {"release": OTHER_KEY}], wav_bytes=10)
one("p2t-release-other-key-keeps-recording",
[{"press": CTRL_R}, {"release": OTHER_KEY}, {"release": CTRL_R}],
wav_bytes=10)
one("p2t-wav-deleted-before-release",
[{"press": CTRL_R}, {"delete_wav": True}, {"release": CTRL_R}],
wav_bytes=4001)
one("p2t-transcribe-raise",
[{"press": CTRL_R}, {"release": CTRL_R}], wav_bytes=4001,
transcriber=RAISE)
one("p2t-transcribe-empty-join",
[{"press": CTRL_R}, {"release": CTRL_R}], wav_bytes=4001,
transcriber=EMPTY_TEXTS)
one("p2t-xdotool-check-true-fails",
[{"press": CTRL_R}, {"release": CTRL_R}], wav_bytes=4001,
xdotool_fail=1)
one("p2t-unicode-text",
[{"press": CTRL_R}, {"release": CTRL_R}], wav_bytes=4001,
transcriber=UNICODE_TEXTS)
return cases
# ── legacy surface (regenerated from the live module) ──────────────────────
TALK_BANNER = ("hold RIGHT-CTRL to talk; release to transcribe+type. "
"Ctrl-C to quit.")
def legacy_cases():
hook = oracle.handle_hook("test_event", {"k": "v"})
hook_empty = oracle.handle_hook("", None)
return [
{"op": "hook", "want": hook},
{"op": "plugin_name", "want": oracle.PLUGIN["name"]},
{"op": "plugin_caps", "want": oracle.PLUGIN["capabilities"]},
{"op": "hook_empty_oracle", "want": hook_empty,
"note": "oracle returns ok for empty event — Go improves by validating"},
{"op": "plugin_full"},
{"op": "hook_data_ignored"},
{"op": "talk_banner", "want": TALK_BANNER},
]
def main():
cases = legacy_cases() + run_cases() + p2t_cases()
doc = {
"frozen": "voicebox-literal-100%",
"oracle": "src/rune/voicebox.py",
"note": "Frozen via scripts/gen_voicebox_truth.py against the live "
"Python oracle (pynput/tempfile/subprocess/faster_whisper rig; "
"temp roots normalized to TMP/). Battery embeds this VERBATIM.",
"cases": cases,
}
if REPLAY_PATH:
with open(REPLAY_PATH) as f:
frozen = json.load(f)["cases"]
diverged = 0
for new, old in zip(cases, frozen):
if json.dumps(new, sort_keys=True) != json.dumps(old, sort_keys=True):
diverged += 1
print(f"DIVERGE {new.get('case') or new['op']}")
print(f"replay: {len(cases) - diverged}/{len(cases)} cases stable")
return 1 if diverged else 0
os.makedirs(os.path.dirname(TRUTH_PATH), exist_ok=True)
with open(TRUTH_PATH, "w") as f:
json.dump(doc, f, indent=1, ensure_ascii=False)
f.write("\n")
print(f"frozen {len(cases)} cases -> {os.path.relpath(TRUTH_PATH, REPO)}")
return 0
if __name__ == "__main__":
sys.exit(main())
+238
View File
@@ -0,0 +1,238 @@
// battery_gaps.go — replay the literal-100% residue truth rows (ops `run`
// and `push2talk`) against the ported fns, using a fake-seam harness that
// mirrors the generator rig (scripts/gen_voicebox_truth.py) case for case:
// same temp-name law, same effect vocabulary, same normalization (TMP/).
// The talk() banner is facade surface — pinned by the `talk_banner` op,
// not emitted here.
package main
import (
"encoding/json"
"errors"
"fmt"
"os"
"path"
"strings"
)
// talkBanner — talk()'s stderr banner, byte-pinned (the pynput listener
// loop itself stays facade; this is its one observable line).
const talkBanner = "hold RIGHT-CTRL to talk; release to transcribe+type. Ctrl-C to quit."
func caseName(c map[string]any) string {
if n, ok := c["case"].(string); ok {
return n
}
return c["op"].(string)
}
func argvStrs(argv []any) []string {
out := make([]string, len(argv))
for i, a := range argv {
out[i], _ = a.(string)
}
return out
}
// pyListRepr — Python repr(["a", "b"]): single-quoted, ", "-joined
// (fixtures carry no embedded quotes — battery fixtures guarantee it).
func pyListRepr(argv []string) string {
quoted := make([]string, len(argv))
for i, a := range argv {
quoted[i] = "'" + a + "'"
}
return "[" + strings.Join(quoted, ", ") + "]"
}
// calledProcessError — str(subprocess.CalledProcessError(rc, argv)) twin.
func calledProcessError(argv []string, rc int) error {
return fmt.Errorf("Command '%s' returned non-zero exit status %d.", pyListRepr(argv), rc)
}
// compareEffects — canonical JSON byte-compare of got vs want (maps
// marshal key-sorted on both sides; python freezes with sort-stable dumps).
func compareEffects(i int, name string, got, want any) int {
gj, _ := json.Marshal(got)
wj, _ := json.Marshal(want)
if string(gj) != string(wj) {
fmt.Printf("case %d %s: MISMATCH\n got: %s\n want: %s\n", i, name, gj, wj)
return 1
}
return 0
}
// replayRun — the _run argv law.
func replayRun(i int, c map[string]any) int {
input := c["input"].(map[string]any)
args := argvStrs(input["args"].([]any))
var gotArgv []string
s := &voiceboxSeams{
venv: VENV,
runCommand: func(argv []string) int {
gotArgv = append([]string{}, argv...)
return 0
},
}
rc := _run(s, args...)
got := map[string]any{
"argv": strList(gotArgv),
"kwargs": map[string]any{},
"returncode": rc,
}
return compareEffects(i, caseName(c), got, c["want"])
}
func strList(xs []string) []string {
if xs == nil {
return []string{}
}
return xs
}
// ── the push-to-talk fake-seam harness ─────────────────────────────────────
type fakeWav struct {
size int64
deleted bool
}
type fakeProc struct{ st *p2tState }
func (p *fakeProc) terminate() error { p.st.terminated++; return nil }
func (p *fakeProc) wait() error { p.st.waited++; return nil }
type p2tState struct {
tempCalls []map[string]any
popen []map[string]any
runs []map[string]any
files map[string]*fakeWav
whisper map[string]any
terminated int
waited int
stderr strings.Builder
}
func (st *p2tState) effects() map[string]any {
wavAfter := map[string]bool{}
for name, f := range st.files {
wavAfter[name] = !f.deleted
}
out := st.stderr.String()
lines := []string{}
if out != "" {
lines = strings.Split(strings.TrimSuffix(out, "\n"), "\n")
}
var whisper any
if st.whisper != nil {
whisper = st.whisper
}
if st.popen == nil {
st.popen = []map[string]any{}
}
if st.runs == nil {
st.runs = []map[string]any{}
}
if st.tempCalls == nil {
st.tempCalls = []map[string]any{}
}
return map[string]any{
"stderr_lines": lines,
"popen": st.popen,
"terminated": st.terminated,
"waited": st.waited,
"runs": st.runs,
"temp_calls": st.tempCalls,
"wav_exists_after": wavAfter,
"whisper": whisper,
}
}
// replayPush2Talk — drive the ported closures over the fake seams exactly
// as the generator rig drove the oracle's nested closures.
func replayPush2Talk(i int, c map[string]any) int {
input := c["input"].(map[string]any)
transcriber := input["transcriber"].(map[string]any)
wavBytes := int64(input["wav_bytes"].(float64))
xdotoolFail := int(input["xdotool_fail"].(float64))
st := &p2tState{files: map[string]*fakeWav{}}
counter := 0
rec := &recording{}
s := &voiceboxSeams{
venv: VENV,
makeTempWav: func() (string, error) {
name := fmt.Sprintf("TMP/push2talk-%03d.wav", counter)
counter++
st.tempCalls = append(st.tempCalls, map[string]any{
"suffix": ".wav", "prefix": "push2talk-", "delete": false, "name": name})
st.files[name] = &fakeWav{size: wavBytes}
return name, nil
},
spawnRecorder: func(argv []string) (recorderProc, error) {
st.popen = append(st.popen, map[string]any{
"argv": strList(append([]string{}, argv...)), "kwargs": map[string]any{}})
return &fakeProc{st: st}, nil
},
transcribe: func(wav string) ([]whisperSegment, error) {
st.whisper = map[string]any{
"model": map[string]any{"size": "small", "device": "auto", "compute_type": "auto"},
"transcribe": map[string]any{
"wav": path.Base(wav), "language": "en"},
}
if kind, _ := transcriber["kind"].(string); kind == "raise" {
return nil, errors.New(transcriber["message"].(string))
}
texts := transcriber["texts"].([]any)
segs := make([]whisperSegment, len(texts))
for j, t := range texts {
segs[j] = whisperSegment{text: t.(string)}
}
return segs, nil
},
typeText: func(argv []string) error {
st.runs = append(st.runs, map[string]any{
"argv": strList(append([]string{}, argv...)),
"kwargs": map[string]any{
"check": true, // subprocess.run(..., check=True)
}})
if xdotoolFail > 0 {
return calledProcessError(argv, xdotoolFail)
}
return nil
},
removeFile: func(p string) error {
if f, ok := st.files[p]; ok && !f.deleted {
f.deleted = true
return nil
}
return os.ErrNotExist // the OSError the oracle swallows
},
fileState: func(p string) (bool, int64) {
if f, ok := st.files[p]; ok && !f.deleted {
return true, f.size
}
return false, 0
},
stderr: &st.stderr,
runCommand: func([]string) int {
return 0
},
}
for _, rawStep := range input["steps"].([]any) {
step := rawStep.(map[string]any)
if k, ok := step["press"].(string); ok {
on_press(k, rec, s)
}
if k, ok := step["release"].(string); ok {
on_release(k, rec, s)
}
if del, _ := step["delete_wav"].(bool); del {
for _, f := range st.files {
f.deleted = true
}
}
}
return compareEffects(i, caseName(c), st.effects(), c["want"])
}
+175
View File
@@ -0,0 +1,175 @@
// gaps.go — the literal-100% residue for voicebox (#2479 dichotomy wave,
// final lane). Name-preserving ports of the six audit-missing oracle fns:
// _run, start_recording, stop_recording, _transcribe_and_type, on_press,
// on_release.
//
// PORT LAW (audio/IO-adjacent module): the control logic ports 1:1; every
// device boundary (arecord spawn, faster-whisper STT, xdotool typing, temp
// file creation, unlink, stderr) sits behind the voiceboxSeams injectable.
// The pynput listener loop itself stays facade (the oracle imports pynput
// inside talk(); the with-Listener/join wiring is UI surface, not logic).
//
// Semantics preserved exactly, including the sharp edges the battery pins:
// - start_recording sets active=true BEFORE the temp/spawn, so a failure
// leaves the partially-mutated state the oracle leaves.
// - stop_recording clears tmp BEFORE the size probe; the elif-wav unlink
// swallows errors (OSError → pass); size gate is >4000, strictly.
// - _transcribe_and_type: the join law " ".join(s.text.strip()).strip();
// xdotool runs check=True — its failure lands in the SAME except arm
// as a whisper failure ("transcribe error: …"), and finally ALWAYS
// unlinks (best-effort).
// - on_press/on_release suppress every exception (the oracle logs at
// debug, which is silent unconfigured) — recover-and-drop twin.
package main
import (
"fmt"
"io"
"strings"
)
// VENV — the oracle's dedicated voice venv interpreter (module-level VENV).
const VENV = "/tmp/opencode/voice-venv/bin/python3"
// keyCtrlR — the pynput keyboard.Key.ctrl_r sentinel twin. Keys cross the
// seam as their pynput names ("ctrl_r", "f9", …).
const keyCtrlR = "ctrl_r"
// recorderProc — the subprocess.Popen handle twin (arecord).
type recorderProc interface {
terminate() error
wait() error
}
// whisperSegment — a faster-whisper segment; only .text is consumed.
type whisperSegment struct {
text string
}
// voiceboxSeams — every device boundary the oracle touches, injectable.
type voiceboxSeams struct {
venv string // argv head for _run (oracle: VENV)
// NamedTemporaryFile(suffix=".wav", prefix="push2talk-", delete=False)
// + .close(): returns the path.
makeTempWav func() (string, error)
// Popen(["arecord", "-f", "S16_LE", "-r", "16000", "-c", "1", tmp])
spawnRecorder func(argv []string) (recorderProc, error)
// faster_whisper WhisperModel("small", device="auto").transcribe(
// wav, language="en") → segments.
transcribe func(wav string) ([]whisperSegment, error)
// subprocess.run(["xdotool", "type", "--delay", "5", text], check=True)
typeText func(argv []string) error
// os.unlink — errors swallowed at both call sites (OSError → pass).
removeFile func(path string) error
// os.path.exists + os.path.getsize in one probe (the oracle's
// TOCTOU window between the two calls is single-threaded-deterministic
// here; documented divergence: one stat, not two).
fileState func(path string) (exists bool, size int64)
// print(..., file=sys.stderr)
stderr io.Writer
// subprocess.run for _run: executes argv to completion, returns the
// CompletedProcess returncode.
runCommand func(argv []string) int
}
// recording — the oracle's `recording` dict twin ("active"/"proc"/"tmp";
// tmp None ⇔ "").
type recording struct {
active bool
proc recorderProc
tmp string
}
// _run — name-preserving port of the oracle `_run(*args, **kw)`:
// subprocess.run([VENV] + list(args), **kw). The oracle's callers pass no
// kw, so the default-CompletedProcess shape (returncode only) is the law;
// argv construction is the pinned observable.
func _run(s *voiceboxSeams, args ...string) int {
return s.runCommand(append([]string{s.venv}, args...))
}
// start_recording — 1:1 with the oracle's nested closure.
func start_recording(rec *recording, s *voiceboxSeams) error {
if rec.active {
return nil
}
rec.active = true
name, err := s.makeTempWav()
if err != nil {
return err // active stays true — oracle's partial-mutation law
}
rec.tmp = name
proc, err := s.spawnRecorder([]string{"arecord", "-f", "S16_LE", "-r", "16000", "-c", "1", rec.tmp})
if err != nil {
return err // tmp stays set — oracle's partial-mutation law
}
rec.proc = proc
fmt.Fprint(s.stderr, "■ recording...\n")
return nil
}
// stop_recording — 1:1 with the oracle's nested closure. The transcribe
// branch deliberately has no error return: the oracle propagates nothing
// (exceptions land in on_release's suppressor or crash the direct caller;
// the deterministic battery path never raises).
func stop_recording(rec *recording, s *voiceboxSeams) {
if !rec.active {
return
}
rec.active = false
if rec.proc != nil {
_ = rec.proc.terminate()
_ = rec.proc.wait()
}
wav := rec.tmp
rec.tmp = ""
if wav == "" {
return
}
if exists, size := s.fileState(wav); exists && size > 4000 {
fmt.Fprint(s.stderr, "■ transcribing...\n")
_transcribe_and_type(wav, s)
return
}
_ = s.removeFile(wav) // OSError → pass
}
// _transcribe_and_type — the STT→type pipeline with the whisper/xdotool
// IO peeled onto seams. The except arm carries BOTH failure sources
// (whisper import/model/transcribe and xdotool check=True); finally
// always unlinks, best-effort.
func _transcribe_and_type(wav string, s *voiceboxSeams) {
defer func() { _ = s.removeFile(wav) }() // finally: unlink, OSError → pass
segments, err := s.transcribe(wav)
if err != nil {
fmt.Fprintf(s.stderr, "transcribe error: %v\n", err)
return
}
parts := make([]string, len(segments))
for i, seg := range segments {
parts[i] = strings.TrimSpace(seg.text)
}
text := strings.TrimSpace(strings.Join(parts, " "))
if text != "" {
if err := s.typeText([]string{"xdotool", "type", "--delay", "5", text}); err != nil {
fmt.Fprintf(s.stderr, "transcribe error: %v\n", err)
}
}
}
// on_press — pynput on_press twin: RIGHT_CTRL arms recording; every
// exception is suppressed (oracle: logger.debug, silent unconfigured).
func on_press(key string, rec *recording, s *voiceboxSeams) {
defer func() { _ = recover() }() // except Exception → debug log → silent
if key == keyCtrlR {
_ = start_recording(rec, s)
}
}
// on_release — pynput on_release twin: RIGHT_CTRL release stops+transcribes.
func on_release(key string, rec *recording, s *voiceboxSeams) {
defer func() { _ = recover() }() // except Exception → debug log → silent
if key == keyCtrlR {
stop_recording(rec, s)
}
}
+37 -7
View File
@@ -52,11 +52,36 @@ func main() {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
var cases []map[string]any
if err := json.Unmarshal(raw, &cases); err != nil {
cases, err := loadCases(raw)
if err != nil {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
if fails := runReplay(cases); fails > 0 {
fmt.Fprintf(os.Stderr, "PARITY FAILED: %d divergence(s)\n", fails)
os.Exit(1)
}
fmt.Printf("PARITY OK: %d cases (2 improvements applied: empty-event validation, manifest pin) vs the Python oracle\n", len(cases))
}
// loadCases — accepts the generator's doc-wrapped form
// ({"frozen":…, "cases":[…]}) and the legacy flat list.
func loadCases(raw []byte) ([]map[string]any, error) {
var doc struct {
Cases []map[string]any `json:"cases"`
}
if err := json.Unmarshal(raw, &doc); err == nil && doc.Cases != nil {
return doc.Cases, nil
}
var cases []map[string]any
if err := json.Unmarshal(raw, &cases); err != nil {
return nil, err
}
return cases, nil
}
// runReplay — the shared battery body (binary `replay` + go test).
func runReplay(cases []map[string]any) int {
fails := 0
for i, c := range cases {
op, _ := c["op"].(string)
@@ -126,11 +151,16 @@ func main() {
fmt.Printf("case %d plugin_caps[%d]\n", i, j)
}
}
case "run":
fails += replayRun(i, c)
case "push2talk":
fails += replayPush2Talk(i, c)
case "talk_banner":
if w, _ := c["want"].(string); w != talkBanner {
fails++
fmt.Printf("case %d talk_banner: got %q want %q\n", i, talkBanner, w)
}
}
}
if fails > 0 {
fmt.Fprintf(os.Stderr, "PARITY FAILED: %d divergence(s)\n", fails)
os.Exit(1)
}
fmt.Printf("PARITY OK: %d cases (2 improvements applied: empty-event validation, manifest pin) vs the Python oracle\n", len(cases))
return fails
}
+103
View File
@@ -0,0 +1,103 @@
package main
import (
"os"
"testing"
)
// TestTruthReplay — the battery: replays the frozen truth file VERBATIM
// (generator: scripts/gen_voicebox_truth.py, frozen from the LIVE oracle).
// VB_TRUTH_PATH overrides the file (live-parity gate: point it at a fresh
// regeneration and the same battery must stay green).
func TestTruthReplay(t *testing.T) {
path := "truth.json"
if p := os.Getenv("VB_TRUTH_PATH"); p != "" {
path = p
}
raw, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read truth: %v", err)
}
cases, err := loadCases(raw)
if err != nil {
t.Fatalf("parse truth: %v", err)
}
if len(cases) < 15 {
t.Fatalf("truth too small: %d cases (regenerate)", len(cases))
}
fails := runReplay(cases)
if fails > 0 {
t.Fatalf("%d divergence(s) against %s", fails, path)
}
t.Logf("PARITY OK: %d cases vs the Python oracle (%s)", len(cases), path)
}
// TestRunArgvLaw — the _run argv law pinned directly (unit twin of the
// replay op): VENV is always the head, args ride in order.
func TestRunArgvLaw(t *testing.T) {
var got []string
s := &voiceboxSeams{
venv: VENV,
runCommand: func(argv []string) int {
got = append([]string{}, argv...)
return 3
},
}
if rc := _run(s, "a", "b"); rc != 3 {
t.Fatalf("returncode: got %d want 3", rc)
}
want := []string{VENV, "a", "b"}
if len(got) != len(want) {
t.Fatalf("argv len: got %v want %v", got, want)
}
for i := range want {
if got[i] != want[i] {
t.Fatalf("argv[%d]: got %q want %q", i, got[i], want[i])
}
}
}
// TestPush2TalkGuardLaws — the two idempotence guards pinned directly.
func TestPush2TalkGuardLaws(t *testing.T) {
st := &p2tState{files: map[string]*fakeWav{}}
n := 0
s := &voiceboxSeams{
venv: VENV,
makeTempWav: func() (string, error) {
name := "TMP/x.wav"
st.files[name] = &fakeWav{size: 10}
return name, nil
},
spawnRecorder: func(argv []string) (recorderProc, error) {
n++
return &fakeProc{st: st}, nil
},
transcribe: func(string) ([]whisperSegment, error) {
return nil, nil
},
typeText: func([]string) error { return nil },
removeFile: func(string) error { return nil },
fileState: func(p string) (bool, int64) {
if f, ok := st.files[p]; ok && !f.deleted {
return true, f.size
}
return false, 0
},
stderr: &st.stderr,
runCommand: func([]string) int { return 0 },
}
rec := &recording{}
start_recording(rec, s)
start_recording(rec, s) // guard: second call is a no-op
if n != 1 {
t.Fatalf("double-start spawned %d procs, want 1", n)
}
stop_recording(rec, s) // small file → unlink branch
if st.terminated != 1 || st.waited != 1 {
t.Fatalf("terminate/wait: %d/%d want 1/1", st.terminated, st.waited)
}
stop_recording(rec, s) // guard: inactive no-op
if st.terminated != 1 {
t.Fatalf("inactive stop re-terminated: %d", st.terminated)
}
}
+790 -36
View File
@@ -1,37 +1,791 @@
[
{
"op": "hook",
"want": {
"status": "ok",
"event": "test_event",
"handler": "default"
}
},
{
"op": "plugin_name",
"want": "voicebox"
},
{
"op": "plugin_caps",
"want": [
"voice",
"write",
"voicebox"
]
},
{
"op": "hook_empty_oracle",
"want": {
"status": "ok",
"event": "",
"handler": "default"
{
"frozen": "voicebox-literal-100%",
"oracle": "src/rune/voicebox.py",
"note": "Frozen via scripts/gen_voicebox_truth.py against the live Python oracle (pynput/tempfile/subprocess/faster_whisper rig; temp roots normalized to TMP/). Battery embeds this VERBATIM.",
"cases": [
{
"op": "hook",
"want": {
"status": "ok",
"event": "test_event",
"handler": "default"
}
},
"note": "oracle returns ok for empty event \u2014 Go improves by validating"
},
{
"op": "plugin_full"
},
{
"op": "hook_data_ignored"
}
]
{
"op": "plugin_name",
"want": "voicebox"
},
{
"op": "plugin_caps",
"want": [
"voice",
"write",
"voicebox"
]
},
{
"op": "hook_empty_oracle",
"want": {
"status": "ok",
"event": "",
"handler": "default"
},
"note": "oracle returns ok for empty event — Go improves by validating"
},
{
"op": "plugin_full"
},
{
"op": "hook_data_ignored"
},
{
"op": "talk_banner",
"want": "hold RIGHT-CTRL to talk; release to transcribe+type. Ctrl-C to quit."
},
{
"op": "run",
"case": "run-say",
"input": {
"args": [
"REPO_ROOT/src/rune/voicebox_say.py",
"hello world"
]
},
"want": {
"argv": [
"/tmp/opencode/voice-venv/bin/python3",
"REPO_ROOT/src/rune/voicebox_say.py",
"hello world"
],
"kwargs": {},
"returncode": 0
}
},
{
"op": "run",
"case": "run-listen",
"input": {
"args": [
"REPO_ROOT/src/rune/voicebox_listen.py"
]
},
"want": {
"argv": [
"/tmp/opencode/voice-venv/bin/python3",
"REPO_ROOT/src/rune/voicebox_listen.py"
],
"kwargs": {},
"returncode": 0
}
},
{
"op": "run",
"case": "run-multi-args",
"input": {
"args": [
"a",
"b",
"c"
]
},
"want": {
"argv": [
"/tmp/opencode/voice-venv/bin/python3",
"a",
"b",
"c"
],
"kwargs": {},
"returncode": 0
}
},
{
"op": "push2talk",
"case": "p2t-press-release-small",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 10,
"transcriber": {
"kind": "texts",
"texts": [
" hello ",
" world "
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [
"■ recording..."
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": null
}
},
{
"op": "push2talk",
"case": "p2t-press-release-big",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 4001,
"transcriber": {
"kind": "texts",
"texts": [
" hello ",
" world "
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [
"■ recording...",
"■ transcribing..."
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [
{
"argv": [
"xdotool",
"type",
"--delay",
"5",
"hello world"
],
"kwargs": {
"check": true
}
}
],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": {
"model": {
"size": "small",
"device": "auto",
"compute_type": "auto"
},
"transcribe": {
"wav": "push2talk-000.wav",
"language": "en"
}
}
}
},
{
"op": "push2talk",
"case": "p2t-double-press-guard",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"press": "ctrl_r"
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 10,
"transcriber": {
"kind": "texts",
"texts": [
" hello ",
" world "
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [
"■ recording..."
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": null
}
},
{
"op": "push2talk",
"case": "p2t-release-cold-noop",
"input": {
"steps": [
{
"release": "ctrl_r"
}
],
"wav_bytes": 10,
"transcriber": {
"kind": "texts",
"texts": [
" hello ",
" world "
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [],
"popen": [],
"terminated": 0,
"waited": 0,
"runs": [],
"temp_calls": [],
"wav_exists_after": {},
"whisper": null
}
},
{
"op": "push2talk",
"case": "p2t-other-key-ignored",
"input": {
"steps": [
{
"press": "f9"
},
{
"release": "f9"
}
],
"wav_bytes": 10,
"transcriber": {
"kind": "texts",
"texts": [
" hello ",
" world "
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [],
"popen": [],
"terminated": 0,
"waited": 0,
"runs": [],
"temp_calls": [],
"wav_exists_after": {},
"whisper": null
}
},
{
"op": "push2talk",
"case": "p2t-release-other-key-keeps-recording",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"release": "f9"
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 10,
"transcriber": {
"kind": "texts",
"texts": [
" hello ",
" world "
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [
"■ recording..."
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": null
}
},
{
"op": "push2talk",
"case": "p2t-wav-deleted-before-release",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"delete_wav": true
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 4001,
"transcriber": {
"kind": "texts",
"texts": [
" hello ",
" world "
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [
"■ recording..."
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": null
}
},
{
"op": "push2talk",
"case": "p2t-transcribe-raise",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 4001,
"transcriber": {
"kind": "raise",
"message": "boom"
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [
"■ recording...",
"■ transcribing...",
"transcribe error: boom"
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": {
"model": {
"size": "small",
"device": "auto",
"compute_type": "auto"
},
"transcribe": {
"wav": "push2talk-000.wav",
"language": "en"
}
}
}
},
{
"op": "push2talk",
"case": "p2t-transcribe-empty-join",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 4001,
"transcriber": {
"kind": "texts",
"texts": [
" ",
"",
"\t"
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [
"■ recording...",
"■ transcribing..."
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": {
"model": {
"size": "small",
"device": "auto",
"compute_type": "auto"
},
"transcribe": {
"wav": "push2talk-000.wav",
"language": "en"
}
}
}
},
{
"op": "push2talk",
"case": "p2t-xdotool-check-true-fails",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 4001,
"transcriber": {
"kind": "texts",
"texts": [
" hello ",
" world "
]
},
"xdotool_fail": 1
},
"want": {
"stderr_lines": [
"■ recording...",
"■ transcribing...",
"transcribe error: Command '['xdotool', 'type', '--delay', '5', 'hello world']' returned non-zero exit status 1."
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [
{
"argv": [
"xdotool",
"type",
"--delay",
"5",
"hello world"
],
"kwargs": {
"check": true
}
}
],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": {
"model": {
"size": "small",
"device": "auto",
"compute_type": "auto"
},
"transcribe": {
"wav": "push2talk-000.wav",
"language": "en"
}
}
}
},
{
"op": "push2talk",
"case": "p2t-unicode-text",
"input": {
"steps": [
{
"press": "ctrl_r"
},
{
"release": "ctrl_r"
}
],
"wav_bytes": 4001,
"transcriber": {
"kind": "texts",
"texts": [
" Grüße ",
"世界"
]
},
"xdotool_fail": 0
},
"want": {
"stderr_lines": [
"■ recording...",
"■ transcribing..."
],
"popen": [
{
"argv": [
"arecord",
"-f",
"S16_LE",
"-r",
"16000",
"-c",
"1",
"TMP/push2talk-000.wav"
],
"kwargs": {}
}
],
"terminated": 1,
"waited": 1,
"runs": [
{
"argv": [
"xdotool",
"type",
"--delay",
"5",
"Grüße 世界"
],
"kwargs": {
"check": true
}
}
],
"temp_calls": [
{
"suffix": ".wav",
"prefix": "push2talk-",
"delete": false,
"name": "TMP/push2talk-000.wav"
}
],
"wav_exists_after": {
"TMP/push2talk-000.wav": false
},
"whisper": {
"model": {
"size": "small",
"device": "auto",
"compute_type": "auto"
},
"transcribe": {
"wav": "push2talk-000.wav",
"language": "en"
}
}
}
}
]
}
+1
View File
@@ -0,0 +1 @@
rc=1
+1
View File
@@ -0,0 +1 @@
rc=1
+2
View File
@@ -0,0 +1,2 @@
BYTE-IDENTICAL
replay-stability rc=0
+1
View File
@@ -0,0 +1 @@
rc=0
+1
View File
@@ -0,0 +1 @@
rc=0
+1
View File
@@ -0,0 +1 @@
rc=0
+1
View File
@@ -0,0 +1 @@
rc=0
+1
View File
@@ -0,0 +1 @@
pin rc=0
+1
View File
@@ -0,0 +1 @@
live rc=0
+57
View File
@@ -0,0 +1,57 @@
# voicebox — literal-100% residue verification summary (rune-wave-f2)
Module: voicebox (GO). Oracle: src/rune/voicebox.py (164L, 11 fns, was 5/11).
Residue ported name-preserving (exact snake names, tool_protocol gaps.go law):
_run, start_recording, stop_recording, _transcribe_and_type, on_press,
on_release — the talk() nested closures, pure control logic 1:1, device IO
(arecord Popen, faster-whisper, xdotool, NamedTemporaryFile, unlink, stderr)
behind the voiceboxSeams injectable; the pynput listener loop stays facade.
## Changes
- src/go/voicebox/gaps.go — the six ports + seams + recording twin.
- src/go/voicebox/main.go — replay loop extracted (runReplay/loadCases);
new ops: run, push2talk, talk_banner.
- src/go/voicebox/battery_gaps.go — fake-seam harness mirroring the
generator rig; replays the frozen effect rows.
- src/go/voicebox/main_test.go — TestTruthReplay (VB_TRUTH_PATH override),
TestRunArgvLaw, TestPush2TalkGuardLaws.
- src/go/voicebox/truth.json — REGENERATED generator-owned (21 cases: 7
legacy incl. talk_banner, 3 _run, 11 push2talk).
- scripts/gen_voicebox_truth.py — freeze/replay generator (pynput/tempfile/
subprocess/faster_whisper rig; real temp files; TMP/ + REPO_ROOT
normalization; venv re-exec guard).
## Verification (all in this checkout)
- Freeze: frozen 21 cases; regen x2 BYTE-IDENTICAL; --replay mode 21/21
stable (verification/vb-freeze-verify.txt, vb-freeze.log,
vb-replay-stability.log). Shared-harness --verify N/A: op-format rig
truth (regen-stability + live battery are the freeze gates, project_
registry precedent).
- Go battery: TestTruthReplay/TestRunArgvLaw/TestPush2TalkGuardLaws PASS
(vb-gotest.log rc=0); gofmt clean (vb-gofmt), go vet clean (vb-govet),
go build clean (vb-gobuild); binary replay: PARITY OK 21 cases.
- Live parity 100%: fresh oracle regen → /tmp/vb_live_truth.json (byte-
identical to frozen) → battery green against it (vb-parity.log).
Divergence found+fixed en route: Go calledProcessError dropped the
quotes around the python list repr — battery caught it (case 19), fixed
to "Command '[…]' returned non-zero exit status N.".
- Python both pins: adjacent pytest has no voicebox importer (plugin
surface — task_gate precedent); gates = canary + battery. Canary
verbose: ✓ Import voicebox BOTH pins (vb-canary-{live,pin}.log; the 3
remaining canary failures are sibling .so contracts — witness/
claim_audit cognition_core+evidence_provenance — identical both pins,
unrelated). Adjacent suite test_p1_03_memory_selection.py (exercises
warm_neuron_store via the selector): 67 passed/1 skipped BOTH pins
(vb-pytest-adjacent{,-pin}.log) after building memory_scoring +
memory_retrieval .so in this worktree (retired-oracle stubs recurse
without them — worktree artifact, not a regression; main checkout
passes with its prebuilt .so's).
- Audit delta: voicebox 5/11 → 11/11 (missing list empty).
## Divergences (documented, not bugs)
- handleHook empty-event validation improvement (pre-existing, pinned by
hook_empty_oracle).
- stop_recording stats the wav ONCE (fileState seam); the oracle's
exists-then-getsize TOCTOU window is unreachable single-threaded.
- on_press/on_release suppression is recover()-based (oracle: except →
logger.debug, silent unconfigured).