#!/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())