wenil/clawdmeter
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 3 Wiki Activity

Fix macOS BLE daemon: connect to HID-held device via retrieveConnected

Browse Source 
On macOS the firmware is auto-connected by the OS as a BLE HID keyboard,
and CoreBluetooth excludes already-connected peripherals from scan
results. bleak's connect-by-address path also scans internally, so the
daemon's scan loop never found the device ("Device not found" forever)
even though it was plainly visible in System Settings.

Discover the target on macOS via CoreBluetooth's
retrieveConnectedPeripheralsWithServices_ and connect to the returned
peripheral directly (no scan). The custom service UUID is matched first
(unambiguous); the generic HID service 0x1812 is only trusted on an exact
DEVICE_NAME match so it can't grab an unrelated keyboard/mouse. A
peripheral that fails to connect is skipped for one cycle so the scan
fallback stays reachable. The device's two connection slots let the OS
HID link and the daemon run simultaneously.

Linux/BlueZ path is unchanged (still scans + caches address); the
redundant per-reconnect address save was removed so caching happens only
on a fresh scan.

Adds daemon/test_macos_connect.py, a foreground smoke test for the macOS
connect path.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
soapiece
2026-05-30 21:32:33 -07:00
co-authored by Claude Opus 4.8
parent 6b5314ef28
commit 18282e0a93
2 changed files with 201 additions and 23 deletions
Download Patch File Download Diff File Expand all files Collapse all files
daemon/claude_usage_daemon.py
+145 -23
View File
@@ -157,6 +157,120 @@ async def scan_for_device() -> str | None:
return None
# --- macOS: recover a device the OS already holds as an HID keyboard --------
#
# The firmware advertises as a BLE HID keyboard so its buttons type into the
# Mac. macOS auto-connects to that HID, and CoreBluetooth then EXCLUDES the
# peripheral from BleakScanner.discover() results (already-connected devices
# never appear in scans). bleak's connect-by-address path also scans
# internally, so a cached address can't help either. The documented escape
# hatch is retrieveConnectedPeripheralsWithServices_, which returns
# peripherals the system is already connected to. We wrap the result in a
# BLEDevice carrying the live (peripheral, manager) details so BleakClient
# connects to it directly without scanning. CoreBluetooth shares the single
# physical link, so this rides the existing HID connection — the keyboard
# keeps working.
_cb_manager = None # reused CentralManagerDelegate (CoreBluetooth)
async def _get_cb_manager():
"""Lazily create and ready a shared CoreBluetooth central manager."""
global _cb_manager
if _cb_manager is None:
from bleak.backends.corebluetooth.CentralManagerDelegate import (
CentralManagerDelegate,
)
mgr = CentralManagerDelegate()
await mgr.wait_until_ready() # raises if Bluetooth is unauthorized/off
_cb_manager = mgr
return _cb_manager
async def retrieve_connected_macos(skip_addr: str | None = None):
"""Return a BLEDevice for a system-connected 'Claude Controller', or None.
Two-step lookup, strongest signal first:
1. Peripherals connected under our CUSTOM service UUID. Membership in
that service is unambiguous (no other device exposes it), so we accept
by service alone — the peripheral's name can be None on macOS.
2. Fall back to the generic HID service 0x1812, but ONLY trust a
peripheral whose name matches DEVICE_NAME. 0x1812 also matches
unrelated keyboards/mice, so picking blindly here could grab the
wrong device.
``skip_addr`` skips a peripheral whose UUID just failed to connect, so a
stale CoreBluetooth handle can't trap us into never trying a fresh scan.
"""
from CoreBluetooth import CBUUID
from bleak.backends.device import BLEDevice
try:
manager = await _get_cb_manager()
except Exception as e: # BleakBluetoothNotAvailableError etc.
log(f"CoreBluetooth unavailable: {e}")
return None
cm = manager.central_manager
def _wrap(p):
addr = p.identifier().UUIDString()
log(f"Found system-connected peripheral: {p.name()!r} [{addr}]")
return BLEDevice(addr, p.name(), (p, manager))
def _ok(p) -> bool:
return not (skip_addr and p.identifier().UUIDString() == skip_addr)
# 1. Custom service — accept by service membership alone.
custom = cm.retrieveConnectedPeripheralsWithServices_(
[CBUUID.UUIDWithString_(SERVICE_UUID)]
)
for p in custom or []:
if _ok(p):
return _wrap(p)
# 2. Generic HID service — require an exact name match.
hid = cm.retrieveConnectedPeripheralsWithServices_(
[CBUUID.UUIDWithString_("1812")]
)
for p in hid or []:
if _ok(p) and p.name() == DEVICE_NAME:
return _wrap(p)
return None
async def discover_target(skip_addr: str | None = None):
"""Return a connectable target, or None.
macOS: prefer the system-connected peripheral (HID-grabbed devices are
invisible to scans); fall back to a normal scan that yields a BLEDevice
so the subsequent connect doesn't have to re-scan. ``skip_addr`` is
forwarded so a just-failed peripheral is skipped, making the scan
fallback reachable.
Other platforms: keep the original cached-address / scan-by-name flow.
A freshly scanned address is cached here (the only place it's saved).
"""
if sys.platform == "darwin":
dev = await retrieve_connected_macos(skip_addr=skip_addr)
if dev is not None:
return dev
log(f"Not held by OS; scanning for '{DEVICE_NAME}' ({SCAN_TIMEOUT}s)...")
dev = await BleakScanner.find_device_by_name(DEVICE_NAME, timeout=SCAN_TIMEOUT)
if dev:
log(f"Found: {dev.address}")
return dev
address = load_cached_address()
if not address:
address = await scan_for_device()
if address:
save_address(address) # cache only freshly-scanned addresses
return address
async def poll_api(token: str) -> dict | None:
headers = dict(API_HEADERS_TEMPLATE)
headers["Authorization"] = f"Bearer {token}"
@@ -226,15 +340,17 @@ class Session:
return False
async def connect_and_run(address: str, stop_event: asyncio.Event) -> bool:
"""Connect to a known address and poll until disconnected or stopped.
async def connect_and_run(target, stop_event: asyncio.Event) -> bool:
"""Connect to a target and poll until disconnected or stopped.
Returns True if the connection was used successfully (so the caller
keeps the cached address), False if the connection failed and the
cache should be invalidated.
``target`` is either an address string (Linux) or a BLEDevice carrying
live CoreBluetooth details (macOS). Returns True if the connection was
used successfully (so the caller keeps the cached address), False if the
connection failed and the cache should be invalidated.
"""
log(f"Connecting to {address}...")
client = BleakClient(address)
display = target if isinstance(target, str) else target.address
log(f"Connecting to {display}...")
client = BleakClient(target)
try:
await client.connect()
except (BleakError, asyncio.TimeoutError) as e:
@@ -299,25 +415,31 @@ async def main() -> None:
log(f"Poll interval: {POLL_INTERVAL}s")
backoff = 1
skip_addr: str | None = None # macOS: a peripheral to skip for one cycle
while not stop_event.is_set():
address = load_cached_address()
if not address:
address = await scan_for_device()
if address:
save_address(address)
else:
log(f"Device not found, retrying in {backoff}s...")
try:
await asyncio.wait_for(stop_event.wait(), timeout=backoff)
except asyncio.TimeoutError:
pass
backoff = min(backoff * 2, 60)
continue
# Apply any pending skip exactly once, then clear it so the next
# cycle re-tries retrieveConnected (the device may have recovered).
target = await discover_target(skip_addr=skip_addr)
skip_addr = None
if not target:
log(f"Device not found, retrying in {backoff}s...")
try:
await asyncio.wait_for(stop_event.wait(), timeout=backoff)
except asyncio.TimeoutError:
pass
backoff = min(backoff * 2, 60)
continue
ok = await connect_and_run(address, stop_event)
addr = target if isinstance(target, str) else target.address
ok = await connect_and_run(target, stop_event)
if not ok:
log("Invalidating cached address")
SAVED_ADDR_FILE.unlink(missing_ok=True)
if sys.platform == "darwin":
# No string cache to drop; instead skip this stale handle on
# the next retrieveConnected so the scan fallback is reachable.
skip_addr = addr
else:
log("Invalidating cached address")
SAVED_ADDR_FILE.unlink(missing_ok=True)
try:
await asyncio.wait_for(stop_event.wait(), timeout=backoff)
except asyncio.TimeoutError: