fix(cc-bridge): run the status token scan off the event loop

tools/cc-bridge/buddy_bridged.py

@@ -776,7 +776,13 @@ async def handle_client(link: BuddyLink, reader, writer) -> None:
             writer.write(b'{"error":"bad json"}\n'); await writer.drain(); return
         op = req.get("op")
         if op == "status":
-            period_tokens = link.period_tokens()
+            # Off-thread, like the heartbeat already does: period_tokens()
+            # globs and reads every transcript line-by-line on EVERY call, and
+            # with period="all" the mtime filter skips nothing, so it re-reads
+            # the whole corpus. Run inline it freezes prompts, heartbeats and
+            # liveness acks for as long as the scan takes -- which grows
+            # without bound as transcripts accumulate.
+            period_tokens = await asyncio.to_thread(link.period_tokens)
             life = link.lifetime_tokens()
             ok, deny = link.approved_denied()
             resp = {

tools/cc-bridge/test_reconnect.py

@@ -33,11 +33,12 @@
 import shutil
 import sys
 import tempfile
+import time
 import types
 
 HERE = os.path.dirname(os.path.abspath(__file__))
 SCENARIOS = ["ble_hang", "ble_ok", "watchdog", "liveness_dead",
-             "liveness_alive"]
+             "liveness_alive", "status_nonblocking"]
 
 
 # ── fakes ────────────────────────────────────────────────────────────────
@@ -188,6 +189,83 @@ async def _liveness(answers: bool) -> tuple[bool, str]:
     return True, f"half-dead link torn down {torn}x and relinked"
 
 
+async def s_status_nonblocking() -> tuple[bool, str]:
+    """An {"op":"status"} must not stall the event loop.
+
+    period_tokens() globs and reads every transcript line-by-line on EVERY
+    status call, and with period="all" the mtime filter skips nothing -- so it
+    re-reads the whole corpus. Run on the event loop that freezes prompts,
+    heartbeats and liveness acks for the duration; observed 2026-07-25 as
+    gr0m_status timing out while prompts still worked.
+    """
+    install_bleak(make_client(answers=True), make_scanner())
+    bridge = load_bridge()
+
+    SLOW_S = 2.0
+    bridge.BuddyLink.period_tokens = lambda self: (time.sleep(SLOW_S) or 0)
+
+    stop = asyncio.Event()
+    ticks = []
+
+    async def monitor():
+        # Record BEFORE checking stop: the tick that lands right after a stall
+        # is the one that reveals it, and checking stop first drops exactly
+        # that sample -- which silently made this scenario unfalsifiable.
+        loop = asyncio.get_running_loop()
+        while True:
+            ticks.append(loop.time())
+            if stop.is_set():
+                return
+            await asyncio.sleep(0.05)
+
+    bridge.SOCK_DIR.mkdir(parents=True, exist_ok=True)
+    daemon = asyncio.create_task(bridge.main())
+    for _ in range(100):                       # wait for the socket to bind
+        if bridge.SOCK_PATH.exists():
+            break
+        await asyncio.sleep(0.1)
+    else:
+        daemon.cancel()
+        return False, "daemon never bound its socket"
+
+    mon = asyncio.create_task(monitor())
+    await asyncio.sleep(0.3)                   # collect a clean baseline
+    try:
+        reader, writer = await asyncio.open_unix_connection(
+            str(bridge.SOCK_PATH))
+        t0 = asyncio.get_running_loop().time()
+        writer.write(b'{"op":"status"}\n')
+        await writer.drain()
+        raw = await asyncio.wait_for(reader.readline(), timeout=30)
+        call_s = asyncio.get_running_loop().time() - t0
+        writer.close()
+    finally:
+        stop.set()
+        await mon
+        daemon.cancel()
+
+    # Without these the scenario is unfalsifiable: a handler that errors out
+    # closes the connection, readline returns b"", nothing ever stalls, and the
+    # test "passes" having measured nothing.
+    try:
+        reply = json.loads(raw.decode())
+    except Exception:
+        return False, f"status returned no usable reply ({raw!r})"
+    if "transport" not in reply:
+        return False, f"status reply missing fields: {reply}"
+    if call_s < SLOW_S * 0.9:
+        return False, (f"status returned in {call_s:.2f}s -- the slow "
+                       f"period_tokens patch never ran, so nothing was tested")
+
+    gaps = [b - a for a, b in zip(ticks, ticks[1:])]
+    worst = max(gaps) if gaps else 0.0
+    if worst > SLOW_S / 2:
+        return False, (f"event loop stalled {worst:.2f}s during one status "
+                       f"call -- prompts and acks are frozen that whole time")
+    return True, (f"loop stayed responsive (worst gap {worst:.2f}s) while a "
+                  f"{call_s:.1f}s status call ran")
+
+
 async def s_liveness_dead() -> tuple[bool, str]:
     return await _liveness(answers=False)