diff --git a/tools/cc-bridge/README.md b/tools/cc-bridge/README.md index 7d777bf..7c0440c 100644 --- a/tools/cc-bridge/README.md +++ b/tools/cc-bridge/README.md @@ -197,6 +197,24 @@ the default window is set with `BUDDY_TOKEN_PERIOD` on the daemon. through to Claude's prompt. - **User doesn't press A or B** → 30 s timeout (configurable in `buddy_prompt.py`). Hook falls through. +- **Daemon up but delivering nothing** → check `~/.cache/claude-buddy/buddy.log` + for a `scanning for Claude* (5 s)…` line with *no outcome after it*. That was + a deadlocked reconnector (fixed; see `test_reconnect.py`). Also check for + `CBATTErrorDomain Code=15 "Encryption is insufficient"` — the NUS + characteristic needs a bond, so the link is up but unpaired. Re-pair, or use + serial, which needs no bond. + +## Tests + +`test_reconnect.py` covers the connect / reconnect / liveness paths offline — +no hardware and no BLE adapter, via a fake `bleak` and a throwaway `$HOME`. + +```bash +./test_reconnect.py # all scenarios + summary +./test_reconnect.py ble_hang # one scenario +``` + +The other `tools/test_*.py` scripts are hardware-in-the-loop; this one is not. ## Run as a launchd service (macOS) diff --git a/tools/cc-bridge/buddy_bridged.py b/tools/cc-bridge/buddy_bridged.py index 05983f0..0a8320b 100755 --- a/tools/cc-bridge/buddy_bridged.py +++ b/tools/cc-bridge/buddy_bridged.py @@ -47,6 +47,32 @@ HEARTBEAT_S = 10 DEFAULT_PROMPT_TIMEOUT_S = 30 +# CoreBluetooth's connectPeripheral: has no timeout by design -- it stays +# pending until the peripheral shows up -- and bleak inherits that. Left +# unbounded it parks the reconnector task forever (see _ble_connect). +BLE_CONNECT_TIMEOUT_S = 20 +# Backstop over the whole connect path, whatever the transport. Must exceed the +# worst-case sum of the inner timeouts (BLE: 5s discover + 20 + 20 + initial +# send) or it will abort attempts that were about to succeed. Env override +# exists so the tests can exercise the watchdog without a 90s wait. +def _watchdog_seconds() -> float: + """Parse defensively: this runs at import, so a typo'd env var would + otherwise raise before the daemon ever starts and leave KeepAlive + restarting it forever.""" + raw = os.environ.get("BUDDY_CONNECT_WATCHDOG_S", "").strip() + try: + value = float(raw) if raw else 90.0 + except ValueError: + value = 90.0 + return value if value > 0 else 90.0 + + +RECONNECT_WATCHDOG_S = _watchdog_seconds() +# A link that stops answering probes is dead even if the transport still claims +# otherwise; tolerate this many consecutive misses before tearing it down. +MAX_MISSED_PROBES = 3 +# Must stay below HEARTBEAT_S so one probe finishes before the next is due. +LIVENESS_PROBE_TIMEOUT_S = 5 # Evolution stage milestones (lifetime tokens → stage). All five stages are # reachable through normal leveling: Stages 1–4 every ~5 levels (250K) and @@ -105,6 +131,8 @@ def __init__(self) -> None: self._query_lock = asyncio.Lock() self._rx_buf = bytearray() self._owner = os.environ.get("BUDDY_OWNER", "CLI") + # Consecutive unanswered liveness probes; see _probe_liveness. + self._missed_probes = 0 self._serial_port = os.environ.get("BUDDY_SERIAL", "").strip() self._host = os.environ.get("BUDDY_HOST", "").strip() @@ -351,8 +379,18 @@ def device_name(self) -> Optional[str]: return self.device.name if self.device else None async def connect(self) -> bool: - if self.transport == "serial": - return await self._serial_connect() + if self._serial_port: + # BUDDY_SERIAL is a *preference*, not a pin: the USB node vanishes + # whenever the cable is pulled, and a hard pin turns that into a + # permanent crash-loop that looks identical to "device offline". + # Re-decide on every attempt so an unplugged cable degrades to BLE + # and a replugged one is picked back up on the next reconnect. + if os.path.exists(self._serial_port): + self.transport = "serial" + if await self._serial_connect(): + return True + self.transport = "ble" + return await self._ble_connect() if self.transport == "tcp": return await self._tcp_connect() if self.transport == "tcp-listen": @@ -360,6 +398,9 @@ async def connect(self) -> bool: return await self._ble_connect() async def _send_initial(self) -> None: + # Every transport funnels through here on a successful link, so it is + # the one place a fresh connection resets the probe counter. + self._missed_probes = 0 now = int(time.time()) tz = -time.timezone if time.daylight == 0 else -time.altzone await self._send_json({"time": [now, tz]}) @@ -471,10 +512,32 @@ async def _ble_connect(self) -> bool: self.device = cands[0] self.client = BleakClient(self.device.address) try: - await self.client.connect() - except Exception as exc: - log(f"BLE connect failed: {exc}"); return False - await self.client.start_notify(NUS_TX, lambda _c, d: self._feed(bytes(d))) + # Bounded: a peripheral that accepts the link but stalls the GATT + # handshake would otherwise park this coroutine for the life of the + # process, and the reconnector that awaits it is the only thing that + # can ever restore the link. + await asyncio.wait_for(self.client.connect(), + timeout=BLE_CONNECT_TIMEOUT_S) + await asyncio.wait_for( + self.client.start_notify( + NUS_TX, lambda _c, d: self._feed(bytes(d))), + timeout=BLE_CONNECT_TIMEOUT_S) + except (Exception, asyncio.TimeoutError) as exc: + # str(TimeoutError()) is "" -- fall back to the class name so the + # log never shows a bare "BLE connect failed:". + log(f"BLE connect failed: {exc or type(exc).__name__}" + if str(exc) else + f"BLE connect failed: {type(exc).__name__} " + f"after {BLE_CONNECT_TIMEOUT_S}s") + # Drop the half-open client: bleak reports .is_connected True while + # the handshake is still stalled, which makes is_connected() lie and + # the daemon look healthy while it delivers nothing. + try: + await asyncio.wait_for(self.client.disconnect(), timeout=5) + except Exception: + pass + self.client = None + return False await self._send_initial() log(f"BLE linked: {self.device.name}") return True @@ -659,6 +722,37 @@ async def heartbeat_loop(self) -> None: log(f"lifetime rescan failed: {exc}") await self._send_json({"total": 0, "running": 0, "waiting": 0, "msg": ""}) await self.push_tokens() + await self._probe_liveness() + + async def _probe_liveness(self) -> None: + """Judge the link by round-trips, not by what the transport claims. + + bleak reports .is_connected True for links that no longer carry data -- + on 2026-07-25 the heartbeat was logging "write failed: disconnected" + while is_connected() still said True, so the reconnector's + `if not link.is_connected()` never fired and the link stayed dead. The + firmware answers {"cmd":"status"} with {"ack":"status"} on every + transport, which makes it a cheap true round-trip. + """ + reply = await self.query({"cmd": "status"}, ack="status", + timeout=LIVENESS_PROBE_TIMEOUT_S) + if not reply.get("error"): + self._missed_probes = 0 + return + self._missed_probes += 1 + log(f"liveness probe unanswered " + f"({self._missed_probes}/{MAX_MISSED_PROBES})") + if self._missed_probes < MAX_MISSED_PROBES: + return + log("link unresponsive — tearing down so the reconnector can retake it") + self._missed_probes = 0 + try: + await asyncio.wait_for(self.disconnect(), timeout=10) + except Exception: + pass + # Drop the client outright rather than re-reading .is_connected: the + # whole point of this path is that the flag cannot be trusted. + self.client = None async def disconnect(self) -> None: if self.transport in ("tcp", "tcp-listen"): @@ -800,7 +894,25 @@ async def reconnector(): backoff = 2 while True: if not link.is_connected(): - if await link.connect(): + # Watchdog: this task is the ONLY thing that can restore the + # link, so it must never be allowed to park. An unbounded await + # anywhere under connect() would otherwise kill reconnection for + # the life of the process -- silently, with the process still up + # and KeepAlive satisfied. + try: + ok = await asyncio.wait_for(link.connect(), + timeout=RECONNECT_WATCHDOG_S) + except Exception as exc: + log(f"connect attempt aborted: {type(exc).__name__} " + f"after {RECONNECT_WATCHDOG_S:g}s") + # Tear down whatever half-open state the aborted attempt + # left behind, bounded so cleanup can't park us either. + try: + await asyncio.wait_for(link.disconnect(), timeout=10) + except Exception: + pass + ok = False + if ok: backoff = 2 else: await asyncio.sleep(backoff); backoff = min(backoff * 2, 60); continue diff --git a/tools/cc-bridge/test_reconnect.py b/tools/cc-bridge/test_reconnect.py new file mode 100755 index 0000000..7e4233e --- /dev/null +++ b/tools/cc-bridge/test_reconnect.py @@ -0,0 +1,245 @@ +#!/usr/bin/env python3 +"""Offline tests for the bridge's connect / reconnect / liveness paths. + +No hardware, no BLE adapter: each scenario injects a fake `bleak` into +sys.modules and redirects $HOME, so buddy_bridged's Path.home()-derived socket, +log and token state land in a throwaway directory. + + ./test_reconnect.py # run every scenario, print a summary + ./test_reconnect.py ble_hang # run one scenario in-process + +Scenarios exist because of a real outage (2026-07-25): the daemon sat dead for +two nights with the process still up and launchd's KeepAlive satisfied. +`_ble_connect` awaited BleakClient.connect() unbounded -- CoreBluetooth's +connectPeripheral: never times out by design -- which parked the reconnector +task, and the reconnector is the only thing that can ever restore the link. + + ble_hang peripheral accepts the link then stalls the GATT handshake + ble_ok healthy peripheral still links (guards the timeout wrap) + watchdog an unbounded await with NO inner timeout, i.e. the next + regression -- only reconnector()'s own watchdog catches it + liveness_dead transport claims .is_connected True but carries no data + liveness_alive responsive link must NOT be torn down + +Note on $HOME: AF_UNIX paths max out around 104 chars on macOS, so the temp +home has to be short or start_unix_server dies with "AF_UNIX path too long". +The runner uses /tmp for exactly that reason -- don't "fix" it to somewhere +deeper. +""" +import asyncio +import json +import os +import subprocess +import shutil +import sys +import tempfile +import types + +HERE = os.path.dirname(os.path.abspath(__file__)) +SCENARIOS = ["ble_hang", "ble_ok", "watchdog", "liveness_dead", + "liveness_alive"] + + +# ── fakes ──────────────────────────────────────────────────────────────── + +class FakeDevice: + name = "Claude-TEST" + address = "00:00:00:00:00:00" + + +def make_scanner(hang: bool = False): + class _Scanner: + @staticmethod + async def discover(timeout=5.0): + if hang: + await asyncio.Event().wait() # unbounded on purpose + return [FakeDevice()] + return _Scanner + + +def make_client(*, hang_connect=False, answers=False, lies=False): + class _Client: + def __init__(self, address): + self.address = address + # `lies` models bleak reporting a live link for a dead one, which + # is what made is_connected() untrustworthy during the outage. + self.is_connected = bool(lies) + self._cb = None + + async def connect(self): + if hang_connect: + await asyncio.Event().wait() + self.is_connected = True + + async def start_notify(self, char, cb): + self._cb = cb + + async def write_gatt_char(self, char, data, response=False): + if not answers or self._cb is None: + return # write "succeeds", nothing returns + try: + msg = json.loads(data.decode().strip()) + except Exception: + return + if msg.get("cmd") == "status": + reply = json.dumps({"ack": "status", "ok": True}).encode() + self._cb(None, bytearray(reply + b"\n")) + + async def disconnect(self): + if not lies: # a lying transport stays "up" + self.is_connected = False + + return _Client + + +def install_bleak(client, scanner) -> None: + mod = types.ModuleType("bleak") + mod.BleakClient = client + mod.BleakScanner = scanner + sys.modules["bleak"] = mod + + +def load_bridge(): + """Import buddy_bridged only AFTER the fake bleak and $HOME are in place.""" + sys.path.insert(0, HERE) + for var in ("BUDDY_SERIAL", "BUDDY_HOST", "BUDDY_LISTEN"): + os.environ.pop(var, None) + import buddy_bridged + return buddy_bridged + + +async def run_daemon(bridge, seconds: float) -> str: + """Run the real main() for a bounded window; return the log it wrote.""" + bridge.SOCK_DIR.mkdir(parents=True, exist_ok=True) + if bridge.LOG_PATH.exists(): + bridge.LOG_PATH.unlink() + try: + await asyncio.wait_for(bridge.main(), timeout=seconds) + except (asyncio.TimeoutError, asyncio.CancelledError): + pass + return bridge.LOG_PATH.read_text() if bridge.LOG_PATH.exists() else "" + + +# ── scenarios ──────────────────────────────────────────────────────────── + +async def s_ble_hang() -> tuple[bool, str]: + install_bleak(make_client(hang_connect=True, lies=True), + make_scanner()) + bridge = load_bridge() + link = bridge.BuddyLink() + # getattr, not attribute access: the scenario must be runnable against a + # build that predates the constant, or it cannot prove it catches the bug. + bound = getattr(bridge, "BLE_CONNECT_TIMEOUT_S", 20) + 10 + try: + result = await asyncio.wait_for(link.connect(), timeout=bound) + except asyncio.TimeoutError: + return False, (f"connect() still parked after {bound}s -- " + f"reconnection is permanently dead") + if result is not False: + return False, f"connect() returned {result!r}, expected False" + return True, "stalled handshake gives up instead of parking" + + +async def s_ble_ok() -> tuple[bool, str]: + install_bleak(make_client(answers=True), make_scanner()) + bridge = load_bridge() + link = bridge.BuddyLink() + ok = await asyncio.wait_for(link.connect(), timeout=15) + if ok is not True: + return False, f"connect() returned {ok!r}, expected True" + if not link.is_connected(): + return False, "is_connected() False after a successful link" + if link.client is None: + return False, "client was cleared on the success path" + return True, "healthy peripheral still links, client retained" + + +async def s_watchdog() -> tuple[bool, str]: + # discover() hangs, so BLE_CONNECT_TIMEOUT_S never applies -- this models + # the NEXT unbounded await someone adds, not the one already fixed. + os.environ["BUDDY_CONNECT_WATCHDOG_S"] = "6" + install_bleak(make_client(), make_scanner(hang=True)) + bridge = load_bridge() + text = await run_daemon(bridge, 26.0) + n = text.count("connect attempt aborted") + if n < 2: + return False, (f"only {n} aborted-attempt line(s) -- the reconnector " + f"parked (outage reproduced)") + return True, f"watchdog fired {n}x, reconnection stays alive" + + +async def _liveness(answers: bool) -> tuple[bool, str]: + install_bleak(make_client(answers=answers, lies=not answers), + make_scanner()) + bridge = load_bridge() + bridge.HEARTBEAT_S = 1 # compress the cadence + bridge.LIVENESS_PROBE_TIMEOUT_S = 1 + text = await run_daemon(bridge, 14.0) + torn = text.count("link unresponsive") + linked = text.count("BLE linked") + if answers: + if torn: + return False, f"healthy link torn down {torn}x -- regression" + return True, "responsive link left alone" + if torn < 1: + return False, "half-dead link never torn down (is_connected() lie wins)" + if linked < 2: + return False, f"torn down but never relinked (BLE linked x{linked})" + return True, f"half-dead link torn down {torn}x and relinked" + + +async def s_liveness_dead() -> tuple[bool, str]: + return await _liveness(answers=False) + + +async def s_liveness_alive() -> tuple[bool, str]: + return await _liveness(answers=True) + + +# ── entry points ───────────────────────────────────────────────────────── + +async def run_one(name: str) -> int: + ok, detail = await globals()[f"s_{name}"]() + print(f"{'PASS' if ok else 'FAIL'}: {name} -- {detail}") + return 0 if ok else 1 + + +def run_all() -> int: + failures = [] + for name in SCENARIOS: + # Short $HOME: AF_UNIX paths are capped near 104 chars. + home = tempfile.mkdtemp(prefix="cb", dir="/tmp") + os.makedirs(os.path.join(home, ".claude", "projects"), exist_ok=True) + env = dict(os.environ, HOME=home) + env.pop("BUDDY_CONNECT_WATCHDOG_S", None) + try: + proc = subprocess.run([sys.executable, os.path.abspath(__file__), + name], + env=env, capture_output=True, text=True, + timeout=120) + line = [ln for ln in proc.stdout.splitlines() + if ln.startswith(("PASS:", "FAIL:"))] + print(line[-1] if line else + f"FAIL: {name} -- no verdict\n{proc.stdout}{proc.stderr}") + if proc.returncode != 0: + failures.append(name) + except subprocess.TimeoutExpired: + print(f"FAIL: {name} -- scenario timed out") + failures.append(name) + finally: + shutil.rmtree(home, ignore_errors=True) + print() + if failures: + print(f"{len(failures)}/{len(SCENARIOS)} failed: {', '.join(failures)}") + return 1 + print(f"all {len(SCENARIOS)} scenarios passed") + return 0 + + +if __name__ == "__main__": + if len(sys.argv) > 1: + if sys.argv[1] not in SCENARIOS: + sys.exit(f"unknown scenario {sys.argv[1]!r}; " + f"pick from {', '.join(SCENARIOS)}") + sys.exit(asyncio.run(run_one(sys.argv[1]))) + sys.exit(run_all())