test_brain_dispatch.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""brain() dispatches on the brain_backend config key.
test_pipeline_bridge.py proves the full bridge chain through build_brain(). This
proves the step before that: config alone selects the backend and transport, so a
real deployment enables the persistent-session brain by editing config.local.json,
with no code change. No models and no bot-spren CLI are needed — config selects the
sim transport and points its paths at temp files, so the repo's own config is
untouched.
"""
from __future__ import annotations
import json
import sys
import tempfile
from pathlib import Path
from types import SimpleNamespace
import brain_bridge
import pipeline
from sim_persona import SimPersona
results: list[tuple[bool, str]] = []
def check(ok: bool, detail: str) -> None:
results.append((ok, detail))
print(f" [{'PASS' if ok else 'FAIL'}] {detail}")
def _write(path: Path, cfg: dict) -> None:
path.write_text(json.dumps(cfg))
def test_routing(d: Path) -> None:
"""brain_backend picks the helper; cli is the default for a bare config."""
pipeline.CONFIG_PATH = d / "config.json"
pipeline.LOCAL_CONFIG_PATH = d / "config.local.json"
pipeline._brain_cli = lambda text, cfg, **kw: "CLI"
pipeline._brain_bridge = lambda text, cfg: "BRIDGE"
_write(pipeline.CONFIG_PATH, {"brain_backend": "cli"})
check(pipeline.brain("hi") == "CLI", "brain_backend=cli routes to the CLI path")
_write(pipeline.CONFIG_PATH, {"brain_backend": "bridge"})
check(
pipeline.brain("hi") == "BRIDGE",
"brain_backend=bridge routes to the bridge path",
)
# A config with no brain_backend key falls back to the DEFAULTS value (cli),
# so a fresh clone runs standalone.
_write(pipeline.CONFIG_PATH, {"wake_word": "hey_jarvis"})
check(pipeline.brain("hi") == "CLI", "missing brain_backend defaults to cli")
def test_bridge_via_config(d: Path) -> None:
"""With only config.local.json set to bridge, brain() answers via the sim."""
inbox = d / "manual-inbox.jsonl"
reply = d / "reply.txt"
# config.json stays minimal; config.local.json enables the bridge. This is the
# committed-base + gitignored-overlay split a real deployment uses.
pipeline.CONFIG_PATH = d / "config.json"
pipeline.LOCAL_CONFIG_PATH = d / "config.local.json"
_write(pipeline.CONFIG_PATH, {"wake_word": "hey_jarvis"})
_write(
pipeline.LOCAL_CONFIG_PATH,
{
"brain_backend": "bridge",
"brain_transport": "sim",
"brain_persona": "syl",
"brain_reply_path": str(reply),
"brain_inbox_path": str(inbox),
"brain_timeout_s": 5,
"brain_poll_s": 0.05,
},
)
sim = SimPersona(inbox, reply, poll_s=0.02)
sim.start()
try:
out = pipeline.brain("what is two plus two?")
finally:
sim.stop()
check(
out == "Two plus two is four.",
f"config-driven sim transport answered: {out!r}",
)
sent = json.loads(inbox.read_text().splitlines()[0])["payload"]
check(
sent == f"{pipeline.SYL_VOICE_SYSTEM_PROMPT}\n\nUser: what is two plus two?",
"the bridge sends Syl the voice dispatcher policy before the transcript",
)
def test_voice_dispatch_policy() -> None:
"""The voice contract pins both escalation paths and the non-blocking handoff."""
policy = pipeline.SYL_VOICE_SYSTEM_PROMPT.casefold()
required = {
"explicit escalation": "explicitly asks you to hand work off",
"implicit escalation": "needs tools, multiple steps, or more than a quick answer",
"ephemeral background worker": "ephemeral subagent in the background",
"immediate voice-loop return": "do not do delegated work inline or wait",
"mid-task correction": "follow-up corrections while the subagent runs",
"spoken result": "summarize each unreported result once",
"one bridge reply per turn": "never emit a second reply for the original turn",
"voice-event scope": "apply these instructions only to voice events",
"confirmation before external change": (
"do not start the subagent or act until joe confirms"
),
"spoken reply budget": "at most 500 characters",
"deferred overflow result": "leave other results unreported",
}
for behavior, instruction in required.items():
check(instruction in policy, f"voice policy includes {behavior}")
check(
pipeline._bridge_voice_system_prompt("syl") == pipeline.SYL_VOICE_SYSTEM_PROMPT,
"the Syl bridge persona receives the dispatcher policy",
)
check(
pipeline._bridge_voice_system_prompt("assistant")
== pipeline.VOICE_SYSTEM_PROMPT,
"a non-Syl bridge persona keeps the neutral voice policy",
)
check(
pipeline._bridge_voice_system_prompt(None) == pipeline.VOICE_SYSTEM_PROMPT,
"a null bridge persona falls back to the neutral voice policy",
)
check(
pipeline._bridge_voice_system_prompt(49) == pipeline.VOICE_SYSTEM_PROMPT,
"a non-string bridge persona falls back to the neutral voice policy",
)
def test_factory_validation() -> None:
"""Invalid deployment settings stay inside the voice error boundary."""
cases = (
({}, "reply path is not configured"),
(
{"brain_reply_path": "/tmp/reply", "brain_transport": "ssh"},
"host is not configured",
),
(
{"brain_reply_path": "/tmp/reply", "brain_transport": "sim"},
"inbox path is not configured",
),
(
{
"brain_reply_path": "/tmp/reply",
"brain_transport": "sim",
"brain_inbox_path": ["/tmp/inbox"],
},
"inbox path is not a filesystem path",
),
(
{"brain_reply_path": "/tmp/reply", "brain_transport": "carrier-pigeon"},
"is not supported",
),
(
{"brain_reply_path": "/tmp/reply", "brain_transport": []},
"is not supported",
),
(
{"brain_reply_path": "/tmp/reply", "brain_transport": {}},
"is not supported",
),
)
for cfg, expected in cases:
reply = brain_bridge.build_brain(cfg)("hello")
check(expected in reply, f"bad bridge config returns a spoken error: {reply!r}")
def test_cli_voice_prompt(real_brain_cli) -> None:
"""The tool-free CLI fallback must not promise a background handoff."""
captured = {}
real_run = pipeline.subprocess.run
def fake_run(*_args, **kwargs):
captured["prompt"] = kwargs["input"]
return SimpleNamespace(returncode=0, stdout="Done.\n", stderr="")
pipeline.subprocess.run = fake_run
try:
out = real_brain_cli("file an issue", dict(pipeline.DEFAULTS))
finally:
pipeline.subprocess.run = real_run
check(out == "Done.", f"CLI fallback still returns its reply: {out!r}")
check(
captured["prompt"]
== f"{pipeline.VOICE_SYSTEM_PROMPT}\n\nUser: file an issue\nAssistant:",
"CLI fallback uses the tool-free voice prompt",
)
check(
"subagent" not in captured["prompt"].casefold(),
"CLI fallback does not claim it can delegate",
)
def test_set_wake_word_no_leak(d: Path) -> None:
"""--set-wake-word writes only the base config, never the local overlay."""
pipeline.CONFIG_PATH = d / "config.json"
pipeline.LOCAL_CONFIG_PATH = d / "config.local.json"
_write(pipeline.CONFIG_PATH, {"wake_word": "alexa"})
_write(
pipeline.LOCAL_CONFIG_PATH,
{
"brain_backend": "bridge",
"brain_host": "secret-host",
"brain_reply_path": "/secret/path",
"brain_persona": "secret-persona",
},
)
pipeline.set_wake_word("hey_jarvis")
written = json.loads(pipeline.CONFIG_PATH.read_text())
local_keys = ("brain_backend", "brain_host", "brain_reply_path", "brain_persona")
leaked = [k for k in local_keys if k in written]
check(not leaked, f"config.json keeps the local overlay out (leaked: {leaked})")
check(written.get("wake_word") == "hey_jarvis", "config.json wake_word was updated")
check(
pipeline.load_config()["brain_backend"] == "bridge",
"effective config still reads the local overlay (runtime unaffected)",
)
def test_set_wake_word_local(d: Path) -> None:
"""--set-wake-word --local writes config.local.json, leaving the base intact."""
pipeline.CONFIG_PATH = d / "config.json"
pipeline.LOCAL_CONFIG_PATH = d / "config.local.json"
_write(pipeline.CONFIG_PATH, {"wake_word": "hey_jarvis"})
_write(
pipeline.LOCAL_CONFIG_PATH,
{"brain_backend": "bridge", "brain_host": "secret-host"},
)
pipeline.set_wake_word("alexa", local=True)
base = json.loads(pipeline.CONFIG_PATH.read_text())
overlay = json.loads(pipeline.LOCAL_CONFIG_PATH.read_text())
check(
base.get("wake_word") == "hey_jarvis",
"committed config.json default is left untouched by a local set",
)
check(
overlay.get("wake_word") == "alexa",
"config.local.json carries the local wake-word override",
)
check(
overlay.get("brain_host") == "secret-host",
"existing config.local.json keys survive the local set",
)
check(
pipeline.load_config()["wake_word"] == "alexa",
"effective config reads the local wake word (overlay wins)",
)
def test_set_wake_word_local_malformed_no_clobber(d: Path) -> None:
"""--set-wake-word --local refuses to overwrite a malformed config.local.json."""
pipeline.CONFIG_PATH = d / "config.json"
pipeline.LOCAL_CONFIG_PATH = d / "config.local.json"
_write(pipeline.CONFIG_PATH, {"wake_word": "hey_jarvis"})
# A truncated overlay an operator left mid-edit, still carrying live settings.
truncated = '{"brain_host": "secret-host",'
pipeline.LOCAL_CONFIG_PATH.write_text(truncated)
raised = False
try:
pipeline.set_wake_word("alexa", local=True)
except ValueError:
raised = True
check(raised, "a malformed config.local.json raises instead of clobbering")
check(
pipeline.LOCAL_CONFIG_PATH.read_text() == truncated,
"the malformed overlay is left intact for the operator to fix",
)
def test_set_wake_word_base_shadowed_by_local(d: Path) -> None:
"""A base set stays shadowed by a local override — the invariant the CLI note keys off."""
pipeline.CONFIG_PATH = d / "config.json"
pipeline.LOCAL_CONFIG_PATH = d / "config.local.json"
_write(pipeline.CONFIG_PATH, {})
_write(pipeline.LOCAL_CONFIG_PATH, {"wake_word": "alexa"})
cfg = pipeline.set_wake_word("hey_jarvis", local=False)
base = json.loads(pipeline.CONFIG_PATH.read_text())
check(base.get("wake_word") == "hey_jarvis", "the base write lands in config.json")
check(
cfg["wake_word"] == "alexa",
"effective wake word stays the local override, so the CLI flags it as shadowed",
)
def main() -> int:
# Snapshot module state so the test restores the repo's real wiring on exit.
saved = (
pipeline.CONFIG_PATH,
pipeline.LOCAL_CONFIG_PATH,
pipeline._brain_cli,
pipeline._brain_bridge,
)
try:
with tempfile.TemporaryDirectory(prefix="brain-dispatch-") as tmp:
d = Path(tmp)
test_routing(d)
test_voice_dispatch_policy()
test_factory_validation()
# Restore the real helpers before the live-ish bridge test.
pipeline._brain_cli, pipeline._brain_bridge = saved[2], saved[3]
test_cli_voice_prompt(saved[2])
test_bridge_via_config(d)
test_set_wake_word_no_leak(d)
test_set_wake_word_local(d)
test_set_wake_word_local_malformed_no_clobber(d)
test_set_wake_word_base_shadowed_by_local(d)
finally:
(
pipeline.CONFIG_PATH,
pipeline.LOCAL_CONFIG_PATH,
pipeline._brain_cli,
pipeline._brain_bridge,
) = saved
n_pass = sum(1 for ok, _ in results if ok)
print(f"=== {n_pass}/{len(results)} checks passed ===")
return 0 if n_pass == len(results) else 1
if __name__ == "__main__":
sys.exit(main())