From 4abb57652c2fa0c9454194108459c0cbdcb001cc Mon Sep 17 00:00:00 2001 From: Michael Adam Groberman <81723568+MichaelAdamGroberman@users.noreply.github.com> Date: Sat, 25 Jul 2026 11:09:50 -0400 Subject: [PATCH] fix(cc-bridge): stop the reconnector deadlocking on a stalled BLE link MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The daemon could sit dead with the process still up and launchd's KeepAlive fully satisfied — twice in one day, once overnight for eight hours. The tell in buddy.log is a "scanning for Claude* (5s)…" line with no outcome after it, then permanent silence, and zero tracebacks in buddy.out. _ble_connect awaited BleakClient.connect() unbounded. CoreBluetooth's connectPeripheral: has no timeout by design — it stays pending until the peripheral turns up — and bleak inherits that, so a device that accepts the link and then stalls the GATT handshake parked the coroutine. reconnector() is a single sequential loop and the only thing that can restore the link, so one parked connect() killed reconnection for the life of the process. Nothing raised, so gather() never propagated and the process never exited for KeepAlive to catch. BLE was the only transport exposed: _tcp_connect already wraps every await in wait_for, and _serial_connect's ser.open() is synchronous. Three layers, because each catches what the others cannot: - BLE_CONNECT_TIMEOUT_S bounds connect() and start_notify(), and clears self.client on failure — bleak reports .is_connected True mid-stall, which made is_connected() lie and the daemon look healthy while delivering nothing. - RECONNECT_WATCHDOG_S bounds link.connect() inside reconnector() whatever the transport, so the next unbounded await added anywhere degrades to a retry instead of a silent outage. - _probe_liveness() judges the link by round-trips instead of the transport flag: the firmware answers {"cmd":"status"} with {"ack":"status"} on every transport (xfer.h), so an unanswered probe MAX_MISSED_PROBES times tears the link down and lets the reconnector retake it. Also make BUDDY_SERIAL a preference rather than a pin. The USB node disappears whenever the cable is pulled, and pinning turned that into a permanent crash-loop indistinguishable from "device offline"; connect() now re-checks the node each attempt and falls back to BLE. test_reconnect.py covers all of it offline via a fake bleak and a throwaway $HOME — no hardware, no adapter. Verified against the pre-fix file: ble_hang, watchdog and liveness_dead fail there and pass here, while ble_ok and liveness_alive pass on both, so the guards aren't false alarms. The liveness probe was also checked against real hardware — zero unanswered probes on a healthy serial link. --- tools/cc-bridge/README.md | 18 +++ tools/cc-bridge/buddy_bridged.py | 126 ++++++++++++++- tools/cc-bridge/test_reconnect.py | 245 ++++++++++++++++++++++++++++++ 3 files changed, 382 insertions(+), 7 deletions(-) create mode 100755 tools/cc-bridge/test_reconnect.py 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())