test_brain_bridge.py
Rama main · réplica de lectura del repositorio original.
#!/usr/bin/env python3
"""Test the bridge brain backend against the sim persona.
No bot-spren CLI, no Claude session, no mic, and no model loads — pure file I/O
and threads, so it is safe to run on the RAM-fragile Pi without a cgroup cap.
Proves brain_via_bridge() sends to an inbox and reads the correct reply back,
and that a second turn does not return the first turn's stale reply.
"""
from __future__ import annotations
import json
import logging
import os
import subprocess
import sys
import tempfile
import time
from pathlib import Path
import brain_bridge
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 main() -> int:
d = Path(tempfile.mkdtemp(prefix="voice-bridge-"))
inbox = d / "manual-inbox.jsonl"
reply = d / "voice-reply.txt"
# One persistent persona for the whole test — this mirrors production, where a
# single Syl session checkpoints its inbox offset and never re-replies to an
# already-consumed message. A fresh SimPersona per turn would reset its offset
# to 0, re-read the first question, and emit a duplicate reply that races the
# next turn's poll (a test-harness artifact, not a brain_via_bridge bug).
sim = SimPersona(inbox, reply, poll_s=0.05)
sim.start()
send = brain_bridge.local_sim_send(inbox)
read = brain_bridge.file_reply_reader(reply)
# One cursor for the whole conversation, as production keeps per reply file.
cursor = brain_bridge.ReplyCursor()
try:
out1 = brain_bridge.brain_via_bridge(
"hey jarvis, what is two plus two?",
persona="syl",
send=send,
read_reply=read,
cursor=cursor,
timeout_s=10,
poll_s=0.05,
)
check(out1 == "Two plus two is four.", f"round trip reply: {out1!r}")
# A second turn must return the new reply, not the first turn's stale block.
out2 = brain_bridge.brain_via_bridge(
"what is the capital of france?",
persona="syl",
send=send,
read_reply=read,
cursor=cursor,
timeout_s=10,
poll_s=0.05,
)
check(
out2 == "The capital of France is Paris.",
f"second turn is fresh, not stale: {out2!r}",
)
finally:
sim.stop()
# Timeout path: no sim running, so no reply ever appears.
inbox2 = d / "dead-inbox.jsonl"
reply2 = d / "dead-reply.txt"
out3 = brain_bridge.brain_via_bridge(
"is anyone there?",
persona="syl",
send=brain_bridge.local_sim_send(inbox2),
read_reply=brain_bridge.file_reply_reader(reply2),
timeout_s=1,
poll_s=0.05,
)
check(
out3.startswith("Sorry, the brain took too long"),
f"timeout returns a spoken error: {out3!r}",
)
# Regression (the one-behind audio bug): a late reply from a timed-out turn must
# not be spoken on the next turn. The cursor reserves a slot per send, so the
# timed-out turn's late block fills its own reserved slot and this turn skips
# past it. The old "next new block" logic spoke that late block — the previous
# prompt's answer — one turn behind.
def _blocks_text(*pairs: tuple[str, str]) -> str:
return "".join(
f"\n--- 2026-01-01T00:00:00+00:00 delivery_id={i} ---\n{p}\n"
for i, p in pairs
)
# State after a good turn A (slot 0, on disk) and a timed-out turn B whose reply
# is still in flight: the cursor has reserved 2 slots. During this turn (C),
# B's late reply lands first (slot 1), then C's real reply (slot 2).
cursor3 = brain_bridge.ReplyCursor(consumed=2)
frames = [
_blocks_text(("a", "A answer")), # at send
_blocks_text(("a", "A answer"), ("b", "late B answer")), # B lands
_blocks_text(("a", "A answer"), ("b", "late B answer"), ("c", "C answer")),
]
reads = {"i": 0}
def _scripted_read() -> str:
text = frames[min(reads["i"], len(frames) - 1)]
reads["i"] += 1
return text
out4 = brain_bridge.brain_via_bridge(
"turn C prompt",
persona="syl",
send=lambda *a, **k: None,
read_reply=_scripted_read,
cursor=cursor3,
timeout_s=5,
poll_s=0.01,
)
check(
out4 == "C answer",
f"late reply from a timed-out turn is skipped, not spoken: {out4!r}",
)
check(
cursor3.consumed == 3,
f"the send reserved this turn's slot: consumed={cursor3.consumed}",
)
# Self-heal (the dropped-reply wedge): if a reply is never written at all — the
# session crashed or its delivery hook failed — the reserved slot can never fill,
# so without recovery the cursor sits one ahead forever and every later turn times
# out. After _RESYNC_AFTER_MISSES consecutive timeouts with the cursor ahead of the
# file, it resyncs to the live end so the next turn returns its own answer. A single
# late reply never triggers this: the following turn reads its slot and resets the
# miss count.
class _DropThenAnswer:
"""First send's reply is dropped (never written); each later send writes one
block immediately, one slot behind the reservations — the wedge signature."""
def __init__(self) -> None:
self.blocks: list[str] = []
self.first = True
def send(
self, _persona: str, prompt: str, *, event_id: str | None = None
) -> None:
if self.first:
self.first = False
return # dropped: no block ever appears for this turn
self.blocks.append(prompt)
def read(self) -> str:
return "".join(
f"\n--- t delivery_id={i} ---\n{p}\n" for i, p in enumerate(self.blocks)
)
brain = _DropThenAnswer()
heal = brain_bridge.ReplyCursor(consumed=0)
turns = [
brain_bridge.brain_via_bridge(
f"q{n}",
persona="syl",
send=brain.send,
read_reply=brain.read,
cursor=heal,
timeout_s=0.3,
poll_s=0.02,
)
for n in range(1, 4)
]
check(
turns[0].startswith("Sorry, the brain took too long"),
f"dropped reply times out (turn 1): {turns[0]!r}",
)
check(
turns[1].startswith("Sorry, the brain took too long"),
f"still wedged before the resync threshold (turn 2): {turns[1]!r}",
)
check(
turns[2] == "q3",
f"cursor resynced; turn 3 speaks its own answer, not a wedge: {turns[2]!r}",
)
# The send argv must carry -d <working_dir> so bot-spren writes to the session's
# inbox, not its default ~/.bot-spren/<name>/state dead-letter dir. Capture the
# command without running anything.
captured: list[list[str]] = []
class _FakeProc:
returncode = 0
def _fake_run(cmd, *a, **k):
captured.append(cmd)
return _FakeProc()
real_run = brain_bridge.subprocess.run
brain_bridge.subprocess.run = _fake_run
try:
brain_bridge.cli_send("bot-spren", working_dir="/x/syl")(
"syl", "hi", event_id="turn-123"
)
check(
captured[-1]
== [
"bot-spren",
"send",
"-d",
"/x/syl",
"--event-id",
"turn-123",
"syl",
"hi",
],
f"cli_send forwards the event id with -d: {captured[-1]}",
)
brain_bridge.cli_send("bot-spren")("syl", "hi")
check(
captured[-1] == ["bot-spren", "send", "syl", "hi"],
f"cli_send without workdir omits -d: {captured[-1]}",
)
brain_bridge.ssh_cli_send("ofj", "bot-spren", working_dir="/x/syl")(
"syl", "hi", event_id="turn-456"
)
remote = captured[-1][-1] # the joined remote command is the last ssh arg
check(
"send -d /x/syl --event-id turn-456 syl hi" in remote,
f"ssh_cli_send forwards the event id in the remote command: {remote!r}",
)
# Staged rollout: officejawn can still have a bot-spren CLI from before
# --event-id existed. Click rejects an unknown option before the command
# handler writes an inbox event, so only that exact error permits one safe
# positional-mode retry. Exercise the local and SSH command shapes.
attempts: list[list[str]] = []
def _old_cli_run(cmd, *a, **kw):
attempts.append(cmd)
if any("--event-id" in part for part in cmd):
raise brain_bridge.subprocess.CalledProcessError(
2,
cmd,
stderr="Error: No such option: --event-id\n",
)
return _FakeProc()
brain_bridge.subprocess.run = _old_cli_run
local_stamped = brain_bridge.cli_send("bot-spren", working_dir="/x/syl")(
"syl", "hi", event_id="turn-789"
)
check(
local_stamped is False,
"the legacy local retry reports that the event id was dropped",
)
check(
len(attempts) == 2
and "--event-id" in attempts[0]
and "--event-id" not in attempts[1],
f"old local CLI gets one flag-free retry: {attempts!r}",
)
attempts.clear()
remote_stamped = brain_bridge.ssh_cli_send(
"ofj", "bot-spren", working_dir="/x/syl"
)("syl", "hi", event_id="turn-987")
check(
remote_stamped is False,
"the legacy remote retry reports that the event id was dropped",
)
check(
len(attempts) == 2
and "--event-id" in attempts[0][-1]
and "--event-id" not in attempts[1][-1],
f"old remote CLI gets one flag-free retry: {attempts!r}",
)
# A transport or send failure may have an unknown outcome. It must surface
# after one attempt rather than risk writing the same prompt twice.
attempts.clear()
def _failed_send(cmd, *a, **kw):
attempts.append(cmd)
raise brain_bridge.subprocess.CalledProcessError(
1, cmd, stderr="connection closed"
)
brain_bridge.subprocess.run = _failed_send
failed_loud = False
try:
brain_bridge.cli_send("bot-spren", working_dir="/x/syl")(
"syl", "hi", event_id="turn-fail"
)
except brain_bridge.subprocess.CalledProcessError:
failed_loud = True
check(
failed_loud and len(attempts) == 1,
f"non-compatibility failures are not retried: {attempts!r}",
)
finally:
brain_bridge.subprocess.run = real_run
# --- #19 correlation key: identity match, positional fallback -------------
# When a reply echoes the request's event_id the bridge matches by identity, so a
# dropped or out-of-order reply on one turn cannot shift another turn's answer --
# the collateral damage #48 hits under pure positional correlation.
answers = {
"what is two plus two?": "Two plus two is four.",
"what is the capital of france?": "The capital of France is Paris.",
}
class _StampedSyl:
"""Stamps each reply with the request's event_id (models the reply producer).
drop = 1-based turn numbers whose reply is never written (models #48)."""
def __init__(self, drop=(), plain=()) -> None:
self.blocks: list[tuple[str, str | None, str]] = []
self.turn = 0
self.drop = set(drop)
self.plain = set(plain) # turns written WITHOUT an event_id (positional)
def send(
self, _persona: str, prompt: str, *, event_id: str | None = None
) -> None:
self.turn += 1
text = prompt.split("User: ", 1)[-1] if "User: " in prompt else prompt
if self.turn in self.drop:
return # reply dropped at source (#48)
eid = (
None if self.turn in self.plain else event_id
) # unstamped -> positional
self.blocks.append(
(
f"d{self.turn}",
eid,
answers.get(text.strip(), f"echo:{text.strip()}"),
)
)
def read(self) -> str:
out = ""
for did, eid, payload in self.blocks:
hdr = f"--- t delivery_id={did}"
if eid is not None:
hdr += f" event_id={eid}"
out += f"\n{hdr} ---\n{payload}\n"
return out
# The parser tolerates the optional event_id token and extracts it; an unstamped
# header still parses with event_id None. Against the pre-#19 regex, the stamped
# block fails to match at all.
parsed = brain_bridge._delivery_blocks(
"\n--- t delivery_id=d1 event_id=e1 ---\nstamped\n"
"\n--- t delivery_id=d2 ---\nunstamped\n"
)
check(
parsed == [("d1", "e1", "stamped"), ("d2", None, "unstamped")],
f"_delivery_blocks parses the optional event_id token: {parsed}",
)
# Identity round trip: two turns, each returns its own stamped answer.
syl = _StampedSyl()
common = dict(
persona="syl",
send=syl.send,
read_reply=syl.read,
cursor=brain_bridge.ReplyCursor(),
timeout_s=0.5,
poll_s=0.01,
)
r1 = brain_bridge.brain_via_bridge("what is two plus two?", **common)
r2 = brain_bridge.brain_via_bridge("what is the capital of france?", **common)
check(r1 == answers["what is two plus two?"], f"identity turn 1: {r1!r}")
check(r2 == answers["what is the capital of france?"], f"identity turn 2: {r2!r}")
# The #48 fix: a dropped reply costs ONLY its own turn. Turn 1's reply is dropped;
# turn 2 still returns its own answer by identity. Under pure positional
# correlation turn 2 would time out too (its block sits at slot 0 while the cursor
# polls slot 1) -- the collateral damage this closes.
syl2 = _StampedSyl(drop={1})
common2 = dict(
persona="syl",
send=syl2.send,
read_reply=syl2.read,
cursor=brain_bridge.ReplyCursor(),
timeout_s=0.3,
poll_s=0.02,
)
d1 = brain_bridge.brain_via_bridge("what is two plus two?", **common2)
d2 = brain_bridge.brain_via_bridge("what is the capital of france?", **common2)
check(
d1.startswith("Sorry, the brain took too long"),
f"dropped reply still times out its own turn: {d1!r}",
)
check(
d2 == answers["what is the capital of france?"],
f"next turn keeps its own answer despite the drop (#48 fixed): {d2!r}",
)
# Mixed/transition seam (the hardest case): a stamped block that is NOT this turn's
# -- a different event_id -- sitting at the positional target slot must be SKIPPED
# by the positional fallback, never consumed. Otherwise a dropped stamped reply
# would just relocate the #48 theft to the next turn. event_id is an internal uuid,
# so the literal eids below never match this turn's id.
seam_file = (
"\n--- t delivery_id=d1 event_id=eid-one ---\nanswer one\n"
"\n--- t delivery_id=dx event_id=eid-x ---\nanswer x\n"
)
seam = brain_bridge.brain_via_bridge(
"stamped reply dropped for this turn",
persona="syl",
send=lambda *a, **k: None,
read_reply=lambda: seam_file,
cursor=brain_bridge.ReplyCursor(consumed=1),
timeout_s=0.1,
poll_s=0.02,
)
check(
seam.startswith("Sorry, the brain took too long"),
f"non-matching stamped block at the target slot is skipped: {seam!r}",
)
# Cursor coherence across mixed resolution: an unstamped turn (positional) then a
# stamped turn (identity) through one cursor each return their own answer, and the
# cursor still counts one slot per send.
mix = _StampedSyl(plain={1})
mcur = brain_bridge.ReplyCursor()
mcommon = dict(
persona="syl",
send=mix.send,
read_reply=mix.read,
cursor=mcur,
timeout_s=0.5,
poll_s=0.01,
)
m1 = brain_bridge.brain_via_bridge("what is two plus two?", **mcommon)
m2 = brain_bridge.brain_via_bridge("what is the capital of france?", **mcommon)
check(
m1 == answers["what is two plus two?"],
f"mixed: unstamped turn resolves positionally: {m1!r}",
)
check(
m2 == answers["what is the capital of france?"],
f"mixed: stamped turn resolves by identity: {m2!r}",
)
check(mcur.consumed == 2, f"mixed: cursor counts both sends: {mcur.consumed}")
# Transition gap (documented, tracked in #59): after a stamped identity match, a
# later UNSTAMPED turn that falls back to position cannot read its own slot, because
# the stamped match left the positional cursor ahead of the file and the consumer
# cannot safely realign it — positional correlation can't tell a dropped reserved
# slot from a late one, so rewinding could speak a late reply as the next answer. The
# full producer deployment (#59) closes this by stamping every reply, removing the
# unstamped fallback. Sequence: stamped turn dropped, stamped turn answered at slot
# 0, then an unstamped turn answered at slot 1.
trans = _StampedSyl(drop={1}, plain={3})
tcur = brain_bridge.ReplyCursor()
tcommon = dict(
persona="syl",
send=trans.send,
read_reply=trans.read,
cursor=tcur,
timeout_s=0.3,
poll_s=0.02,
)
t1 = brain_bridge.brain_via_bridge("what is two plus two?", **tcommon)
t2 = brain_bridge.brain_via_bridge("what is the capital of france?", **tcommon)
t3 = brain_bridge.brain_via_bridge("what is two plus two?", **tcommon)
check(
t1.startswith("Sorry, the brain took too long"),
f"transition: dropped stamped turn times out: {t1!r}",
)
check(
t2 == answers["what is the capital of france?"],
f"transition: stamped turn matches by identity regardless of position: {t2!r}",
)
check(
t3.startswith("Sorry, the brain took too long"),
f"transition: mixed-window unstamped turn after a stamped match strands (deferred to #59): {t3!r}",
)
# End-to-end identity match through the real components (model-free): the actual
# local_sim_send stamps the inbox event_id, a real SimPersona echoes it into the
# reply block, and the bridge identity-matches -- proving the wire composes, not
# just the in-test stub.
e2e_inbox = d / "e2e-inbox.jsonl"
e2e_reply = d / "e2e-reply.txt"
e2e_sim = SimPersona(e2e_inbox, e2e_reply, poll_s=0.05)
e2e_sim.start()
try:
ecommon = dict(
persona="syl",
send=brain_bridge.local_sim_send(e2e_inbox),
read_reply=brain_bridge.file_reply_reader(e2e_reply),
cursor=brain_bridge.ReplyCursor(),
timeout_s=10,
poll_s=0.05,
)
e1 = brain_bridge.brain_via_bridge("what is two plus two?", **ecommon)
e2 = brain_bridge.brain_via_bridge("what is the capital of france?", **ecommon)
check(e1 == "Two plus two is four.", f"e2e identity turn 1: {e1!r}")
check(e2 == "The capital of France is Paris.", f"e2e identity turn 2: {e2!r}")
finally:
e2e_sim.stop()
# Late-reply regression through the real simulator: two short timeouts put the
# positional cursor ahead of the reply file. The third turn resyncs while the
# first answer is still delayed. Positional matching would take that stale answer
# from slot 0; identity matching waits for the third turn's stamped answer.
late_inbox = d / "late-inbox.jsonl"
late_reply = d / "late-reply.txt"
late_calls = {"count": 0}
late_answers = {
"first question": "late answer one",
"second question": "late answer two",
"third question": "current answer three",
}
def _delayed_first_reply(text: str) -> str:
late_calls["count"] += 1
if late_calls["count"] == 1:
time.sleep(0.3)
return late_answers[text.strip()]
late_sim = SimPersona(
late_inbox, late_reply, reply_fn=_delayed_first_reply, poll_s=0.01
)
late_sim.start()
try:
late_common = dict(
persona="syl",
send=brain_bridge.local_sim_send(late_inbox),
read_reply=brain_bridge.file_reply_reader(late_reply),
cursor=brain_bridge.ReplyCursor(),
poll_s=0.01,
)
late_first = brain_bridge.brain_via_bridge(
"first question", timeout_s=0.05, **late_common
)
late_second = brain_bridge.brain_via_bridge(
"second question", timeout_s=0.05, **late_common
)
late_third = brain_bridge.brain_via_bridge(
"third question", timeout_s=2, **late_common
)
check(
late_first.startswith("Sorry, the brain took too long"),
f"late first reply times out its own turn: {late_first!r}",
)
check(
late_second.startswith("Sorry, the brain took too long"),
f"second timeout triggers positional resync: {late_second!r}",
)
check(
late_third == "current answer three",
f"resynced turn skips stale replies by identity: {late_third!r}",
)
inbox_ids = [
json.loads(line)["event_id"]
for line in late_inbox.read_text(encoding="utf-8").splitlines()
]
reply_ids = [
event_id
for _delivery_id, event_id, _payload in brain_bridge._delivery_blocks(
late_reply.read_text(encoding="utf-8")
)
]
check(
reply_ids == inbox_ids,
"simulator reply headers preserve each inbox event id",
)
finally:
late_sim.stop()
# --- #44 landing check: a non-landing send fails loudly, not as a slow brain ---
# bot-spren send can exit 0 yet write to an inbox the session never reads, so the
# turn used to look identical to a slow brain: a full timeout with no diagnostic.
# With a confirm_landing probe on the inbox the session consumes, a send that does
# not land there is caught fast and reported distinctly, while a landed-but-slow
# turn still falls through to the ordinary timeout.
corrupt_inbox = d / "land-corrupt-inbox.jsonl"
corrupt_inbox.write_bytes(b"\xff\n")
check(
brain_bridge.file_inbox_probe(corrupt_inbox)("any-id") is None,
"a non-UTF-8 inbox makes the local landing probe unobservable, not fatal",
)
# #98: the probe matches this turn's exact event_id, not a line count, so another
# producer's event or a malformed line in the same inbox cannot fake a landing.
local_probe_inbox = d / "land-local-probe.jsonl"
local_probe = brain_bridge.file_inbox_probe(local_probe_inbox)
check(
local_probe("id-a") is False,
"a missing local inbox is a definite non-landing (False)",
)
local_probe_inbox.write_text(
'not json\n[1, 2]\n{"event_id": "id-other"}\n', encoding="utf-8"
)
check(
local_probe("id-a") is False,
"other events and malformed lines do not count as this turn's landing",
)
with local_probe_inbox.open("a", encoding="utf-8") as f:
f.write(json.dumps({"event_id": "id-a", "payload": "hi"}) + "\n")
check(local_probe("id-a") is True, "the exact event_id in the inbox is a landing")
# The ssh probe runs one remote command per lookup and maps its answer; any
# transport failure is unobservable (None), never a false non-landing.
real_run = brain_bridge.subprocess.run
ssh_calls: list[list[str]] = []
def fake_ssh(result):
def _run(argv, **_kwargs):
ssh_calls.append(argv)
if isinstance(result, BaseException):
raise result
code, out = result
return subprocess.CompletedProcess(argv, code, stdout=out, stderr="")
return _run
ssh_probe = brain_bridge.ssh_inbox_probe("persona-host", "/srv/in box.jsonl")
try:
for result, want, label in [
((0, "FOUND\n"), True, "FOUND is a landing"),
((0, "ABSENT\n"), False, "ABSENT is a non-landing"),
((0, "MISSING\n"), False, "a missing remote inbox is a non-landing"),
((0, "7\n"), None, "unexpected output is unobservable"),
((255, ""), None, "an ssh failure is unobservable"),
(subprocess.TimeoutExpired("ssh", 40), None, "a down host is unobservable"),
]:
brain_bridge.subprocess.run = fake_ssh(result)
check(ssh_probe("id-ssh") is want, f"ssh probe: {label}")
finally:
brain_bridge.subprocess.run = real_run
remote_cmd = ssh_calls[0][-1]
check(
ssh_calls[0][:4] == ["ssh", "-o", "ConnectTimeout=15", "persona-host"]
and "'/srv/in box.jsonl'" in remote_cmd,
f"ssh probe quotes the inbox path: {remote_cmd!r}",
)
# A malformed path (say, a list from config.local.json) must fail when the probe
# is built, before any send, not mid-turn after the prompt went out.
try:
brain_bridge.ssh_inbox_probe("persona-host", ["/srv/inbox.jsonl"])
bad_path_fails_early = False
except TypeError:
bad_path_fails_early = True
check(bad_path_fails_early, "a non-string ssh inbox path fails at construction")
# Run the real remote command through a local shell in place of ssh, so the
# grep pattern, its exit-status mapping, and the quoting are exercised for real.
def shell_ssh(argv, **kwargs):
return real_run(["sh", "-c", argv[-1]], **kwargs)
sh_inbox = d / "ssh inbox.jsonl"
sh_probe = brain_bridge.ssh_inbox_probe("persona-host", str(sh_inbox))
brain_bridge.subprocess.run = shell_ssh
try:
check(sh_probe("abc-123") is False, "sh: a missing remote inbox is False")
sh_inbox.write_text(
json.dumps({"payload": "quoting abc-123", "event_id": "other"}) + "\n",
encoding="utf-8",
)
check(
sh_probe("abc-123") is False,
"sh: the id quoted in another field of another event is not a landing",
)
with sh_inbox.open("a", encoding="utf-8") as f:
f.write(json.dumps({"event_id": "abc-123"}) + "\n")
check(sh_probe("abc-123") is True, "sh: the exact event_id key is a landing")
check(
sh_probe("x'; echo FOUND; '") is None,
"sh: an id outside the uuid alphabet is unobservable, never run",
)
sh_inbox.chmod(0)
if not os.access(sh_inbox, os.R_OK): # root reads through mode 000
check(
sh_probe("abc-123") is None,
"sh: an unreadable remote inbox is unobservable, not absent",
)
sh_inbox.chmod(0o600)
finally:
brain_bridge.subprocess.run = real_run
# A down host costs one transport timeout: the send fails first, and the landing
# probe never runs, before or after it (#98 removed the pre-send probe).
probe_calls: list[str] = []
def failing_send(persona, prompt, *, event_id=None):
raise subprocess.TimeoutExpired("ssh", 40)
out_down = brain_bridge.brain_via_bridge(
"anyone there?",
persona="syl",
send=failing_send,
read_reply=lambda: "",
confirm_landing=lambda eid: probe_calls.append(eid) or True,
timeout_s=30,
)
check(
out_down.startswith("Sorry, I couldn't reach the brain (TimeoutExpired)"),
f"a down host returns the spoken send error: {out_down!r}",
)
check(probe_calls == [], f"a failed send runs no landing probe: {probe_calls!r}")
# A legacy send (older bot-spren, no --event-id) wrote bot-spren's own id, so a
# lookup of ours would be a false non-landing: the check is skipped and the turn
# reaches the ordinary reply wait instead.
legacy_probe_calls: list[str] = []
out_legacy = brain_bridge.brain_via_bridge(
"legacy send",
persona="syl",
send=lambda persona, prompt, *, event_id=None: False,
read_reply=lambda: "",
confirm_landing=lambda eid: legacy_probe_calls.append(eid) or False,
timeout_s=0.05,
landing_timeout_s=0.05,
poll_s=0.01,
)
check(
out_legacy.startswith("Sorry, the brain took too long")
and legacy_probe_calls == [],
f"a legacy send skips the landing check: {out_legacy!r}",
)
# The probe runs only after the send, with the event_id the send carried.
order: list[tuple[str, str]] = []
def recording_send(persona, prompt, *, event_id=None):
order.append(("send", event_id))
brain_bridge.brain_via_bridge(
"order check",
persona="syl",
send=recording_send,
read_reply=lambda: "",
confirm_landing=lambda eid: order.append(("probe", eid)) or True,
timeout_s=0.05,
poll_s=0.01,
)
check(
[step for step, _ in order] == ["send", "probe"] and order[0][1] == order[1][1],
f"one post-send probe for the sent event_id, none before it: {order!r}",
)
log_records: list[logging.LogRecord] = []
class _ListHandler(logging.Handler):
def emit(self, record: logging.LogRecord) -> None:
log_records.append(record)
handler = _ListHandler()
brain_bridge.logger.addHandler(handler)
brain_bridge.logger.setLevel(logging.ERROR)
try:
# Non-landing: the send writes to inbox A, but the probe watches inbox B (the
# inbox the session actually reads), which never gets this turn's event.
sent_inbox = d / "land-sent.jsonl"
# The watched inbox already holds another producer's event and gets no new
# one from this send; a line count could not tell those apart (#98).
watched_inbox = d / "land-watched.jsonl"
watched_inbox.write_text(
json.dumps({"event_id": "someone-else"}) + "\n", encoding="utf-8"
)
no_land_cursor = brain_bridge.ReplyCursor()
out_noland = brain_bridge.brain_via_bridge(
"did this land?",
persona="syl",
send=brain_bridge.local_sim_send(sent_inbox),
read_reply=brain_bridge.file_reply_reader(d / "land-missing-reply.txt"),
cursor=no_land_cursor,
confirm_landing=brain_bridge.file_inbox_probe(watched_inbox),
timeout_s=30, # would dominate if the landing check did not short-circuit
landing_timeout_s=0.2,
poll_s=0.02,
)
check(
out_noland.startswith("Sorry, I sent that but it never reached"),
f"non-landing send returns the distinct inbox error: {out_noland!r}",
)
check(
not out_noland.startswith("Sorry, the brain took too long"),
"non-landing error is distinct from the slow-brain timeout",
)
check(
no_land_cursor.consumed == 0,
f"a non-landing send reserves no reply slot: consumed={no_land_cursor.consumed}",
)
landed_logs = [r for r in log_records if "did not land" in r.getMessage()]
check(
len(landed_logs) == 1 and landed_logs[0].levelno == logging.ERROR,
f"non-landing send logs one distinct ERROR: {len(landed_logs)} record(s)",
)
finally:
brain_bridge.logger.removeHandler(handler)
# Landed-but-slow: the send reaches the watched inbox, so the landing check passes;
# with no reply ever written the turn still returns the ordinary slow-brain timeout,
# proving the check does not false-alarm on a genuinely slow brain.
slow_inbox = d / "slow-inbox.jsonl"
out_slow = brain_bridge.brain_via_bridge(
"landed but slow",
persona="syl",
send=brain_bridge.local_sim_send(slow_inbox),
read_reply=brain_bridge.file_reply_reader(d / "slow-missing-reply.txt"),
confirm_landing=brain_bridge.file_inbox_probe(slow_inbox),
timeout_s=1,
landing_timeout_s=1,
poll_s=0.02,
)
check(
out_slow.startswith("Sorry, the brain took too long"),
f"landed-but-slow send still times out normally: {out_slow!r}",
)
# An unobservable inbox (probe returns None) skips the check rather than
# false-alarming: a live sim answers and the reply comes back as usual.
unobs_inbox = d / "unobs-inbox.jsonl"
unobs_reply = d / "unobs-reply.txt"
unobs_sim = SimPersona(unobs_inbox, unobs_reply, poll_s=0.05)
unobs_sim.start()
try:
out_unobs = brain_bridge.brain_via_bridge(
"what is two plus two?",
persona="syl",
send=brain_bridge.local_sim_send(unobs_inbox),
read_reply=brain_bridge.file_reply_reader(unobs_reply),
cursor=brain_bridge.ReplyCursor(),
confirm_landing=lambda _eid: None, # inbox cannot be observed this turn
timeout_s=10,
poll_s=0.05,
)
check(
out_unobs == "Two plus two is four.",
f"an unobservable inbox skips the check, reply still returns: {out_unobs!r}",
)
finally:
unobs_sim.stop()
# Happy path with a real probe: the send lands in the watched inbox and the sim
# answers, so the landing check passes through to the normal reply.
land_inbox = d / "land-ok-inbox.jsonl"
land_reply = d / "land-ok-reply.txt"
land_sim = SimPersona(land_inbox, land_reply, poll_s=0.05)
land_sim.start()
try:
out_land = brain_bridge.brain_via_bridge(
"what is the capital of france?",
persona="syl",
send=brain_bridge.local_sim_send(land_inbox),
read_reply=brain_bridge.file_reply_reader(land_reply),
cursor=brain_bridge.ReplyCursor(),
confirm_landing=brain_bridge.file_inbox_probe(land_inbox),
timeout_s=10,
poll_s=0.05,
)
check(
out_land == "The capital of France is Paris.",
f"landed send with a real probe returns the reply: {out_land!r}",
)
finally:
land_sim.stop()
n_pass = sum(1 for ok, _ in results if ok)
print(f"\n=== {n_pass}/{len(results)} checks passed ===")
return 0 if n_pass == len(results) else 1
if __name__ == "__main__":
sys.exit(main())