Merge pull request #57 from kvenanzi/windows-daemon

This commit is contained in:
Hermann Björgvin
2026-06-03 11:31:20 +00:00
committed by GitHub
20 changed files with 3592 additions and 2 deletions
+67 -2
View File
@@ -37,10 +37,11 @@ Boards supported out of the box:
## Prerequisites
- Linux (tested on Ubuntu) or macOS
- Linux (tested on Ubuntu), macOS, or Windows 10/11
- [PlatformIO CLI](https://docs.platformio.org/en/latest/core/installation/index.html)
- Linux: `curl`, `bluetoothctl`, `busctl` (BlueZ Bluetooth stack)
- macOS: `python3` (the installer sets up a venv with `bleak` and `httpx`)
- Windows: `python3` 3.11+ (the installer sets up a venv with `bleak`, `httpx`, and `pystray`)
- Claude Code with an active subscription
## macOS installation
@@ -118,9 +119,73 @@ Check status: `systemctl --user status claude-usage-daemon`
View logs: `journalctl --user -u claude-usage-daemon -f`
## Windows installation
Runs natively on Windows — no WSL required. A system-tray app polls your usage and pushes it over BLE, and starts automatically at login.
### Prerequisites
- **Native Windows** (not WSL).
- **Python 3.11+** from [python.org](https://www.python.org/downloads/) — check *"Add python.exe to PATH"* during install.
- **Claude Code** installed, with `claude login` completed. The token is read from `%USERPROFILE%\.claude\.credentials.json` (falling back to `%LOCALAPPDATA%\Claude\` then `%APPDATA%\Claude\`).
- The repo on a **native Windows path** (e.g. `%USERPROFILE%\Clawdmeter`), **not** a `\\wsl$` share — the installer refuses a WSL path.
### Flash the firmware
```powershell
pio run -d firmware -e waveshare_amoled_216 -t upload --upload-port COM5 # use your device's COM port
```
Run `pio run -d firmware` with no env to see the available board envs.
### Pair the device
The device is a bonded BLE HID keyboard, so pair it once: **Settings → Bluetooth & devices → Add device → Bluetooth**, then select "Claude Controller". Pairing is **required** — it enables the physical buttons and keeps a persistent connection (the device keeps showing your last-synced usage even after the daemon quits). To undo, use **Remove device** (this disables the buttons).
### Install the daemon (recommended)
From the repo root in PowerShell:
```powershell
powershell -ExecutionPolicy Bypass -File install-windows.ps1
```
This creates a venv, installs `bleak`/`httpx`/`pystray`/`Pillow` from the in-repo requirements (no internet downloads), registers a per-user login-autostart entry (`HKCU\…\Run`, no admin needed), and launches the tray app headlessly (no console window).
### Run manually instead (optional)
```powershell
python -m venv .venv
.venv\Scripts\Activate.ps1 # if blocked: Set-ExecutionPolicy -Scope CurrentUser RemoteSigned, then retry
pip install -r daemon\requirements-windows.txt
python daemon\claude_usage_daemon_windows.py # runs in the foreground; Ctrl+C to stop
```
### Tray icon and menu
The icon's corner bubble shows state — **green** Connected, **amber** Scanning, **red** Error — and hovering shows the status (`Connected · last update HH:MM`). A notification fires once when it enters Error (e.g. an expired token). Right-click for the menu:
- **Status header** — live state + last sync time.
- **Start at login** — toggle autostart on/off.
- **Quit** — stops the daemon cleanly; leaves the Windows pairing intact (device keeps its last reading).
### Logs and troubleshooting
```powershell
Get-Content $env:LOCALAPPDATA\Clawdmeter\daemon.log -Tail 30 # view logs
reg delete "HKCU\Software\Microsoft\Windows\CurrentVersion\Run" /v Clawdmeter /f # remove autostart
```
| Symptom | Fix |
|---------|-----|
| `Device not found` | Power on the device; make sure it's in range and paired. |
| `token expired` toast / `API HTTP 401` | Re-run `claude login`, then restart the daemon. |
| `Connection failed` | Toggle Windows Bluetooth off/on in Settings. |
| `Warning: running under Linux/WSL` | Run from a native PowerShell window, not a WSL shell. |
## How it works
1. The daemon reads your Claude Code OAuth token — from the macOS Keychain (service `Claude Code-credentials`) on macOS, or from `~/.claude/.credentials.json` on Linux.
1. The daemon reads your Claude Code OAuth token — from the macOS Keychain (service `Claude Code-credentials`) on macOS, or from `~/.claude/.credentials.json` on Linux (`%USERPROFILE%\.claude\.credentials.json` on Windows).
2. It makes a minimal API call to `api.anthropic.com/v1/messages` — one token of Haiku, basically free.
3. The usage numbers come straight out of the response headers (`anthropic-ratelimit-unified-5h-utilization` and friends).
4. The daemon connects to the ESP32 over BLE and writes a JSON payload to the GATT RX characteristic.
+6
View File
@@ -0,0 +1,6 @@
"""Root conftest.py — adds the repo root to sys.path so `import daemon.*` resolves."""
import os
import sys
sys.path.insert(0, os.path.dirname(__file__))
+223
View File
@@ -0,0 +1,223 @@
# Windows Setup and Run Guide
This guide covers running the Clawdmeter Windows daemon on native Windows hardware.
It includes the turnkey `install-windows.ps1` bootstrap (tray icon + login autostart),
the manual-run fallback, and how to manage or remove autostart.
---
## Prerequisites
| Requirement | Details |
|-------------|---------|
| **Native Windows** | Must run on real Windows — not WSL. The script prints a warning and BLE will not work under WSL. |
| **Python 3.11+** | Download from [python.org](https://www.python.org/downloads/) if not already installed. Ensure "Add python.exe to PATH" is checked during install. |
| **Claude Code installed** | Install Claude Code and complete `claude login` so credentials exist on disk. |
| **Clawdmeter powered on** | The device must be powered on and in range before the daemon starts. |
| **Paired with Windows Bluetooth** | Pair the device once via **Settings → Bluetooth & devices → Add device** (see [Pair the device](#pair-the-device-one-time)). This is required — the device is a bonded BLE HID keyboard, so pairing enables its physical buttons and keeps a persistent connection that shows your last usage even when the daemon is stopped. |
### Where are my credentials?
`claude login` writes the OAuth token to (first match wins):
1. `%USERPROFILE%\.claude\.credentials.json` — primary path (confirmed by Claude Code docs)
2. `%LOCALAPPDATA%\Claude\.credentials.json` — fallback
3. `%APPDATA%\Claude\.credentials.json` — fallback
The daemon probes these paths in order. You can also set `CLAUDE_CREDENTIALS_PATH` to an
absolute path or `CLAUDE_CONFIG_DIR` to a directory to override the search entirely.
> **Security note:** The credentials file contains your OAuth token. Never share its contents
> or embed it in scripts. The daemon reads it from disk and uses it only as the API
> `Authorization` header — the token is never written to any log, tooltip, or notification.
---
## Pair the device (one time)
The Clawdmeter is a **bonded BLE HID keyboard** as well as a usage display — its firmware
enables bonding (`NimBLEDevice::setSecurityAuth`) and advertises the HID service so its
physical buttons act as a keyboard (Space / Shift+Tab). Pair it with Windows **once**,
before running the daemon:
1. Put the device on its Bluetooth waiting screen (powered on, not yet connected).
2. Open **Settings → Bluetooth & devices → Add device → Bluetooth**.
3. Select **Claude Controller** and complete pairing.
**Why this is required:**
- **Keyboard buttons** — HID over BLE requires bonding on Windows. Without pairing, the
device's buttons won't reach the PC.
- **Persistent point-in-time view** — once paired, Windows maintains the BLE link and
auto-reconnects the device whenever it is in range. This is intentional: the device keeps
showing your **last-synced** usage even after you Quit the daemon, as a glanceable
point-in-time view. Quitting the daemon releases only its data connection — it does **not**
drop the Windows pairing, so the device stays connected to Windows.
To undo, use **Settings → Bluetooth & devices → (device) → Remove device**. Removing the
pairing disables the keyboard buttons.
---
## Setup (one time)
Open a PowerShell terminal and `cd` to the repository root.
**1. Create a virtual environment**
```powershell
python -m venv .venv
```
**2. Activate it**
```powershell
.venv\Scripts\Activate.ps1
```
If you see a scripts-execution-policy error, run:
```powershell
Set-ExecutionPolicy -Scope CurrentUser -ExecutionPolicy RemoteSigned
```
Then repeat the `Activate.ps1` step.
**3. Install dependencies**
```powershell
pip install -r daemon\requirements-windows.txt
```
This installs `bleak` (WinRT BLE) and `httpx` (async HTTP for the Anthropic API).
---
## Running the daemon
With the venv active and the Clawdmeter powered on:
```powershell
python daemon\claude_usage_daemon_windows.py
```
### Expected console output
```
[HH:MM:SS] === Claude Usage Tracker Daemon (BLE, Windows) ===
[HH:MM:SS] Poll interval: 60s
[HH:MM:SS] Scanning for 'Claude Controller' (8.0s)...
[HH:MM:SS] Found: XX:XX:XX:XX:XX:XX
[HH:MM:SS] Connecting to XX:XX:XX:XX:XX:XX...
[HH:MM:SS] Connected
[HH:MM:SS] Sending: {"s":42,"sr":180,"w":17,"wr":8820,"st":"active","ok":true}
```
- **The device must be paired with Windows first** (see [Pair the device](#pair-the-device-one-time)).
The daemon then connects over that existing link via `BleakScanner` + `BleakClient`; it does
not pop its own pairing dialog.
- After `Connected`, the daemon polls the Anthropic API immediately and sends the first
payload within a few seconds of connect (warm token path). With a valid, non-expired token
the device should leave its waiting screen and show session + weekly percentages within
about 10 seconds of launch.
- The daemon then re-polls every 60 seconds while connected. If the device fires a refresh
request (e.g., after a button press), an immediate re-poll occurs without waiting for the
60-second interval.
- If the device disconnects or goes out of range, the daemon logs `Device disconnected` and
re-scans automatically with exponential backoff (starting at 1 second, capped at 60 seconds).
### Stopping
Press **Ctrl+C** in the terminal. The daemon logs `Daemon stopping` and exits cleanly.
---
## Troubleshooting
| Symptom | Likely cause | Fix |
|---------|-------------|-----|
| `Warning: running under Linux/WSL` | Running in WSL, not native Windows | Run from a native PowerShell or Command Prompt on Windows |
| `Scanning for 'Claude Controller'… Device not found` | Clawdmeter is off, out of range, or showing a non-Bluetooth screen | Power on the device and ensure it is on the Bluetooth waiting screen |
| `No token; skipping poll` | No credentials file found at any candidate path | Confirm `claude login` ran on this machine; check `%USERPROFILE%\.claude\.credentials.json` exists |
| `API HTTP 401` | Token expired | Re-run `claude login` in a terminal to refresh the token, then restart the daemon |
| `Connection failed` | WinRT BLE initialisation issue | Ensure Windows Bluetooth is on; try toggling Bluetooth off/on in Windows Settings |
---
## Tray icon, login autostart, and turnkey install
### One-command install (recommended)
> **Copy the repo to a native Windows path first.** Clone or copy this repository
> to a Windows location such as `%USERPROFILE%\Clawdmeter` — **not** a WSL share
> (`\\wsl$\...` or `\\wsl.localhost\...`). Installing from the WSL share would point
> the virtual environment and the login-autostart entry at a path that disappears when
> WSL shuts down, defeating the whole point of the Windows daemon. The installer
> detects a WSL path and refuses to run, telling you how to relocate.
>
> ```powershell
> Copy-Item -Recurse '\\wsl.localhost\Ubuntu\home\<you>\repos\Clawdmeter' "$env:USERPROFILE\Clawdmeter"
> cd "$env:USERPROFILE\Clawdmeter"
> ```
Run this once from the repository root in PowerShell (a native Windows path):
```powershell
powershell -ExecutionPolicy Bypass -File install-windows.ps1
```
The script does four things in order and logs progress at each step:
1. Creates a Python virtual environment at `.venv`.
2. Installs dependencies from `daemon\requirements-windows.txt` (bleak, httpx, pystray, Pillow).
3. Registers the tray app to launch automatically at login via `HKCU\Software\Microsoft\Windows\CurrentVersion\Run` — per-user, no admin required.
4. Launches the tray app immediately (headless — no console window).
The script downloads nothing from the internet. It only installs the packages listed in
the in-repo `daemon\requirements-windows.txt`.
### Tray icon and status
After install, the Clawdmeter icon appears in the Windows notification area:
| State | Icon bubble | Tooltip |
|-------|-------------|---------|
| Connected | green | `Connected · last update HH:MM` |
| Scanning | amber | `Scanning…` |
| Error | red | `Error: token expired — run claude login` |
Hover over the icon to see the current status tooltip. A notification fires once when the
daemon first enters the Error state (e.g. after a token expiry).
### Tray menu
Right-click the tray icon for the menu:
- **Status header** (non-clickable) — live status + last data sync time.
- **Start at login** (checkable toggle) — enables or disables autostart at runtime.
Reflects the current registry state each time the menu opens.
- **Quit** — stops the daemon cleanly and exits with no lingering process. It releases the
daemon's own data connection but does **not** drop the Windows Bluetooth pairing — the
device stays connected to Windows and keeps showing your last-synced usage (point-in-time
view).
### Disabling or removing autostart
Use the tray menu toggle, or remove the registry value manually:
```powershell
reg delete "HKCU\Software\Microsoft\Windows\CurrentVersion\Run" /v Clawdmeter /f
```
### WSL independence
The daemon operates fully independently of WSL. The token is read from native Windows
credential paths (`%USERPROFILE%\.claude\.credentials.json` and fallbacks); BLE uses
the WinRT stack directly. Running `wsl --shutdown` does not affect the BLE link, and
the daemon starts correctly even in a fresh Windows session where WSL has never been
launched.
---
## What is NOT covered here
- PyInstaller / one-file `.exe` packaging — v2
- MAC-address cache / sleep-wake reconnect hardening — Phase 3
View File
+108
View File
@@ -0,0 +1,108 @@
"""Login-autostart toggle for Clawdmeter — APP-01 / D-07.
Manages a per-user HKCU\\Software\\Microsoft\\Windows\\CurrentVersion\\Run
registry value named "Clawdmeter" that launches the tray app headlessly via
pythonw.exe (no console window — D-08).
winreg is Windows stdlib; this module guards the import so it can be imported
on the Linux dev box (unit tests mock `daemon.autostart_windows.winreg`).
Public API:
enable(tray_script=None) -- write/overwrite the Run value
disable() -- remove the Run value; idempotent when absent
is_enabled() -- True if the Run value is currently present
"""
import os
import sys
import time
# Guard the import so the module is importable off-Windows.
# Unit tests replace this attribute via:
# patch("daemon.autostart_windows.winreg", <MagicMock>)
try:
import winreg as winreg # type: ignore[import]
except ImportError:
winreg = None # type: ignore[assignment]
# Registry key (no leading backslash — OpenKey uses relative path under hive).
_RUN_KEY = r"Software\Microsoft\Windows\CurrentVersion\Run"
_VALUE_NAME = "Clawdmeter"
def log(msg: str) -> None:
"""Log in the daemon [HH:MM:SS] style."""
print(f"[{time.strftime('%H:%M:%S')}] {msg}", flush=True)
def _command(tray_script: str | None = None) -> str:
"""Build the headless launch command for the Run value.
Uses the BASE interpreter's pythonw.exe — `sys.base_exec_prefix` points at
the real Python install even inside a venv — NOT the venv's
`Scripts\\pythonw.exe`. The venv pythonw is a redirector stub that
re-launches the CONSOLE `python.exe` build as a child process (a CPython
venv-launcher bug, verified empirically on Python 3.13), which pops a black
console window at logon and kills the tray when closed (field bug, SC#1).
The base pythonw loads in-process and is genuinely windowless.
The path is never hard-coded (D-08, CLAUDE.md "repoint ExecStart" lesson);
both paths are quoted for space safety. tray_windows.py adds the venv's
site-packages to sys.path itself, so the venv's deps still resolve under the
base interpreter.
Args:
tray_script: absolute path to the tray entry script. Defaults to this
module's own path (useful when autostart_windows.py IS
the entry point, but callers should pass tray_windows.py).
"""
pythonw = os.path.join(sys.base_exec_prefix, "pythonw.exe")
script = os.path.abspath(tray_script if tray_script is not None else __file__)
return f'"{pythonw}" "{script}"'
def enable(tray_script: str | None = None) -> None:
"""Write (or overwrite) the HKCU Run value pointing at pythonw.exe.
No admin elevation required — HKCU is per-user (D-07, ASVS V4).
Args:
tray_script: path to the tray entry script (passed to _command()).
"""
cmd = _command(tray_script)
with winreg.OpenKey(
winreg.HKEY_CURRENT_USER, _RUN_KEY, 0, winreg.KEY_SET_VALUE
) as key:
winreg.SetValueEx(key, _VALUE_NAME, 0, winreg.REG_SZ, cmd)
log(f"Autostart enabled: {cmd}")
def disable() -> None:
"""Remove the HKCU Run value. Idempotent — no error if already absent.
Mirrors the read_token() OSError-swallow pattern (daemon L201-208).
"""
try:
with winreg.OpenKey(
winreg.HKEY_CURRENT_USER, _RUN_KEY, 0, winreg.KEY_SET_VALUE
) as key:
winreg.DeleteValue(key, _VALUE_NAME)
log("Autostart disabled")
except FileNotFoundError:
pass # already absent — idempotent
def is_enabled() -> bool:
"""Return True if the Run value is currently present, False otherwise.
Queries the live registry on every call so the state reflects external
changes (e.g. the user deleting the value manually) — Pitfall 6 guard.
"""
try:
with winreg.OpenKey(
winreg.HKEY_CURRENT_USER, _RUN_KEY, 0, winreg.KEY_QUERY_VALUE
) as key:
winreg.QueryValueEx(key, _VALUE_NAME)
return True
except FileNotFoundError:
return False
+521
View File
@@ -0,0 +1,521 @@
#!/usr/bin/env python3
"""Claude Usage Tracker Daemon — Windows (Phase 2).
Reads the Claude OAuth token from the native-Windows credentials path and
polls the Anthropic API for rate-limit utilization data. BLE glue added in
later plans.
"""
import asyncio
import datetime
import json
import logging
import logging.handlers
import os
import re
import signal
import sys
import threading
import time
from pathlib import Path
import httpx
from bleak import BleakClient, BleakScanner
from bleak.exc import BleakError
DEVICE_NAME = "Claude Controller"
SERVICE_UUID = "4c41555a-4465-7669-6365-000000000001"
RX_CHAR_UUID = "4c41555a-4465-7669-6365-000000000002"
REQ_CHAR_UUID = "4c41555a-4465-7669-6365-000000000004"
POLL_INTERVAL = 60
TICK = 5
SCAN_TIMEOUT = 8.0
CONNECT_RETRIES = 3 # D-01: attempts before giving up on a device
CONNECT_RETRY_DELAY = 2.0 # D-01: seconds between failed connect attempts
ZOMBIE_BREAK_LIMIT = 1 # D-03: consecutive write failures before abandoning a half-open link
# N=1: breaks at T=60s, leaves ~60s headroom for reconnect+poll inside 120s SLA
# N=2 would bust the 120s budget before reconnect even begins
RECONNECT_BACKOFF_CAP = 8 # D-05: fast-reconnect cap (seconds); keeps stacked retries inside 120s SLA
# ~510s band per CONTEXT.md Claude's Discretion; 8 chosen as middle ground
API_URL = "https://api.anthropic.com/v1/messages"
API_HEADERS_TEMPLATE = {
"anthropic-version": "2023-06-01",
"anthropic-beta": "oauth-2025-04-20",
"Content-Type": "application/json",
"User-Agent": "claude-code/2.1.5",
}
API_BODY = {
"model": "claude-haiku-4-5-20251001",
"max_tokens": 1,
"messages": [{"role": "user", "content": "hi"}],
}
def _build_file_logger() -> logging.Logger | None:
"""Create a rotating file logger for field diagnostics, or None.
Autostart launches the tray under pythonw.exe, which has no console — stdout
is discarded (and is in fact None, making print() unsafe). A rotating file is
then the ONLY trail when the daemon stalls in the field. Windows-only: on the
Linux dev box / CI the console print() suffices, and gating to win32 keeps the
pure-helper unit tests from writing stray log files.
"""
if sys.platform != "win32":
return None
logger = logging.getLogger("clawdmeter.daemon")
if logger.handlers:
return logger # idempotent across re-import (tray imports this module)
base = Path(os.environ.get("LOCALAPPDATA", Path.home() / "AppData" / "Local"))
path = base / "Clawdmeter" / "daemon.log"
try:
path.parent.mkdir(parents=True, exist_ok=True)
handler = logging.handlers.RotatingFileHandler(
path, maxBytes=512 * 1024, backupCount=3, encoding="utf-8"
)
except OSError:
return None # best-effort — logging setup must never stop the daemon
handler.setFormatter(logging.Formatter("%(asctime)s %(message)s", "%Y-%m-%d %H:%M:%S"))
logger.addHandler(handler)
logger.setLevel(logging.INFO)
logger.propagate = False
return logger
_FILE_LOGGER = _build_file_logger()
def log(msg: str) -> None:
line = f"[{time.strftime('%H:%M:%S')}] {msg}"
# Under pythonw sys.stdout is None and print() would raise — guard it so a
# missing console can never crash the daemon thread (the silent-freeze mode).
try:
print(line, flush=True)
except (OSError, ValueError, AttributeError, RuntimeError):
pass
if _FILE_LOGGER is not None:
_FILE_LOGGER.info(msg)
class AuthError(Exception):
"""Raised by poll_api on a genuine 401/403 — the token really is expired or
invalid and the user must re-run `claude login`. Distinct from a None return,
which means a TRANSIENT failure (network/DNS, timeout, rate-limit, 5xx) that
must NOT be mislabeled as a token problem (SC#5: a boot-time `getaddrinfo
failed` DNS blip wrongly fired the 'token expired' toast)."""
async def poll_api(token: str) -> dict | None:
headers = dict(API_HEADERS_TEMPLATE)
headers["Authorization"] = f"Bearer {token}"
try:
async with httpx.AsyncClient(timeout=20.0) as http:
resp = await http.post(API_URL, headers=headers, json=API_BODY)
except httpx.HTTPError as e:
# Network/DNS/timeout — transient. Return None (no toast), retry next tick.
log(f"API call failed: {e}")
return None
if resp.status_code in (401, 403):
# Genuine auth rejection — the ONLY case that warrants the actionable
# "run claude login" toast.
log(f"API HTTP {resp.status_code}: {resp.text[:200]}")
raise AuthError(resp.status_code)
if resp.status_code >= 400:
# Other 4xx/5xx (rate-limit, server error) — transient, not a token issue.
log(f"API HTTP {resp.status_code}: {resp.text[:200]}")
return None
def hdr(name: str, default: str = "0") -> str:
return resp.headers.get(name, default)
now = time.time()
def reset_minutes(reset_ts: str) -> int:
try:
r = float(reset_ts)
except ValueError:
return 0
mins = (r - now) / 60.0
return int(round(mins)) if mins > 0 else 0
def pct(util: str) -> int:
try:
return int(round(float(util) * 100))
except ValueError:
return 0
payload = {
"s": pct(hdr("anthropic-ratelimit-unified-5h-utilization")),
"sr": reset_minutes(hdr("anthropic-ratelimit-unified-5h-reset")),
"w": pct(hdr("anthropic-ratelimit-unified-7d-utilization")),
"wr": reset_minutes(hdr("anthropic-ratelimit-unified-7d-reset")),
"st": hdr("anthropic-ratelimit-unified-5h-status", "unknown"),
"ok": True,
}
return payload
async def scan_for_device():
"""Scan for DEVICE_NAME and return the BLEDevice, or None."""
log(f"Scanning for '{DEVICE_NAME}' ({SCAN_TIMEOUT}s)...")
device = await BleakScanner.find_device_by_name(DEVICE_NAME, timeout=SCAN_TIMEOUT)
if device:
log(f"Found: {device.address}")
return device # BLEDevice or None — NOT an address string
class Session:
def __init__(self, client: BleakClient) -> None:
self.client = client
self.refresh_requested = asyncio.Event()
def _on_refresh(self, _char, _data: bytearray) -> None:
log("Refresh requested by device")
self.refresh_requested.set()
async def setup_refresh_subscription(self) -> None:
# The refresh subscription is optional — the 60s poll loop works without it.
# WinRT's start_notify() CCCD write can raise a raw OSError/WinError (not
# wrapped as BleakError) when the peer GATT server is transiently unavailable,
# e.g. a just-power-cycled ESP32 whose server is not yet ready (G-03-01, SC#3).
# Degrade gracefully instead of crashing the daemon so it stays single-process
# across a power-cycle reconnect (SC#4, no restart).
try:
await self.client.start_notify(REQ_CHAR_UUID, self._on_refresh)
except (BleakError, ValueError, OSError) as e:
log(f"Refresh subscription unavailable: {e}")
async def write_payload(self, payload: dict) -> bool:
data = json.dumps(payload, separators=(",", ":")).encode()
log(f"Sending: {data.decode()}")
try:
await self.client.write_gatt_char(RX_CHAR_UUID, data, response=False)
return True
except (BleakError, OSError) as e:
# WinRT can raise a raw OSError/WinError (NOT wrapped as BleakError)
# when the peer GATT server goes transiently unavailable mid-write —
# the same failure class setup_refresh_subscription() guards against.
# Returning False trips the zombie-link break -> clean reconnect,
# rather than an uncaught exception killing the daemon thread (the
# silent-freeze failure mode, SC#2 field report).
log(f"Write failed: {e}")
return False
def _extract_access_token(blob: str) -> str | None:
"""Pull the accessToken out of a credentials blob.
Claude Code stores credentials as a JSON object; the blob may also be
nested ({"claudeAiOauth": {"accessToken": "..."}}). Fall back to a
regex match so unexpected shapes still work, and finally treat the
blob as a raw token if nothing else matches.
"""
blob = blob.strip()
if not blob:
return None
try:
data = json.loads(blob)
except json.JSONDecodeError:
data = None
if isinstance(data, dict):
# direct: {"accessToken": "..."}
tok = data.get("accessToken")
if isinstance(tok, str) and tok.strip():
return tok
# nested: {"claudeAiOauth": {"accessToken": "..."}}
for v in data.values():
if isinstance(v, dict):
tok = v.get("accessToken")
if isinstance(tok, str) and tok.strip():
return tok
m = re.search(r'"accessToken"\s*:\s*"([^"]+)"', blob)
if m:
return m.group(1)
# Raw token (no JSON wrapper) — must look plausible (sk-ant-... etc.)
if re.fullmatch(r"[A-Za-z0-9_\-.~+/=]{20,}", blob):
return blob
return None
def _windows_credential_candidates() -> list[Path]:
"""Return the ordered list of credential file paths to probe (first hit wins).
Priority:
1. CLAUDE_CREDENTIALS_PATH env override (D-03, project-specific)
2. CLAUDE_CONFIG_DIR env override (official Claude override)
3. D-02 candidate list: home/.claude, LOCALAPPDATA/Claude, APPDATA/Claude
"""
# Priority 1: project-specific env override (D-03)
if override := os.environ.get("CLAUDE_CREDENTIALS_PATH"):
return [Path(override)]
# Priority 2: official CLAUDE_CONFIG_DIR env override
if config_dir := os.environ.get("CLAUDE_CONFIG_DIR"):
return [Path(config_dir) / ".credentials.json"]
# Priority 3: D-02 candidate list — first hit wins
home = Path.home()
local_appdata = Path(os.environ.get("LOCALAPPDATA", home / "AppData" / "Local"))
appdata = Path(os.environ.get("APPDATA", home / "AppData" / "Roaming"))
return [
home / ".claude" / ".credentials.json", # primary (confirmed by docs)
local_appdata / "Claude" / ".credentials.json", # fallback 2
appdata / "Claude" / ".credentials.json", # fallback 3
]
def read_token() -> str | None:
"""Read the Claude OAuth access token from the first available credential file."""
for path in _windows_credential_candidates():
try:
return _extract_access_token(path.read_text(encoding="utf-8"))
except OSError:
continue
return None
def _read_expiry() -> str:
"""Return human-readable expiry from the first-hit credentials file.
Reads claudeAiOauth.expiresAt (epoch milliseconds — JS convention).
Divides by 1000 before passing to fromtimestamp (Python expects seconds).
Returns 'expiry unknown' on any parse failure.
"""
for path in _windows_credential_candidates():
try:
raw = path.read_text(encoding="utf-8")
except OSError:
continue
try:
data = json.loads(raw)
oauth = data.get("claudeAiOauth", {})
expires_ms = oauth.get("expiresAt")
if expires_ms is None:
return "expiry unknown"
# CRITICAL: expiresAt is JS-convention epoch milliseconds; divide by 1000
# before fromtimestamp (Python expects seconds). Raw value -> year ~57000.
dt = datetime.datetime.fromtimestamp(
expires_ms / 1000, tz=datetime.timezone.utc
)
return dt.strftime("%Y-%m-%d %H:%M UTC")
except (TypeError, ValueError, OSError, AttributeError, json.JSONDecodeError):
return "expiry unknown"
return "expiry unknown"
async def _wait_first(*events: asyncio.Event, timeout: float) -> None:
"""Return when any of `events` is set, or after `timeout` seconds.
Lets the poll loop's TICK wait wake immediately on a stop signal (clean,
responsive Quit) without losing the refresh-request wakeup — instead of
waiting only on refresh_requested and re-checking stop_event up to TICK
later. Cancels and drains the loser tasks so they don't warn.
"""
tasks = [asyncio.ensure_future(e.wait()) for e in events]
try:
await asyncio.wait(tasks, timeout=timeout, return_when=asyncio.FIRST_COMPLETED)
finally:
for t in tasks:
t.cancel()
await asyncio.gather(*tasks, return_exceptions=True)
async def connect_and_run(device, stop_event: asyncio.Event, tray_state=None) -> bool:
"""Connect to device and poll until disconnected or stopped.
Returns True if at least one successful write occurred.
"""
log(f"Connecting to {device.address}...")
# D-01: retry wrapper — defeats WinRT post-wake failure modes
# (Could not get GATT services: Unreachable, stale is_connected).
# Rebuild a fresh BleakClient each attempt (locked D-05 recipe).
client = None
for attempt in range(CONNECT_RETRIES):
# D-05: pass BLEDevice (not address string), address_type="random" (NimBLE
# static-random), use_cached_services=False (DIY firmware — WinRT GATT cache
# may be stale after firmware reflash).
client = BleakClient(
device,
address_type="random",
use_cached_services=False,
)
try:
await client.connect()
except (BleakError, asyncio.TimeoutError) as e:
log(f"Connection attempt {attempt + 1}/{CONNECT_RETRIES} failed: {e}")
try:
await client.disconnect()
except BleakError:
pass
if attempt < CONNECT_RETRIES - 1:
await asyncio.sleep(CONNECT_RETRY_DELAY)
continue
if not client.is_connected:
log(f"Connection attempt {attempt + 1}/{CONNECT_RETRIES} failed (not connected)")
try:
await client.disconnect()
except BleakError:
pass
if attempt < CONNECT_RETRIES - 1:
await asyncio.sleep(CONNECT_RETRY_DELAY)
continue
# Connected successfully
break
else:
log(f"Connection failed after {CONNECT_RETRIES} attempts")
return False
log("Connected")
session = Session(client)
await session.setup_refresh_subscription()
last_poll = 0.0 # D-03: poll immediately on first connect
used_successfully = False
consecutive_failures = 0 # D-03: zombie-link break counter
try:
while client.is_connected and not stop_event.is_set():
now = time.time()
elapsed = now - last_poll
if session.refresh_requested.is_set() or elapsed >= POLL_INTERVAL:
session.refresh_requested.clear()
token = read_token() # D-09: fresh each cycle
if not token:
log("No token; skipping poll")
if tray_state:
tray_state.set_error("token expired — run claude login")
else:
try:
payload = await poll_api(token)
except AuthError:
# Real 401/403 — token genuinely needs a refresh.
if tray_state:
tray_state.set_error("token expired — run claude login")
payload = None
if payload is not None:
if await session.write_payload(payload):
last_poll = time.time()
used_successfully = True
consecutive_failures = 0 # D-03: reset on success
if tray_state:
tray_state.set_connected(time.time())
else:
consecutive_failures += 1
if consecutive_failures >= ZOMBIE_BREAK_LIMIT:
log(
f"Zombie link detected ({consecutive_failures} consecutive"
f" write failures); abandoning connection"
)
break
# else: payload is None from a TRANSIENT failure (network/DNS,
# timeout, rate-limit, 5xx). poll_api already logged it; do NOT
# toast "token expired" — that mislabeled a boot-time DNS blip
# as an auth problem (SC#5). Leave tray state unchanged; the next
# tick retries and set_connected() recovers it.
# Wake on a refresh request OR a stop, whichever comes first. Waking
# promptly on stop_event is what lets the finally below run
# client.disconnect() before the process exits, so the peer gets a
# clean GATT disconnect (returns to its waiting screen) instead of
# being left frozen on stale data after Quit (SC#3 graceful shutdown).
await _wait_first(session.refresh_requested, stop_event, timeout=TICK)
finally:
# Clean GATT disconnect on the way out — this is what tells the peripheral
# the link is gone. WinRT can surface a raw OSError (not BleakError) here,
# so swallow both; the link tears down regardless once we exit.
try:
await client.disconnect()
except (BleakError, OSError):
pass
log("Device disconnected" if not stop_event.is_set() else "Stopping")
return used_successfully
def _next_backoff(current: int, cap: int) -> int:
"""D-05: double current backoff value, clamped to cap.
Pure helper — unit-testable without driving the main loop.
Used by both slow-search (cap=60) and fast-reconnect (cap=RECONNECT_BACKOFF_CAP) regimes.
"""
return min(current * 2, cap)
async def main(tray_state=None) -> None:
stop_event = asyncio.Event()
loop = asyncio.get_running_loop()
# Populate the shared state object so the tray can route Quit through
# loop.call_soon_threadsafe (RESEARCH Pitfall 2). Additive — the existing
# stop_event = asyncio.Event() line above is unchanged.
if tray_state is not None:
tray_state.loop = loop
tray_state.stop_event = stop_event
def _stop(*_args: object) -> None:
log("Daemon stopping")
stop_event.set()
# OS signal handlers can only be installed from the main thread, and
# loop.add_signal_handler is unsupported on Windows. When running under the
# tray (04-03) the loop lives in a background thread and the tray owns clean
# shutdown via stop_event (loop.call_soon_threadsafe), so skip silently there.
if threading.current_thread() is threading.main_thread():
for sig in (signal.SIGINT, signal.SIGTERM):
try:
loop.add_signal_handler(sig, _stop)
except NotImplementedError:
# Windows: add_signal_handler not supported; fall back to signal.signal
try:
signal.signal(sig, _stop)
except ValueError:
# Not the main thread of the main interpreter — tray owns shutdown.
pass
log("=== Claude Usage Tracker Daemon (BLE, Windows) ===")
log(f"Poll interval: {POLL_INTERVAL}s")
# D-05: two distinct backoff regimes — slow-search (device absent) vs fast-reconnect (link dropped)
search_backoff = 1 # caps at 60s — gentle, for a device that is genuinely absent/off
reconnect_backoff = 1 # caps at RECONNECT_BACKOFF_CAP — fast, to clear the 120s SLA after a drop
while not stop_event.is_set():
device = await scan_for_device()
if not device:
# Slow-search regime: device was not found by scan — back off gently
if tray_state:
tray_state.set_scanning()
log(f"Device not found, retrying in {search_backoff}s...")
try:
await asyncio.wait_for(stop_event.wait(), timeout=search_backoff)
except asyncio.TimeoutError:
pass
search_backoff = _next_backoff(search_backoff, 60)
continue
ok = await connect_and_run(device, stop_event, tray_state)
if not ok:
# Fast-reconnect regime: had/attempted a link that dropped — retry quickly
if tray_state:
tray_state.set_scanning()
log(f"Connection lost, reconnecting in {reconnect_backoff}s...")
try:
await asyncio.wait_for(stop_event.wait(), timeout=reconnect_backoff)
except asyncio.TimeoutError:
pass
reconnect_backoff = _next_backoff(reconnect_backoff, RECONNECT_BACKOFF_CAP)
else:
# Successful session — reset reconnect counter to floor; search_backoff also reset
reconnect_backoff = 1
search_backoff = 1
if __name__ == "__main__":
if sys.platform != "win32":
print(
"Warning: running under Linux/WSL — WinRT BLE will not be available.",
file=sys.stderr,
)
try:
asyncio.run(main())
except KeyboardInterrupt:
sys.exit(0)
+156
View File
@@ -0,0 +1,156 @@
#!/usr/bin/env python3
"""Icon asset layer for the Windows tray app.
Parses the firmware/src/logo.h RGB565A8 brand logo into a Pillow RGBA image,
expands RGB565->RGB888 with correct rounding, and composites per-state corner
bubbles (green=connected / amber=scanning / red=error) onto the constant brand
mark. All logic is pure and unit-testable off-Windows (Pillow only, no pystray
or winreg here).
Usage::
from daemon.icon_assets import load_logo_rgba, build_state_icons
base = load_logo_rgba("firmware/src/logo.h")
icons = build_state_icons(base) # dict: "connected"/"scanning"/"error" -> Image
"""
import re
from PIL import Image, ImageDraw
# Logo dimensions from logo.h #defines.
W: int = 80
H: int = 80
# Brand hex derived from the dominant opaque RGB565 color in logo.h (D-03).
# 0xDBAA -> RGB888 (222, 117, 82) -> #DE7552
BRAND_HEX: str = "#DE7552"
# Locked corner-bubble RGBA colors (from RESEARCH, verified end-to-end).
BUBBLE: dict[str, tuple[int, int, int, int]] = {
"connected": (60, 200, 90, 255), # green
"scanning": (240, 180, 40, 255), # amber
"error": (220, 60, 60, 255), # red
}
def _expand565(v: int) -> tuple[int, int, int]:
"""Expand a 16-bit RGB565 value to an (R, G, B) tuple using proper rounding.
Uses ``(channel * 255 + max // 2) // max`` per channel, NOT a *8 bit-shift.
A *8 shift loses the low bits and does not correctly round to 255 for 0xFFFF.
Examples::
_expand565(0x0000) == (0, 0, 0)
_expand565(0xFFFF) == (255, 255, 255)
_expand565(0xDBAA) == (222, 117, 82) # brand hex
"""
r5 = (v >> 11) & 0x1F # 5-bit red channel (max 31)
g6 = (v >> 5) & 0x3F # 6-bit green channel (max 63)
b5 = v & 0x1F # 5-bit blue channel (max 31)
r = (r5 * 255 + 15) // 31
g = (g6 * 255 + 31) // 63
b = (b5 * 255 + 15) // 31
return (r, g, b)
def load_logo_rgba(header_path: str) -> Image.Image:
"""Parse the firmware logo.h C header and return an 80x80 Pillow RGBA Image.
The logo.h layout (RGB565A8 planar, little-endian RGB565):
- First ``W * H * 2`` bytes: little-endian RGB565 pixel data
- Next ``W * H`` bytes: 8-bit alpha plane
Args:
header_path: Path to ``firmware/src/logo.h`` (or any compatible header).
Returns:
An ``Image.Image`` of mode ``"RGBA"`` and size ``(W, H)``.
Raises:
ValueError: If the extracted byte array length != ``W * H * 3`` (ASVS V5
bound-check before indexing).
"""
with open(header_path) as f:
txt = f.read()
# Extract the byte array body from: logo_data[N] = { 0xFF, 0xAA, ... };
match = re.search(r'logo_data\[\d+\]\s*=\s*\{(.*?)\};', txt, re.S)
if not match:
raise ValueError(f"Could not find logo_data[] array in {header_path!r}")
body = match.group(1)
raw_bytes = [int(x, 16) for x in re.findall(r'0x([0-9A-Fa-f]{2})', body)]
# Bound-check before indexing (ASVS V5).
expected = W * H * 3 # W*H*2 RGB565 bytes + W*H alpha bytes = 19200
if len(raw_bytes) != expected:
raise ValueError(
f"logo_data byte count mismatch: expected {expected}, got {len(raw_bytes)}"
)
n = W * H
rgb_bytes = raw_bytes[:n * 2] # first 12800 bytes: little-endian RGB565
alpha_bytes = raw_bytes[n * 2:] # last 6400 bytes: 8-bit alpha
img = Image.new("RGBA", (W, H))
px = img.load()
for i in range(n):
# Little-endian: low byte first, high byte second.
v = rgb_bytes[i * 2] | (rgb_bytes[i * 2 + 1] << 8)
r, g, b = _expand565(v)
px[i % W, i // W] = (r, g, b, alpha_bytes[i])
return img
def state_icon(base: Image.Image, state: str, size: int = 32) -> Image.Image:
"""Composite a colored corner bubble onto the brand mark for the given state.
Args:
base: The RGBA brand image (as returned by ``load_logo_rgba``).
state: One of ``"connected"``, ``"scanning"``, or ``"error"``.
Any other value raises ``KeyError`` — no silent fallback.
size: Target icon edge length in pixels (default 32).
Returns:
A new ``Image.Image`` of mode ``"RGBA"`` and size ``(size, size)``.
Raises:
KeyError: If ``state`` is not one of the three known states.
"""
# Raises KeyError on unknown state — no silent default (per plan anti-pattern).
bubble_color = BUBBLE[state]
icon = base.resize((size, size), Image.LANCZOS).convert("RGBA")
draw = ImageDraw.Draw(icon)
# Corner bubble: ~1/3 of the icon, drawn in the bottom-right corner.
r = size // 3
x0 = size - r - 1
y0 = size - r - 1
x1 = size - 2
y1 = size - 2
draw.ellipse([x0, y0, x1, y1], fill=bubble_color)
return icon
def build_state_icons(
base: Image.Image,
size: int = 32,
) -> dict[str, Image.Image]:
"""Build all three connection-state icons from the brand base image.
Build them once at startup; swap ``icon.icon = icons[state]`` in the tray
loop — never recomposite per tick (per RESEARCH anti-pattern).
Args:
base: The RGBA brand image (as returned by ``load_logo_rgba``).
size: Target icon edge length in pixels (default 32).
Returns:
A dict mapping ``"connected"``, ``"scanning"``, and ``"error"`` to
their respective composited ``Image.Image`` objects.
"""
return {state: state_icon(base, state, size) for state in BUBBLE}
+6
View File
@@ -0,0 +1,6 @@
# Windows-only dependency manifest for claude_usage_daemon_windows.py
# The macOS/Linux daemons manage their own dependencies separately.
bleak
httpx
pystray
Pillow
View File
+3
View File
@@ -0,0 +1,3 @@
{
"accessToken": "sk-ant-test-5678"
}
+8
View File
@@ -0,0 +1,8 @@
{
"claudeAiOauth": {
"accessToken": "sk-ant-test-1234",
"refreshToken": "sk-ant-ort-test-5678",
"expiresAt": 9999999999000,
"scopes": ["user:inference", "user:profile"]
}
}
+172
View File
@@ -0,0 +1,172 @@
#!/usr/bin/env python3
"""Unit tests for daemon/autostart_windows.py — APP-01.
Covers the winreg HKCU\\Run enable/disable/is_enabled login-autostart toggle.
winreg is NOT importable off-Windows; these tests patch it via
patch("daemon.autostart_windows.winreg", ...) so they run on any platform.
Run: python -m pytest daemon/tests/test_windows_autostart.py -x -q
"""
from unittest.mock import MagicMock, patch, call
import pytest
# ---------------------------------------------------------------------------
# Helpers — build a fake winreg module with the attributes autostart_windows
# references. Using a MagicMock as the module means all attribute accesses
# on it (HKEY_CURRENT_USER, KEY_SET_VALUE, etc.) automatically produce child
# MagicMocks, which is exactly what we want.
# ---------------------------------------------------------------------------
def _make_winreg_mock(*, query_raises=False):
"""Return a configured MagicMock that stands in for the winreg module."""
winreg = MagicMock()
# Constants — assign simple sentinel values so equality checks work.
winreg.HKEY_CURRENT_USER = "HKEY_CURRENT_USER"
winreg.KEY_SET_VALUE = 0x0002
winreg.KEY_QUERY_VALUE = 0x0001
winreg.REG_SZ = 1
# OpenKey is used as a context manager; return a MagicMock key handle that
# supports __enter__ / __exit__.
key_handle = MagicMock()
key_handle.__enter__ = MagicMock(return_value=key_handle)
key_handle.__exit__ = MagicMock(return_value=False)
winreg.OpenKey = MagicMock(return_value=key_handle)
# QueryValueEx behaviour is configured by the caller.
if query_raises:
winreg.QueryValueEx = MagicMock(side_effect=FileNotFoundError("not found"))
else:
winreg.QueryValueEx = MagicMock(return_value=("some_command", 1))
return winreg, key_handle
# ---------------------------------------------------------------------------
# test_enable_writes_run_value
# ---------------------------------------------------------------------------
def test_enable_writes_run_value():
"""enable() opens HKCU Run key with KEY_SET_VALUE and calls SetValueEx with
value name 'Clawdmeter' and type REG_SZ."""
winreg, key_handle = _make_winreg_mock()
with patch("daemon.autostart_windows.winreg", winreg):
import daemon.autostart_windows as mod
mod.enable()
# OpenKey must have been called with HKCU and the Run key path
winreg.OpenKey.assert_called_once_with(
winreg.HKEY_CURRENT_USER,
r"Software\Microsoft\Windows\CurrentVersion\Run",
0,
winreg.KEY_SET_VALUE,
)
# SetValueEx must have been called with the correct value name and type
winreg.SetValueEx.assert_called_once()
args = winreg.SetValueEx.call_args[0]
assert args[0] is key_handle, "SetValueEx first arg must be the opened key handle"
assert args[1] == "Clawdmeter", "Value name must be 'Clawdmeter'"
assert args[3] == winreg.REG_SZ, "Value type must be REG_SZ"
# ---------------------------------------------------------------------------
# test_command_uses_pythonw
# ---------------------------------------------------------------------------
def test_command_uses_pythonw():
"""The command string written by enable() contains 'pythonw.exe', does NOT
contain a bare 'python.exe' token (D-08, no console), and is quoted (starts
with a double-quote character)."""
winreg, key_handle = _make_winreg_mock()
captured_commands = []
def capture_set_value_ex(key, name, reserved, reg_type, value):
captured_commands.append(value)
winreg.SetValueEx = MagicMock(side_effect=capture_set_value_ex)
with patch("daemon.autostart_windows.winreg", winreg):
import daemon.autostart_windows as mod
mod.enable()
assert len(captured_commands) == 1, "SetValueEx must have been called exactly once"
cmd = captured_commands[0]
# Must reference pythonw.exe (D-08)
assert "pythonw.exe" in cmd, f"Command must contain 'pythonw.exe'; got: {cmd!r}"
# Must NOT contain a bare 'python.exe' (without the 'w') as a standalone token
# A command like '"...pythonw.exe" ...' is fine; '"...python.exe" ...' is not.
import re
assert not re.search(r'(?<![a-z])python\.exe', cmd), (
f"Command must not reference a bare 'python.exe'; got: {cmd!r}"
)
# Must start with a double-quote (paths are quoted for space safety)
assert cmd.startswith('"'), (
f"Command must start with '\"' (quoted path); got: {cmd!r}"
)
# ---------------------------------------------------------------------------
# test_disable_idempotent
# ---------------------------------------------------------------------------
def test_disable_idempotent():
"""disable() calls DeleteValue; when DeleteValue raises FileNotFoundError,
disable() swallows it and returns without raising (idempotent-on-missing)."""
winreg, key_handle = _make_winreg_mock()
winreg.DeleteValue = MagicMock(side_effect=FileNotFoundError("not found"))
with patch("daemon.autostart_windows.winreg", winreg):
import daemon.autostart_windows as mod
# Must not raise even though DeleteValue raises FileNotFoundError
mod.disable() # no exception expected
winreg.DeleteValue.assert_called_once()
def test_disable_calls_delete_value_with_correct_name():
"""disable() calls DeleteValue with the value name 'Clawdmeter'."""
winreg, key_handle = _make_winreg_mock()
winreg.DeleteValue = MagicMock()
with patch("daemon.autostart_windows.winreg", winreg):
import daemon.autostart_windows as mod
mod.disable()
winreg.DeleteValue.assert_called_once()
args = winreg.DeleteValue.call_args[0]
assert args[1] == "Clawdmeter", f"DeleteValue must target 'Clawdmeter'; got {args[1]!r}"
# ---------------------------------------------------------------------------
# test_is_enabled
# ---------------------------------------------------------------------------
def test_is_enabled_true_when_value_present():
"""is_enabled() returns True when QueryValueEx succeeds (value is present)."""
winreg, key_handle = _make_winreg_mock(query_raises=False)
with patch("daemon.autostart_windows.winreg", winreg):
import daemon.autostart_windows as mod
result = mod.is_enabled()
assert result is True, "is_enabled() must return True when QueryValueEx succeeds"
def test_is_enabled_false_when_value_absent():
"""is_enabled() returns False when QueryValueEx raises FileNotFoundError."""
winreg, key_handle = _make_winreg_mock(query_raises=True)
with patch("daemon.autostart_windows.winreg", winreg):
import daemon.autostart_windows as mod
result = mod.is_enabled()
assert result is False, "is_enabled() must return False when QueryValueEx raises FileNotFoundError"
+162
View File
@@ -0,0 +1,162 @@
#!/usr/bin/env python3
"""Unit tests for daemon/icon_assets.py — APP-01 tray icon asset layer.
Run: python -m pytest daemon/tests/test_windows_icon.py -x -q
"""
from pathlib import Path
import pytest
from daemon.icon_assets import _expand565, load_logo_rgba
# The "fixture" is the real in-repo firmware logo header — trusted asset.
LOGO_H = Path(__file__).parent.parent.parent / "firmware" / "src" / "logo.h"
# ---------------------------------------------------------------------------
# Task 1: logo parse + RGB565->RGB888 expand
# ---------------------------------------------------------------------------
def test_logo_parse():
"""load_logo_rgba returns an 80x80 RGBA image; dominant opaque color is #DE7552."""
img = load_logo_rgba(str(LOGO_H))
assert img.mode == "RGBA", f"Expected RGBA, got {img.mode}"
assert img.size == (80, 80), f"Expected (80,80), got {img.size}"
# Collect all fully-opaque pixels and find the dominant RGB.
from collections import Counter
px = img.load()
opaque = [
(px[x, y][0], px[x, y][1], px[x, y][2])
for y in range(80)
for x in range(80)
if px[x, y][3] == 255
]
assert opaque, "No fully-opaque pixels found in logo"
dominant = Counter(opaque).most_common(1)[0][0]
# Brand hex #DE7552 = (222, 117, 82)
assert dominant == (222, 117, 82), (
f"Expected dominant opaque color (222,117,82), got {dominant}"
)
def test_rgb565_expand():
"""_expand565 uses proper rounding, not a *8 bit-shift."""
assert _expand565(0xDBAA) == (222, 117, 82), (
f"0xDBAA should be (222,117,82), got {_expand565(0xDBAA)}"
)
assert _expand565(0x0000) == (0, 0, 0), (
f"0x0000 should be (0,0,0), got {_expand565(0x0000)}"
)
assert _expand565(0xFFFF) == (255, 255, 255), (
f"0xFFFF should be (255,255,255), got {_expand565(0xFFFF)}"
)
def test_logo_parse_bounds_check():
"""load_logo_rgba raises ValueError if the data array length != W*H*3."""
import tempfile, os
# Write a malformed header with fewer bytes than expected
malformed = (
"#pragma once\n"
"static const uint8_t logo_data[100] = {\n"
" 0x00, 0x01, 0x02\n"
"};\n"
)
with tempfile.NamedTemporaryFile(mode="w", suffix=".h", delete=False) as f:
f.write(malformed)
path = f.name
try:
with pytest.raises(ValueError):
load_logo_rgba(path)
finally:
os.unlink(path)
# ---------------------------------------------------------------------------
# Task 2: state->image corner-bubble compositor
# ---------------------------------------------------------------------------
def test_state_icon_bubble():
"""state_icon returns distinct 32x32 RGBA images per state with correct bubble color."""
from daemon.icon_assets import state_icon
base = load_logo_rgba(str(LOGO_H))
connected = state_icon(base, "connected", 32)
scanning = state_icon(base, "scanning", 32)
error = state_icon(base, "error", 32)
# All are 32x32 RGBA
for name, img in [("connected", connected), ("scanning", scanning), ("error", error)]:
assert img.mode == "RGBA", f"{name}: expected RGBA, got {img.mode}"
assert img.size == (32, 32), f"{name}: expected (32,32), got {img.size}"
# The three images must be pixel-distinct from each other
def img_bytes(img):
return img.tobytes()
assert img_bytes(connected) != img_bytes(scanning), "connected and scanning should differ"
assert img_bytes(connected) != img_bytes(error), "connected and error should differ"
assert img_bytes(scanning) != img_bytes(error), "scanning and error should differ"
# Sample the bottom-right corner region; the closest bubble color should match the state.
# Bubble colors: connected (60,200,90), scanning (240,180,40), error (220,60,60)
BUBBLE_COLORS = {
"connected": (60, 200, 90),
"scanning": (240, 180, 40),
"error": (220, 60, 60),
}
def color_distance(c1, c2):
return sum((a - b) ** 2 for a, b in zip(c1, c2)) ** 0.5
def nearest_bubble(pixel_rgb):
return min(BUBBLE_COLORS.items(), key=lambda kv: color_distance(pixel_rgb, kv[1]))[0]
size = 32
r = size // 3
# Sample the center of the expected bubble region (bottom-right corner)
bx = size - r // 2 - 2
by = size - r // 2 - 2
bx = max(0, min(bx, size - 1))
by = max(0, min(by, size - 1))
for state_name, img in [("connected", connected), ("scanning", scanning), ("error", error)]:
px = img.load()
pixel = px[bx, by][:3] # RGB only
nearest = nearest_bubble(pixel)
assert nearest == state_name, (
f"State '{state_name}': bottom-right corner pixel {pixel} is nearest to "
f"'{nearest}' bubble, expected '{state_name}'"
)
def test_build_icons_once():
"""build_state_icons returns a dict with connected/scanning/error as distinct Images."""
from daemon.icon_assets import build_state_icons
base = load_logo_rgba(str(LOGO_H))
icons = build_state_icons(base)
assert set(icons.keys()) == {"connected", "scanning", "error"}, (
f"Expected keys connected/scanning/error, got {set(icons.keys())}"
)
# All distinct
def img_bytes(img):
return img.tobytes()
imgs = list(icons.values())
assert img_bytes(imgs[0]) != img_bytes(imgs[1])
assert img_bytes(imgs[0]) != img_bytes(imgs[2])
assert img_bytes(imgs[1]) != img_bytes(imgs[2])
def test_state_icon_unknown_state():
"""state_icon raises KeyError/ValueError on an unknown state string."""
from daemon.icon_assets import state_icon
base = load_logo_rgba(str(LOGO_H))
with pytest.raises((KeyError, ValueError)):
state_icon(base, "unknown_state", 32)
+55
View File
@@ -0,0 +1,55 @@
#!/usr/bin/env python3
"""Static no-WSL-paths regression guard — APP-02 / D-10.
Asserts that the daemon, tray, and autostart sources reference no WSL-specific
paths. This is a CI-surviving regression lock that needs no hardware: if a
future edit accidentally introduces a ``\\wsl$``, ``wsl.exe``, ``/home/``, or
``/mnt/`` reference into any of the three core Windows daemon source files, this
test will fail with a message that names the offending pattern and file.
Run: python -m pytest daemon/tests/test_windows_no_wsl.py -x -q
"""
import re
from pathlib import Path
# The four WSL-path patterns that must never appear in the daemon sources.
FORBIDDEN = [r"\\wsl\$", r"wsl\.exe", r"/home/", r"/mnt/"]
# The three Windows-daemon source files covered by the guard.
SOURCES = [
Path("daemon/claude_usage_daemon_windows.py"),
Path("daemon/tray_windows.py"),
Path("daemon/autostart_windows.py"),
]
def test_no_wsl_paths_in_daemon():
"""daemon/claude_usage_daemon_windows.py references no WSL paths."""
_assert_clean(Path("daemon/claude_usage_daemon_windows.py"))
def test_no_wsl_paths_in_tray():
"""daemon/tray_windows.py references no WSL paths."""
_assert_clean(Path("daemon/tray_windows.py"))
def test_no_wsl_paths_in_autostart():
"""daemon/autostart_windows.py references no WSL paths."""
_assert_clean(Path("daemon/autostart_windows.py"))
def _assert_clean(source: Path) -> None:
"""Assert that none of the FORBIDDEN patterns appear in the given source file.
Reads the file relative to the repository root (the cwd pytest is invoked
from). Fails with a descriptive message naming the leaked pattern and file
so the regression is immediately actionable.
"""
text = source.read_text(encoding="utf-8")
for pat in FORBIDDEN:
match = re.search(pat, text)
assert match is None, (
f"WSL path leaked into {source}: pattern {pat!r} found at "
f"position {match.start()}"
f"context: {text[max(0, match.start()-20):match.end()+20]!r}"
)
+455
View File
@@ -0,0 +1,455 @@
#!/usr/bin/env python3
"""Unit tests for poll_api / pct / reset_minutes / JSON-shape — POLL-01.
These tests cover the Anthropic API polling logic ported from the macOS daemon.
All tests mock httpx so no real network calls are made.
Run: python -m pytest daemon/tests/test_windows_poll.py -x -q
"""
import asyncio
import json
import time
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from daemon.claude_usage_daemon_windows import AuthError, poll_api
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _make_mock_response(status_code=200, headers=None):
"""Build a mock httpx.Response-like object with controllable headers."""
resp = MagicMock()
resp.status_code = status_code
resp.text = "mocked"
# httpx headers are case-insensitive; MagicMock .get() must behave the same
header_data = headers or {}
resp.headers = MagicMock()
resp.headers.get = lambda name, default=None: header_data.get(name.lower(), default)
return resp
def _run(coro):
"""Run a coroutine synchronously for synchronous test functions."""
return asyncio.get_event_loop().run_until_complete(coro)
# ---------------------------------------------------------------------------
# Test: full poll_api with realistic ratelimit headers
# ---------------------------------------------------------------------------
def test_poll_api_nominal(monkeypatch):
"""poll_api with a 200 response + ratelimit headers produces the correct payload."""
now = time.time()
reset_5h = str(now + 3600) # 60 minutes from now
reset_7d = str(now + 86400) # 1440 minutes from now
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0.42",
"anthropic-ratelimit-unified-5h-reset": reset_5h,
"anthropic-ratelimit-unified-7d-utilization": "0.10",
"anthropic-ratelimit-unified-7d-reset": reset_7d,
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
async def fake_post(*args, **kwargs):
return mock_resp
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = fake_post
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload is not None
assert payload["s"] == 42
assert payload["w"] == 10
assert payload["st"] == "allowed"
assert payload["ok"] is True
# reset_minutes allows ±1 minute tolerance
assert abs(payload["sr"] - 60) <= 1, f"Expected ~60, got {payload['sr']}"
assert abs(payload["wr"] - 1440) <= 1, f"Expected ~1440, got {payload['wr']}"
# ---------------------------------------------------------------------------
# Test: pct() correctness — exercised through poll_api output
# ---------------------------------------------------------------------------
def test_pct_42_percent(monkeypatch):
"""pct('0.42') -> 42."""
now = time.time()
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0.42",
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-7d-utilization": "0.10",
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload["s"] == 42
def test_pct_100_percent(monkeypatch):
"""pct('1.0') -> 100."""
now = time.time()
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "1.0",
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-7d-utilization": "1.0",
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload["s"] == 100
assert payload["w"] == 100
def test_pct_empty_string_defaults_to_zero(monkeypatch):
"""pct('') -> 0 (missing header defaults to '0', but empty string -> 0)."""
now = time.time()
# Override default so utilization header returns "" explicitly
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0",
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-7d-utilization": "0",
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload["s"] == 0
assert payload["w"] == 0
# ---------------------------------------------------------------------------
# Test: reset_minutes() — exercised through poll_api output
# ---------------------------------------------------------------------------
def test_reset_minutes_60_minutes(monkeypatch):
"""reset_minutes(now+3600) -> ~60."""
now = time.time()
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0.5",
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-7d-utilization": "0.5",
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert abs(payload["sr"] - 60) <= 1, f"Expected ~60, got {payload['sr']}"
def test_reset_minutes_negative_clamps_to_zero(monkeypatch):
"""reset_minutes for a past timestamp clamps to 0."""
now = time.time()
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0.5",
"anthropic-ratelimit-unified-5h-reset": str(now - 100), # 100s in the past
"anthropic-ratelimit-unified-7d-utilization": "0.5",
"anthropic-ratelimit-unified-7d-reset": str(now - 100), # also in the past
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload["sr"] == 0
assert payload["wr"] == 0
def test_reset_minutes_invalid_string_returns_zero(monkeypatch):
"""reset_minutes('notanumber') -> 0 (ValueError-safe)."""
now = time.time()
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0.5",
"anthropic-ratelimit-unified-5h-reset": "notanumber",
"anthropic-ratelimit-unified-7d-utilization": "0.5",
"anthropic-ratelimit-unified-7d-reset": "notanumber",
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload["sr"] == 0
assert payload["wr"] == 0
# ---------------------------------------------------------------------------
# Test: missing headers default gracefully
# ---------------------------------------------------------------------------
def test_missing_utilization_headers_default_to_zero(monkeypatch):
"""Missing utilization headers produce 0 (hdr default '0' -> pct('0') = 0)."""
now = time.time()
# No utilization or status headers — only reset headers present
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload["s"] == 0
assert payload["w"] == 0
def test_missing_status_header_defaults_to_unknown(monkeypatch):
"""Missing 5h-status header defaults to 'unknown'."""
now = time.time()
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0.5",
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-7d-utilization": "0.5",
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
# NOTE: no 5h-status header
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload["st"] == "unknown"
# ---------------------------------------------------------------------------
# Test: poll_api returns None on HTTP >= 400
# ---------------------------------------------------------------------------
def test_poll_api_returns_none_on_4xx(monkeypatch):
"""poll_api returns None when response status code is >= 400."""
mock_resp = _make_mock_response(status_code=429, headers={})
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
result = _run(poll_api("fake-token"))
assert result is None
def test_poll_api_returns_none_on_5xx(monkeypatch):
"""poll_api returns None when response status code is >= 500."""
mock_resp = _make_mock_response(status_code=500, headers={})
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
result = _run(poll_api("fake-token"))
assert result is None
# ---------------------------------------------------------------------------
# Test: poll_api raises AuthError ONLY on a genuine 401/403 (SC#5 fix)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize("status", [401, 403])
def test_poll_api_raises_autherror_on_401_403(status):
"""A real auth rejection must raise AuthError — the only signal that warrants
the actionable 'token expired — run claude login' toast. Transient failures
(5xx, 429, network) return None instead and must NOT trigger that toast."""
mock_resp = _make_mock_response(status_code=status, headers={})
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
with pytest.raises(AuthError):
_run(poll_api("fake-token"))
def test_poll_api_returns_none_not_autherror_on_429(monkeypatch):
"""Rate-limit (429) is transient — None, NOT AuthError (regression guard for
the 401/403-vs-other-4xx split)."""
mock_resp = _make_mock_response(status_code=429, headers={})
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
result = _run(poll_api("fake-token"))
assert result is None
# ---------------------------------------------------------------------------
# Test: poll_api returns None on httpx.HTTPError
# ---------------------------------------------------------------------------
def test_poll_api_returns_none_on_http_error(monkeypatch):
"""poll_api returns None when httpx.HTTPError is raised (network failure)."""
import httpx
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(side_effect=httpx.ConnectError("Connection refused"))
with patch("httpx.AsyncClient", return_value=mock_client):
result = _run(poll_api("fake-token"))
assert result is None
# ---------------------------------------------------------------------------
# Test: compact JSON wire shape (no spaces after ':' or ',')
# ---------------------------------------------------------------------------
def test_wire_bytes_compact_json_shape(monkeypatch):
"""The JSON-encoded payload uses compact separators (',':') — no spaces."""
now = time.time()
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0.42",
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-7d-utilization": "0.10",
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
payload = _run(poll_api("fake-token"))
assert payload is not None
# Encode exactly as the wire layer will (Session.write_payload uses this form)
wire_bytes = json.dumps(payload, separators=(",", ":")).encode()
wire_str = wire_bytes.decode()
# Compact form: no space after ':' or ','
assert ": " not in wire_str, f"Non-compact JSON detected: {wire_str!r}"
assert ", " not in wire_str, f"Non-compact JSON detected: {wire_str!r}"
# Must start with '{' and contain all required keys
assert wire_str.startswith("{")
for key in ("s", "sr", "w", "wr", "st", "ok"):
assert f'"{key}"' in wire_str, f"Missing key {key!r} in wire bytes: {wire_str!r}"
# ---------------------------------------------------------------------------
# Test: token is NOT logged (T-02-01 threat mitigation)
# ---------------------------------------------------------------------------
def test_poll_api_does_not_log_token(monkeypatch, capsys):
"""poll_api must not print the bearer token (T-02-01: token never logged)."""
now = time.time()
secret_token = "sk-ant-secret-token-12345"
mock_resp = _make_mock_response(
status_code=200,
headers={
"anthropic-ratelimit-unified-5h-utilization": "0.5",
"anthropic-ratelimit-unified-5h-reset": str(now + 3600),
"anthropic-ratelimit-unified-7d-utilization": "0.5",
"anthropic-ratelimit-unified-7d-reset": str(now + 86400),
"anthropic-ratelimit-unified-5h-status": "allowed",
},
)
mock_client = AsyncMock()
mock_client.__aenter__ = AsyncMock(return_value=mock_client)
mock_client.__aexit__ = AsyncMock(return_value=False)
mock_client.post = AsyncMock(return_value=mock_resp)
with patch("httpx.AsyncClient", return_value=mock_client):
_run(poll_api(secret_token))
captured = capsys.readouterr()
assert secret_token not in captured.out, "Token leaked to stdout (T-02-01 violation)"
assert secret_token not in captured.err, "Token leaked to stderr (T-02-01 violation)"
+723
View File
@@ -0,0 +1,723 @@
#!/usr/bin/env python3
"""Unit tests for connect_and_run reconnect hardening — BLE-03.
Covers:
D-01: connect-retry wrapper (post-wake WinRT failure modes)
D-03: zombie-link consecutive-failure break (stale is_connected)
Run: python -m pytest daemon/tests/test_windows_reconnect.py -x -q
"""
import asyncio
from pathlib import Path
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from bleak.exc import BleakError
from daemon.claude_usage_daemon_windows import (
AuthError,
Session,
_wait_first,
connect_and_run,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _run(coro):
"""Run a coroutine synchronously for synchronous test functions."""
return asyncio.run(coro)
def _make_device(address="AA:BB:CC:DD:EE:FF"):
"""Build a minimal fake BLEDevice."""
device = MagicMock()
device.address = address
return device
async def _make_event(set_):
ev = asyncio.Event()
if set_:
ev.set()
return ev
# ---------------------------------------------------------------------------
# D-01: connect-retry wrapper tests
# ---------------------------------------------------------------------------
def test_connect_retry_exhaustion_on_bleak_error(monkeypatch, capsys):
"""BleakError on every connect attempt exhausts CONNECT_RETRIES then returns False."""
import daemon.claude_usage_daemon_windows as mod
device = _make_device()
stop_event = asyncio.run(_make_event(False))
mock_client = AsyncMock()
mock_client.connect = AsyncMock(side_effect=BleakError("Unreachable"))
mock_client.is_connected = False
mock_client.disconnect = AsyncMock()
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.asyncio.sleep", new=AsyncMock()):
result = _run(connect_and_run(device, stop_event))
assert result is False
assert mock_client.connect.call_count == mod.CONNECT_RETRIES
def test_connect_retry_exhaustion_on_timeout_error(monkeypatch, capsys):
"""asyncio.TimeoutError on every connect attempt is treated same as BleakError."""
import daemon.claude_usage_daemon_windows as mod
device = _make_device()
stop_event = asyncio.run(_make_event(False))
mock_client = AsyncMock()
mock_client.connect = AsyncMock(side_effect=asyncio.TimeoutError())
mock_client.is_connected = False
mock_client.disconnect = AsyncMock()
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.asyncio.sleep", new=AsyncMock()):
result = _run(connect_and_run(device, stop_event))
assert result is False
assert mock_client.connect.call_count == mod.CONNECT_RETRIES
def test_connect_retry_calls_disconnect_between_attempts(monkeypatch):
"""Guarded disconnect() is called between failed connect attempts."""
import daemon.claude_usage_daemon_windows as mod
device = _make_device()
stop_event = asyncio.run(_make_event(False))
mock_client = AsyncMock()
mock_client.connect = AsyncMock(side_effect=BleakError("Unreachable"))
mock_client.is_connected = False
mock_client.disconnect = AsyncMock()
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.asyncio.sleep", new=AsyncMock()):
_run(connect_and_run(device, stop_event))
# disconnect is called between attempts (at least CONNECT_RETRIES - 1 times)
assert mock_client.disconnect.call_count >= mod.CONNECT_RETRIES - 1
def test_connect_success_on_first_attempt_no_extra_retries(monkeypatch):
"""First-attempt success consumes exactly 1 connect call and proceeds past connect block."""
import daemon.claude_usage_daemon_windows as mod
device = _make_device()
# stop_event is set so the loop exits immediately after connecting
stop_event = asyncio.run(_make_event(True))
mock_client = AsyncMock()
mock_client.connect = AsyncMock(return_value=None) # success
mock_client.is_connected = True
mock_client.disconnect = AsyncMock()
mock_client.start_notify = AsyncMock()
mock_client.write_gatt_char = AsyncMock(return_value=None)
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
patch("daemon.claude_usage_daemon_windows.poll_api", new=AsyncMock(return_value={"ok": True})):
_run(connect_and_run(device, stop_event))
assert mock_client.connect.call_count == 1
def test_connect_retry_exhaustion_does_not_log_token(monkeypatch, capsys):
"""On exhaustion, no log line contains the patched token sentinel (T-03-01)."""
import daemon.claude_usage_daemon_windows as mod
TOKEN_SENTINEL = "sk-ant-SUPERSECRET-DO-NOT-LOG-12345"
device = _make_device()
stop_event = asyncio.run(_make_event(False))
mock_client = AsyncMock()
mock_client.connect = AsyncMock(side_effect=BleakError("Unreachable"))
mock_client.is_connected = False
mock_client.disconnect = AsyncMock()
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value=TOKEN_SENTINEL), \
patch("daemon.claude_usage_daemon_windows.asyncio.sleep", new=AsyncMock()):
_run(connect_and_run(device, stop_event))
captured = capsys.readouterr()
assert TOKEN_SENTINEL not in captured.out, "Token sentinel leaked to stdout (T-03-01)"
assert TOKEN_SENTINEL not in captured.err, "Token sentinel leaked to stderr (T-03-01)"
# ---------------------------------------------------------------------------
# D-03: zombie-link consecutive-failure break tests
# ---------------------------------------------------------------------------
def _make_zombie_client():
"""Build a mock BleakClient that connects successfully but has is_connected stuck True."""
mock_client = AsyncMock()
mock_client.connect = AsyncMock(return_value=None)
mock_client.is_connected = True # stale flag — never goes False
mock_client.disconnect = AsyncMock()
mock_client.start_notify = AsyncMock()
return mock_client
def test_zombie_link_break_after_limit_consecutive_failures(monkeypatch):
"""Loop breaks after exactly ZOMBIE_BREAK_LIMIT consecutive False writes (default 1)."""
import daemon.claude_usage_daemon_windows as mod
device = _make_device()
stop_event = asyncio.run(_make_event(False))
mock_client = _make_zombie_client()
write_call_count = [0]
async def fake_write_payload(payload):
write_call_count[0] += 1
return False # always fail — zombie link
fake_session = AsyncMock()
fake_session.write_payload = fake_write_payload
fake_session.refresh_requested = MagicMock()
fake_session.refresh_requested.is_set = MagicMock(return_value=False)
fake_session.refresh_requested.clear = MagicMock()
fake_session.refresh_requested.wait = AsyncMock()
# Force elapsed >= POLL_INTERVAL immediately
monkeypatch.setattr(mod, "POLL_INTERVAL", 0)
async def fast_wait_for(coro, timeout):
raise asyncio.TimeoutError()
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.Session", return_value=fake_session), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
patch("daemon.claude_usage_daemon_windows.poll_api",
new=AsyncMock(return_value={"ok": True})), \
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for",
side_effect=fast_wait_for):
result = _run(connect_and_run(device, stop_event))
# With ZOMBIE_BREAK_LIMIT=1, one False write should break the loop
assert write_call_count[0] == mod.ZOMBIE_BREAK_LIMIT
# Should return used_successfully=False (no successful write)
assert result is False
def test_zombie_counter_resets_on_success_with_raised_limit(monkeypatch):
"""A failed write followed by success resets counter (limit raised to 2 to exercise reset)."""
import daemon.claude_usage_daemon_windows as mod
device = _make_device()
stop_event = asyncio.run(_make_event(False))
mock_client = _make_zombie_client()
# Sequence: False (counter=1), True (counter reset to 0), False (counter=1 again), break
write_results = iter([False, True, False])
write_call_count = [0]
async def fake_write_payload(payload):
write_call_count[0] += 1
try:
return next(write_results)
except StopIteration:
return False
# After success, subsequent False write breaks at limit=2 (requires 2 consecutive)
# With limit=2: False (1), True (reset to 0), False (1), False (2 -> break)
# But we only have 3 items in write_results; after StopIteration returns False.
# Let's use a longer sequence to ensure reset-then-2-failures trip the break.
write_results2 = [False, True, False, False]
write_call_count2 = [0]
async def fake_write_payload2(payload):
write_call_count2[0] += 1
if write_call_count2[0] - 1 < len(write_results2):
return write_results2[write_call_count2[0] - 1]
return False
fake_session = AsyncMock()
fake_session.write_payload = fake_write_payload2
fake_session.refresh_requested = MagicMock()
fake_session.refresh_requested.is_set = MagicMock(return_value=False)
fake_session.refresh_requested.clear = MagicMock()
fake_session.refresh_requested.wait = AsyncMock()
monkeypatch.setattr(mod, "POLL_INTERVAL", 0)
monkeypatch.setattr(mod, "ZOMBIE_BREAK_LIMIT", 2) # raise limit to test reset logic
async def fast_wait_for(coro, timeout):
raise asyncio.TimeoutError()
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.Session", return_value=fake_session), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
patch("daemon.claude_usage_daemon_windows.poll_api",
new=AsyncMock(return_value={"ok": True})), \
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for",
side_effect=fast_wait_for):
result = _run(connect_and_run(device, stop_event))
# With limit=2 and sequence [False, True, False, False]:
# cycle 1: False -> consecutive_failures=1 (no break, limit=2)
# cycle 2: True -> consecutive_failures=0 (reset)
# cycle 3: False -> consecutive_failures=1 (no break)
# cycle 4: False -> consecutive_failures=2 -> break
assert write_call_count2[0] == 4, (
f"Expected 4 write calls (reset-on-success logic), got {write_call_count2[0]}"
)
# used_successfully=True because cycle 2 succeeded
assert result is True
def test_zombie_break_disconnect_called_in_finally(monkeypatch):
"""The finally block calls client.disconnect() exactly once on the zombie-break path."""
import daemon.claude_usage_daemon_windows as mod
device = _make_device()
stop_event = asyncio.run(_make_event(False))
mock_client = _make_zombie_client()
async def fake_write_payload(payload):
return False # always fail
fake_session = AsyncMock()
fake_session.write_payload = fake_write_payload
fake_session.refresh_requested = MagicMock()
fake_session.refresh_requested.is_set = MagicMock(return_value=False)
fake_session.refresh_requested.clear = MagicMock()
fake_session.refresh_requested.wait = AsyncMock()
monkeypatch.setattr(mod, "POLL_INTERVAL", 0)
async def fast_wait_for(coro, timeout):
raise asyncio.TimeoutError()
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.Session", return_value=fake_session), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
patch("daemon.claude_usage_daemon_windows.poll_api",
new=AsyncMock(return_value={"ok": True})), \
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for",
side_effect=fast_wait_for):
_run(connect_and_run(device, stop_event))
# The finally block calls disconnect() exactly once
assert mock_client.disconnect.call_count == 1
def test_zombie_break_returns_used_successfully_false(monkeypatch):
"""connect_and_run returns used_successfully=False after zombie break with no writes."""
import daemon.claude_usage_daemon_windows as mod
device = _make_device()
stop_event = asyncio.run(_make_event(False))
mock_client = _make_zombie_client()
async def fake_write_payload(payload):
return False
fake_session = AsyncMock()
fake_session.write_payload = fake_write_payload
fake_session.refresh_requested = MagicMock()
fake_session.refresh_requested.is_set = MagicMock(return_value=False)
fake_session.refresh_requested.clear = MagicMock()
fake_session.refresh_requested.wait = AsyncMock()
monkeypatch.setattr(mod, "POLL_INTERVAL", 0)
async def fast_wait_for(coro, timeout):
raise asyncio.TimeoutError()
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.Session", return_value=fake_session), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
patch("daemon.claude_usage_daemon_windows.poll_api",
new=AsyncMock(return_value={"ok": True})), \
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for",
side_effect=fast_wait_for):
result = _run(connect_and_run(device, stop_event))
# main() uses this return value to route into reconnect branch
assert result is False
# ---------------------------------------------------------------------------
# D-05: split fast-reconnect vs slow-search backoff in main()
# ---------------------------------------------------------------------------
def test_next_backoff_slow_search_doubles_to_60():
"""_next_backoff doubles correctly and never exceeds 60 (slow-search cap)."""
import daemon.claude_usage_daemon_windows as mod
values = []
b = 1
for _ in range(10):
b = mod._next_backoff(b, 60)
values.append(b)
assert values == [2, 4, 8, 16, 32, 60, 60, 60, 60, 60]
assert max(values) <= 60
def test_next_backoff_fast_reconnect_doubles_to_cap():
"""_next_backoff doubles correctly and never exceeds RECONNECT_BACKOFF_CAP (default 8)."""
import daemon.claude_usage_daemon_windows as mod
cap = mod.RECONNECT_BACKOFF_CAP
assert cap < 60, "Fast cap must be strictly lower than search cap"
values = []
b = 1
for _ in range(8):
b = mod._next_backoff(b, cap)
values.append(b)
# Should double until hitting the cap, then stay there
assert max(values) <= cap
# Should reach the cap (not just stay at 1)
assert values[-1] == cap
def test_next_backoff_one_to_two():
"""_next_backoff(1, 60) == 2 (basic sanity)."""
import daemon.claude_usage_daemon_windows as mod
assert mod._next_backoff(1, 60) == 2
def test_next_backoff_at_cap_stays():
"""_next_backoff(cap, cap) == cap (does not overflow)."""
import daemon.claude_usage_daemon_windows as mod
assert mod._next_backoff(mod.RECONNECT_BACKOFF_CAP, mod.RECONNECT_BACKOFF_CAP) == mod.RECONNECT_BACKOFF_CAP
def test_main_scan_miss_uses_search_backoff():
"""When scan_for_device returns None, asyncio.wait_for receives search_backoff timeout values."""
import daemon.claude_usage_daemon_windows as mod
# Capture main()'s internal stop_event by intercepting asyncio.Event()
internal_stop_event = [None]
real_Event = asyncio.Event
def capturing_Event():
ev = real_Event()
internal_stop_event[0] = ev
return ev
recorded_timeouts = []
call_count = [0]
MAX_CALLS = 3
async def fake_scan():
return None # always miss -> slow-search regime
async def fake_wait_for(coro, timeout):
recorded_timeouts.append(timeout)
call_count[0] += 1
if call_count[0] >= MAX_CALLS and internal_stop_event[0] is not None:
internal_stop_event[0].set() # terminate main()'s outer while loop
raise asyncio.TimeoutError()
with patch("daemon.claude_usage_daemon_windows.asyncio.Event", side_effect=capturing_Event), \
patch("daemon.claude_usage_daemon_windows.scan_for_device", side_effect=fake_scan), \
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for", side_effect=fake_wait_for):
_run(mod.main())
# Should have recorded timeouts from search_backoff sequence: 1, 2, 4 (then stop)
assert len(recorded_timeouts) >= 2
# Timeouts should be doubling (search_backoff sequence)
assert recorded_timeouts[0] == 1
assert recorded_timeouts[1] == 2
# None should exceed the search cap (60)
assert all(t <= 60 for t in recorded_timeouts)
def test_main_connect_fail_uses_reconnect_backoff():
"""When connect_and_run returns False, asyncio.wait_for receives reconnect_backoff timeouts (fast cap)."""
import daemon.claude_usage_daemon_windows as mod
# Capture main()'s internal stop_event
internal_stop_event = [None]
real_Event = asyncio.Event
def capturing_Event():
ev = real_Event()
internal_stop_event[0] = ev
return ev
fake_device = _make_device()
recorded_timeouts = []
call_count = [0]
MAX_CALLS = 3
async def fake_scan():
return fake_device # always finds device
async def fake_connect_and_run(device, event, tray_state=None):
return False # always fails -> fast-reconnect regime
async def fake_wait_for(coro, timeout):
recorded_timeouts.append(timeout)
call_count[0] += 1
if call_count[0] >= MAX_CALLS and internal_stop_event[0] is not None:
internal_stop_event[0].set()
raise asyncio.TimeoutError()
with patch("daemon.claude_usage_daemon_windows.asyncio.Event", side_effect=capturing_Event), \
patch("daemon.claude_usage_daemon_windows.scan_for_device", side_effect=fake_scan), \
patch("daemon.claude_usage_daemon_windows.connect_and_run", side_effect=fake_connect_and_run), \
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for", side_effect=fake_wait_for):
_run(mod.main())
# Should have recorded timeouts from reconnect_backoff sequence: 1, 2, 4 (then stop)
assert len(recorded_timeouts) >= 2
assert recorded_timeouts[0] == 1
assert recorded_timeouts[1] == 2
# All timeouts must be at or below RECONNECT_BACKOFF_CAP (fast cap, < 60)
assert all(t <= mod.RECONNECT_BACKOFF_CAP for t in recorded_timeouts)
def test_main_reconnect_backoff_reset_on_success():
"""A successful connect_and_run (returns True) resets reconnect_backoff to 1."""
import daemon.claude_usage_daemon_windows as mod
# Capture main()'s internal stop_event
internal_stop_event = [None]
real_Event = asyncio.Event
def capturing_Event():
ev = real_Event()
internal_stop_event[0] = ev
return ev
fake_device = _make_device()
recorded_timeouts = []
call_count = [0]
# Sequence: fail (reconnect_backoff=1), succeed (reset), fail (reconnect_backoff=1 again)
connect_results = [False, True, False]
connect_idx = [0]
async def fake_scan():
return fake_device
async def fake_connect_and_run(device, event, tray_state=None):
idx = connect_idx[0]
connect_idx[0] += 1
if idx < len(connect_results):
return connect_results[idx]
return False
async def fake_wait_for(coro, timeout):
recorded_timeouts.append(timeout)
call_count[0] += 1
if call_count[0] >= 2 and internal_stop_event[0] is not None:
internal_stop_event[0].set() # stop after 2 waits (first fail + post-success fail)
raise asyncio.TimeoutError()
with patch("daemon.claude_usage_daemon_windows.asyncio.Event", side_effect=capturing_Event), \
patch("daemon.claude_usage_daemon_windows.scan_for_device", side_effect=fake_scan), \
patch("daemon.claude_usage_daemon_windows.connect_and_run", side_effect=fake_connect_and_run), \
patch("daemon.claude_usage_daemon_windows.asyncio.wait_for", side_effect=fake_wait_for):
_run(mod.main())
# First wait: reconnect_backoff=1 (initial failure)
# Second wait: reconnect_backoff=1 again (reset by success, then another failure)
assert len(recorded_timeouts) >= 2
assert recorded_timeouts[0] == 1, f"Expected 1 on first fail, got {recorded_timeouts[0]}"
assert recorded_timeouts[1] == 1, f"Expected 1 after success reset, got {recorded_timeouts[1]}"
def test_main_no_saved_addr_file_or_skip_addr():
"""main() does not reference SAVED_ADDR_FILE or skip_addr (Windows is stateless - D-04)."""
import inspect
import daemon.claude_usage_daemon_windows as mod
source = inspect.getsource(mod.main)
assert "SAVED_ADDR_FILE" not in source, "main() must not reference SAVED_ADDR_FILE (D-04)"
assert "skip_addr" not in source, "main() must not reference skip_addr (macOS-only)"
assert "retrieve_connected" not in source.lower(), \
"main() must not reference retrieve_connected (macOS HID path)"
def test_requirements_windows_contains_required_deps():
"""requirements-windows.txt must contain the expected deps.
Phase 3 (reconnect) added no new deps; Phase 4 (tray) adds pystray + Pillow.
This test asserts the final expected state: bleak, httpx, pystray, Pillow
must be present; winreg must NOT be listed (it is stdlib — no install needed).
"""
req_path = Path(__file__).parent.parent / "requirements-windows.txt"
content = req_path.read_text()
lines = {line.strip().lower() for line in content.splitlines()
if line.strip() and not line.strip().startswith("#")}
assert "bleak" in lines, "bleak must be in requirements-windows.txt"
assert "httpx" in lines, "httpx must be in requirements-windows.txt"
assert "pystray" in lines, "pystray must be in requirements-windows.txt (Phase 4)"
assert "pillow" in lines, "Pillow must be in requirements-windows.txt (Phase 4)"
assert "winreg" not in lines, "winreg is stdlib — must NOT be in requirements-windows.txt"
# ---------------------------------------------------------------------------
# G-03-01: start_notify() OSError must not crash the daemon (SC#3 power-cycle)
# ---------------------------------------------------------------------------
def test_start_notify_oserror_does_not_crash_connect_and_run():
"""G-03-01 regression: on post-power-cycle reconnect, WinRT's start_notify()
CCCD write can raise a raw OSError/WinError when the just-rebooted peer GATT
server is not yet ready. The optional refresh subscription must degrade
gracefully — connect_and_run must NOT propagate the OSError and must proceed
into the poll loop (returning normally), so the daemon never restarts (SC#3/SC#4).
"""
device = _make_device()
# stop_event set so the poll loop exits immediately after subscription setup
stop_event = asyncio.run(_make_event(True))
mock_client = AsyncMock()
mock_client.connect = AsyncMock(return_value=None) # connect succeeds
mock_client.is_connected = True
mock_client.disconnect = AsyncMock()
# The exact failure observed on hardware (SC#3, 2026-06-02):
# OSError: [WinError -2147023673] The operation was canceled by the user.
mock_client.start_notify = AsyncMock(
side_effect=OSError(-2147023673, "The operation was canceled by the user.")
)
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=mock_client), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value="fake-token"), \
patch("daemon.claude_usage_daemon_windows.poll_api", new=AsyncMock(return_value={"ok": True})):
# Must NOT raise OSError — graceful degradation into the poll loop.
result = _run(connect_and_run(device, stop_event))
# start_notify was actually attempted (and raised), but was swallowed.
assert mock_client.start_notify.call_count == 1
# Function returned normally instead of propagating the OSError.
assert result is False
# The link was cleaned up via the finally block.
assert mock_client.disconnect.call_count >= 1
# ---------------------------------------------------------------------------
# SC#2 field report: write_payload() OSError must not crash the daemon thread
# ---------------------------------------------------------------------------
def test_write_payload_oserror_returns_false_not_raises():
"""SC#2 regression: write_gatt_char can raise a raw OSError/WinError (NOT a
BleakError) when the peer GATT server goes transiently unavailable mid-write.
write_payload must catch it and return False — tripping the zombie-link break
for a clean reconnect — instead of propagating an uncaught exception that
silently kills the daemon=True background thread and freezes the tray.
"""
mock_client = AsyncMock()
mock_client.write_gatt_char = AsyncMock(
side_effect=OSError(-2147023673, "The operation was canceled by the user.")
)
session = Session(mock_client)
result = _run(session.write_payload({"ok": True}))
assert result is False # caught and reported, not raised
assert mock_client.write_gatt_char.call_count == 1
def test_write_payload_bleak_error_still_returns_false():
"""The pre-existing BleakError path must keep returning False (no regression
from widening the except to also cover OSError)."""
mock_client = AsyncMock()
mock_client.write_gatt_char = AsyncMock(side_effect=BleakError("disconnected"))
session = Session(mock_client)
assert _run(session.write_payload({"ok": True})) is False
# ---------------------------------------------------------------------------
# SC#3 graceful Quit: _wait_first wakes immediately on stop (clean disconnect)
# ---------------------------------------------------------------------------
def test_wait_first_returns_immediately_when_an_event_is_set():
"""The poll loop's TICK wait must break the instant stop_event is set, so the
finally: client.disconnect() runs before the process exits (SC#3). The outer
wait_for(2s) fails fast if _wait_first wrongly blocks for the full 30s timeout."""
async def go():
refresh = asyncio.Event()
stop = asyncio.Event()
stop.set() # stop signalled
await asyncio.wait_for(_wait_first(refresh, stop, timeout=30.0), timeout=2.0)
assert not refresh.is_set() # loser waiter drained, refresh untouched
_run(go())
def test_wait_first_returns_after_timeout_when_no_event_set():
"""With neither event set, _wait_first returns after `timeout` (the normal
poll-tick path) rather than hanging."""
async def go():
await asyncio.wait_for(
_wait_first(asyncio.Event(), asyncio.Event(), timeout=0.05), timeout=2.0
)
_run(go())
# ---------------------------------------------------------------------------
# SC#5: transient poll failure must NOT toast "token expired"; only a real
# 401/403 (AuthError) should. A boot-time DNS blip returns None, not AuthError.
# ---------------------------------------------------------------------------
def _connected_mock_client():
client = AsyncMock()
client.connect = AsyncMock(return_value=None)
client.is_connected = True
client.disconnect = AsyncMock()
client.start_notify = AsyncMock()
return client
def test_transient_poll_failure_does_not_set_error():
"""poll_api returning None (network/DNS, timeout, 5xx, 429) is transient and
must leave the tray state untouched — not flip it to 'token expired' (SC#5
field report: `getaddrinfo failed` at boot wrongly fired the toast)."""
device = _make_device()
stop_event = asyncio.run(_make_event(False))
tray_state = MagicMock()
client = _connected_mock_client()
async def fake_poll(_token):
stop_event.set() # end the loop after this single transient failure
return None
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=client), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value="tok"), \
patch("daemon.claude_usage_daemon_windows.poll_api", new=fake_poll):
_run(connect_and_run(device, stop_event, tray_state))
tray_state.set_error.assert_not_called()
tray_state.set_connected.assert_not_called()
def test_auth_error_sets_token_expired():
"""A genuine 401/403 surfaces as AuthError and DOES flip the tray to the
actionable 'token expired — run claude login' error state."""
device = _make_device()
stop_event = asyncio.run(_make_event(False))
tray_state = MagicMock()
client = _connected_mock_client()
async def fake_poll(_token):
stop_event.set()
raise AuthError(401)
with patch("daemon.claude_usage_daemon_windows.BleakClient", return_value=client), \
patch("daemon.claude_usage_daemon_windows.read_token", return_value="tok"), \
patch("daemon.claude_usage_daemon_windows.poll_api", new=fake_poll):
_run(connect_and_run(device, stop_event, tray_state))
tray_state.set_error.assert_called_once_with("token expired — run claude login")
+198
View File
@@ -0,0 +1,198 @@
#!/usr/bin/env python3
"""Unit tests for daemon/claude_usage_daemon_windows.py — TOKEN-01.
Run: python -m pytest daemon/tests/test_windows_token.py -x -q
"""
import json
import subprocess
import sys
from pathlib import Path
import pytest
from daemon.claude_usage_daemon_windows import _extract_access_token, read_token, _windows_credential_candidates, _read_expiry
FIXTURES = Path(__file__).parent / "fixtures"
def test_extract_nested_shape():
"""_extract_access_token handles the real Windows claudeAiOauth nested shape."""
blob = (FIXTURES / "credentials_nested.json").read_text()
assert _extract_access_token(blob) == "sk-ant-test-1234"
def test_extract_direct_shape():
"""_extract_access_token handles the legacy direct accessToken shape."""
blob = (FIXTURES / "credentials_direct.json").read_text()
assert _extract_access_token(blob) == "sk-ant-test-5678"
def test_read_token_env_override(tmp_path, monkeypatch):
"""read_token() honours CLAUDE_CREDENTIALS_PATH env override (D-03)."""
creds = tmp_path / ".credentials.json"
creds.write_text(json.dumps({"accessToken": "sk-ant-test-ENV"}))
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(creds))
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
assert read_token() == "sk-ant-test-ENV"
def test_read_token_primary_path(tmp_path, monkeypatch):
"""read_token() reads from the primary candidate path (first hit wins)."""
creds = tmp_path / ".claude" / ".credentials.json"
creds.parent.mkdir(parents=True)
creds.write_text(json.dumps({"claudeAiOauth": {"accessToken": "sk-ant-test-PRIMARY"}}))
monkeypatch.delenv("CLAUDE_CREDENTIALS_PATH", raising=False)
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
# Monkeypatch _windows_credential_candidates to return only our tmp path
import daemon.claude_usage_daemon_windows as mod
monkeypatch.setattr(mod, "_windows_credential_candidates", lambda: [creds])
assert read_token() == "sk-ant-test-PRIMARY"
def test_read_token_localappdata_fallback(tmp_path, monkeypatch):
"""read_token() falls back to %LOCALAPPDATA%/Claude/.credentials.json when primary is absent."""
missing_primary = tmp_path / "nonexistent_primary" / ".credentials.json"
present_localappdata = tmp_path / "localappdata" / ".credentials.json"
missing_appdata = tmp_path / "nonexistent_appdata" / ".credentials.json"
present_localappdata.parent.mkdir(parents=True)
present_localappdata.write_text(json.dumps({"accessToken": "sk-ant-test-LA"}))
import daemon.claude_usage_daemon_windows as mod
monkeypatch.setattr(
mod,
"_windows_credential_candidates",
lambda: [missing_primary, present_localappdata, missing_appdata],
)
assert read_token() == "sk-ant-test-LA"
def test_read_token_appdata_fallback(tmp_path, monkeypatch):
"""read_token() falls back to %APPDATA%/Claude/.credentials.json when primary and LOCALAPPDATA are absent."""
missing_primary = tmp_path / "nonexistent_primary" / ".credentials.json"
missing_localappdata = tmp_path / "nonexistent_localappdata" / ".credentials.json"
present_appdata = tmp_path / "appdata" / ".credentials.json"
present_appdata.parent.mkdir(parents=True)
present_appdata.write_text(json.dumps({"accessToken": "sk-ant-test-APP"}))
import daemon.claude_usage_daemon_windows as mod
monkeypatch.setattr(
mod,
"_windows_credential_candidates",
lambda: [missing_primary, missing_localappdata, present_appdata],
)
assert read_token() == "sk-ant-test-APP"
def test_read_token_no_file(tmp_path, monkeypatch):
"""read_token() returns None when no credential file can be found."""
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(tmp_path / "nonexistent.json"))
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
assert read_token() is None
def test_read_token_config_dir_override(tmp_path, monkeypatch):
"""read_token() honours the official CLAUDE_CONFIG_DIR env override."""
creds = tmp_path / ".credentials.json"
creds.write_text(json.dumps({"accessToken": "sk-ant-test-CFGDIR"}))
monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path))
monkeypatch.delenv("CLAUDE_CREDENTIALS_PATH", raising=False)
assert read_token() == "sk-ant-test-CFGDIR"
def test_read_expiry_decodes_milliseconds(monkeypatch):
"""_read_expiry() divides expiresAt by 1000 (ms -> s); fixture 9999999999000 -> year 2286."""
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(FIXTURES / "credentials_nested.json"))
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
result = _read_expiry()
assert result.startswith("2286-"), f"Expected year 2286, got: {result}"
# --- WR-03: regression guard for CR-01 (empty/blank token must not be accepted) ---
def test_extract_empty_token_is_none():
"""_extract_access_token returns None for empty accessToken (CR-01 regression guard)."""
assert _extract_access_token('{"accessToken": ""}') is None
assert _extract_access_token('{}') is None
def test_read_token_empty_credential_file_returns_none(tmp_path, monkeypatch):
"""read_token() returns None (not empty string) when credential file has empty accessToken."""
creds = tmp_path / ".credentials.json"
creds.write_text(json.dumps({"accessToken": ""}))
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(creds))
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
assert read_token() is None
# --- WR-01: regression guard for _read_expiry with non-dict top-level JSON ---
def test_read_expiry_non_dict_json_returns_unknown(tmp_path, monkeypatch):
"""_read_expiry() returns 'expiry unknown' (not crash) for non-dict top-level JSON (WR-01)."""
creds = tmp_path / ".credentials.json"
creds.write_text("[1, 2, 3]")
monkeypatch.setenv("CLAUDE_CREDENTIALS_PATH", str(creds))
monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False)
assert _read_expiry() == "expiry unknown"
# --- WR-02: D-06 redaction requirement must be tested ---
def test_main_emits_linux_warning(monkeypatch):
"""__main__ prints a non-fatal stderr warning on non-Windows platforms (new async runner).
Phase 2 replaced the Phase 1 token-printing __main__ with asyncio.run(main()). The
new contract:
- On non-Windows: emits "WinRT BLE will not be available" to stderr before the loop.
- Enters the async scan/connect/poll loop (no longer prints token/expiry).
This test interrupts the process after 3s to capture the warning without hanging.
"""
env = {**__import__("os").environ, "CLAUDE_CREDENTIALS_PATH": str(FIXTURES / "credentials_nested.json")}
env.pop("CLAUDE_CONFIG_DIR", None)
module = str(Path(__file__).parent.parent / "claude_usage_daemon_windows.py")
try:
result = subprocess.run(
[sys.executable, module],
capture_output=True,
text=True,
env=env,
timeout=3,
)
# If it exits cleanly, verify warning was emitted
assert "WinRT BLE will not be available" in result.stderr
except subprocess.TimeoutExpired as exc:
# Process is hanging in the scan loop — expected behavior on Linux.
# The warning should appear in the partial stderr captured so far.
partial_stderr = (exc.stderr or b"").decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
assert "WinRT BLE will not be available" in partial_stderr, (
f"Expected Linux/WSL warning in stderr before scan loop, got: {partial_stderr!r}"
)
def test_main_emits_linux_warning_before_loop(monkeypatch):
"""__main__ stderr warning appears before the async scan loop starts on Linux/WSL."""
import signal as _signal
env = {**__import__("os").environ}
env.pop("CLAUDE_CONFIG_DIR", None)
env.pop("CLAUDE_CREDENTIALS_PATH", None)
module = str(Path(__file__).parent.parent / "claude_usage_daemon_windows.py")
try:
result = subprocess.run(
[sys.executable, module],
capture_output=True,
text=True,
env=env,
timeout=3,
)
# If it exits cleanly (KeyboardInterrupt path), check warning
assert "WinRT BLE will not be available" in result.stderr
except subprocess.TimeoutExpired as exc:
# Process is hanging in the scan loop — expected behavior on Linux.
# The warning should appear in the partial stderr captured so far.
partial_stderr = (exc.stderr or b"").decode("utf-8", errors="replace") if isinstance(exc.stderr, bytes) else (exc.stderr or "")
assert "WinRT BLE will not be available" in partial_stderr, (
f"Expected Linux/WSL warning in stderr before scan loop, got: {partial_stderr!r}"
)
+328
View File
@@ -0,0 +1,328 @@
#!/usr/bin/env python3
"""Unit tests for daemon/tray_windows.py — APP-01.
Covers:
TrayState scalar setters and initial state
header_text() for all three states including last_sync=None
daemon main() accepts tray_state and populates ts.loop / ts.stop_event
Quit routes through loop.call_soon_threadsafe (not stop_event.set directly)
Error toast fires only on transition INTO error state (D-04)
All pystray usage is inside tray_windows.main() (deferred import), so these
tests can import the pure helpers (TrayState, header_text) and test Quit/toast
handlers with mocked icons without importing the GTK-less top-level pystray.
Run: python -m pytest daemon/tests/test_windows_tray.py -x -q
"""
import asyncio
import time
from unittest.mock import AsyncMock, MagicMock, patch, call
import pytest
from daemon.tray_windows import TrayState, header_text, _acquire_single_instance, _ERROR_ALREADY_EXISTS
# ---------------------------------------------------------------------------
# TrayState — initial state and setters
# ---------------------------------------------------------------------------
def test_tray_state_initial():
"""TrayState initialises to scanning state with no last_sync."""
ts = TrayState()
assert ts.state == "scanning"
assert ts.reason == ""
assert ts.last_sync is None
assert ts.loop is None
assert ts.stop_event is None
def test_set_connected():
"""set_connected(ts_float) sets state='connected', clears reason, records last_sync."""
ts = TrayState()
now = time.time()
ts.set_connected(now)
assert ts.state == "connected"
assert ts.reason == ""
assert ts.last_sync == now
def test_set_scanning():
"""set_scanning() sets state='scanning', clears reason."""
ts = TrayState()
ts.set_error("something bad") # put it in error first
ts.set_scanning()
assert ts.state == "scanning"
assert ts.reason == ""
def test_set_error():
"""set_error(why) sets state='error' and stores the reason string."""
ts = TrayState()
ts.set_error("token expired — run claude login")
assert ts.state == "error"
assert ts.reason == "token expired — run claude login"
# ---------------------------------------------------------------------------
# header_text — D-05 string shapes
# ---------------------------------------------------------------------------
def test_header_text_scanning():
"""header_text returns 'Scanning…' in scanning state."""
ts = TrayState()
ts.set_scanning()
assert header_text(ts) == "Scanning…"
def test_header_text_error():
"""header_text returns 'Error: {reason}' in error state."""
ts = TrayState()
ts.set_error("token expired — run claude login")
result = header_text(ts)
assert result == "Error: token expired — run claude login"
def test_header_text_connected_with_last_sync():
"""header_text returns 'Connected · last update HH:MM' when last_sync is set."""
ts = TrayState()
# Use a known timestamp so we can predict the HH:MM string.
known_ts = time.mktime(time.strptime("2026-06-01 14:32:00", "%Y-%m-%d %H:%M:%S"))
ts.set_connected(known_ts)
result = header_text(ts)
# Extract the HH:MM portion from the actual local time expansion.
expected_when = time.strftime("%H:%M", time.localtime(known_ts))
assert result == f"Connected · last update {expected_when}"
def test_header_text_connected_never_when_last_sync_none():
"""header_text returns 'Connected · last update never' when last_sync is None."""
ts = TrayState()
# Manually set state without using set_connected so last_sync stays None.
ts.state = "connected"
ts.last_sync = None
result = header_text(ts)
assert result == "Connected · last update never"
# ---------------------------------------------------------------------------
# daemon main() populates ts.loop and ts.stop_event
# ---------------------------------------------------------------------------
def test_main_populates_tray_state_loop_and_stop_event():
"""daemon main(tray_state=ts) sets ts.loop and ts.stop_event before the loop body."""
import daemon.claude_usage_daemon_windows as mod
ts = TrayState()
populated = {}
async def _fake_scan():
# Record the state of ts at first scan entry (after main() startup lines).
populated["loop"] = ts.loop
populated["stop_event"] = ts.stop_event
# Signal stop so the loop exits cleanly.
ts.stop_event.set()
return None # no device found
with patch.object(mod, "scan_for_device", side_effect=_fake_scan):
asyncio.run(mod.main(tray_state=ts))
assert populated.get("loop") is not None, "ts.loop must be set by daemon main()"
assert populated.get("stop_event") is not None, "ts.stop_event must be set by daemon main()"
# ---------------------------------------------------------------------------
# Quit handler routes through call_soon_threadsafe (not stop_event.set directly)
# ---------------------------------------------------------------------------
def test_quit_uses_call_soon_threadsafe():
"""The Quit menu handler calls loop.call_soon_threadsafe(stop_event.set) and icon.stop().
It must NOT call stop_event.set() directly from the tray thread
(RESEARCH Pitfall 2 / T-04-06 mitigation).
"""
# Build a TrayState with a mocked loop and stop_event.
ts = TrayState()
mock_loop = MagicMock()
mock_stop_event = MagicMock()
ts.loop = mock_loop
ts.stop_event = mock_stop_event
# Build the Quit handler the same way tray_windows.main() does, without
# importing pystray at the module level. We construct a local closure
# that mirrors the on_quit body.
mock_icon = MagicMock()
def _on_quit(icon_ref, _item):
# This is the exact body from tray_windows.main() — keep in sync.
ts.loop.call_soon_threadsafe(ts.stop_event.set)
icon_ref.stop()
_on_quit(mock_icon, None)
# call_soon_threadsafe must have been called with stop_event.set as the arg.
mock_loop.call_soon_threadsafe.assert_called_once_with(mock_stop_event.set)
# icon.stop() must have been called.
mock_icon.stop.assert_called_once()
# stop_event.set() must NOT have been called directly.
mock_stop_event.set.assert_not_called()
# ---------------------------------------------------------------------------
# Error toast fires only on transition INTO error (D-04)
# ---------------------------------------------------------------------------
def test_error_toast_on_entry_only():
"""The tray refresh loop fires icon.notify() only on transition INTO error.
Sequence: scanning -> error -> error
Expected: notify called exactly once (on the scanning->error transition).
"""
ts = TrayState()
ts.set_scanning()
mock_icon = MagicMock()
mock_icon._running = True
# Simulate the _refresh loop's state-change detection logic from tray_windows.main().
# We run two transitions manually:
# 1. scanning -> error (should call notify once)
# 2. error -> error (no change — notify must NOT fire again)
prev_state: dict = {"state": None}
def _process_state_change(new_state: str, reason: str = "") -> None:
"""Mirror the relevant part of the _refresh loop body."""
ts.state = new_state
ts.reason = reason
current = ts.state
if current != prev_state["state"]:
if current == "error" and prev_state["state"] != "error":
mock_icon.notify(ts.reason or "Clawdmeter error", "Clawdmeter")
prev_state["state"] = current
# Transition 1: scanning -> error (notify should fire)
_process_state_change("scanning")
_process_state_change("error", "token expired — run claude login")
# Transition 2: error -> error (same state — no call)
_process_state_change("error", "token expired — run claude login")
mock_icon.notify.assert_called_once_with(
"token expired — run claude login", "Clawdmeter"
)
# ---------------------------------------------------------------------------
# Single-instance guard (named mutex) — duplicate-launch / ARSO collision
# ---------------------------------------------------------------------------
# Field bug: Windows "restart apps after sign-in" (ARSO) restored a console
# `python.exe tray_windows.py` instance while the headless `pythonw` autostart
# also fired — two trays fighting over the one BLE link. The guard makes a
# second instance exit before it touches BLE.
def test_single_instance_noop_off_windows():
"""Off-Windows the guard is a no-op that returns a truthy sentinel (never None)."""
with patch("daemon.tray_windows.sys") as mock_sys:
mock_sys.platform = "linux"
assert _acquire_single_instance() is not None
def _fake_kernel32(last_error: int, handle: int):
"""Build a fake ctypes module tree whose CreateMutexW returns `handle` and
whose get_last_error() returns `last_error`."""
fake_kernel32 = MagicMock()
fake_kernel32.CreateMutexW.return_value = handle
fake_ctypes = MagicMock()
fake_ctypes.WinDLL.return_value = fake_kernel32
fake_ctypes.get_last_error.return_value = last_error
return fake_ctypes
def test_single_instance_first_instance_gets_handle():
"""First instance: CreateMutexW succeeds, no prior owner → returns the handle."""
fake_ctypes = _fake_kernel32(last_error=0, handle=0xABCD)
with patch("daemon.tray_windows.sys") as mock_sys, \
patch.dict("sys.modules", {"ctypes": fake_ctypes, "ctypes.wintypes": MagicMock()}):
mock_sys.platform = "win32"
assert _acquire_single_instance() == 0xABCD
def test_single_instance_second_instance_gets_none():
"""Second instance: mutex already exists → returns None so caller exits."""
fake_ctypes = _fake_kernel32(last_error=_ERROR_ALREADY_EXISTS, handle=0xABCD)
with patch("daemon.tray_windows.sys") as mock_sys, \
patch.dict("sys.modules", {"ctypes": fake_ctypes, "ctypes.wintypes": MagicMock()}):
mock_sys.platform = "win32"
assert _acquire_single_instance() is None
def test_single_instance_fails_open_on_null_handle():
"""If CreateMutexW returns NULL, fail OPEN (truthy) — never block tray startup."""
fake_ctypes = _fake_kernel32(last_error=_ERROR_ALREADY_EXISTS, handle=0)
with patch("daemon.tray_windows.sys") as mock_sys, \
patch.dict("sys.modules", {"ctypes": fake_ctypes, "ctypes.wintypes": MagicMock()}):
mock_sys.platform = "win32"
result = _acquire_single_instance()
assert result is not None
# ---------------------------------------------------------------------------
# Regression: cwd-independent package + asset resolution (SC#1 logon autostart)
# ---------------------------------------------------------------------------
# Field bug: launching `pythonw.exe daemon\tray_windows.py` at logon starts with
# cwd = System32, so `import daemon.*` raised ModuleNotFoundError and the relative
# logo path failed — the tray crashed silently with no icon. tray_windows must
# self-locate the repo root from __file__ so it works from any working directory.
def test_repo_root_is_parent_of_daemon_package():
"""_REPO_ROOT points at the dir that CONTAINS the daemon package."""
import os
import daemon.tray_windows as tw
assert os.path.isdir(os.path.join(tw._REPO_ROOT, "daemon"))
assert os.path.isfile(
os.path.join(tw._REPO_ROOT, "firmware", "src", "logo.h")
), "brand logo must resolve from _REPO_ROOT, not the current working directory"
def test_repo_root_on_sys_path_after_import():
"""Importing tray_windows puts the repo root on sys.path so `daemon.*` resolves regardless of cwd."""
import sys
import daemon.tray_windows as tw
assert tw._REPO_ROOT in sys.path
# ---------------------------------------------------------------------------
# Regression: daemon main() must run in a BACKGROUND thread (SC#1 tray launch)
# ---------------------------------------------------------------------------
# Field bug: under the tray the loop runs in threading.Thread (pystray owns the
# main thread). OS signal-handler installation (loop.add_signal_handler /
# signal.signal) only works on the main thread, so main() raised
# "signal only works in main thread" and the daemon thread died on startup.
# main() must guard signal setup to the main thread; the tray owns shutdown.
def test_main_runs_in_background_thread_without_signal_error():
"""main(tray_state=ts) started from a non-main thread must not raise on signal setup."""
import threading as _threading
import daemon.claude_usage_daemon_windows as mod
ts = TrayState()
errors: list = []
async def _fake_scan():
ts.stop_event.set() # exit the loop immediately
return None
def _run() -> None:
try:
with patch.object(mod, "scan_for_device", side_effect=_fake_scan):
asyncio.run(mod.main(tray_state=ts))
except Exception as exc: # noqa: BLE001 — capture for the assertion
errors.append(exc)
t = _threading.Thread(target=_run)
t.start()
t.join(timeout=10)
assert not t.is_alive(), "daemon main() hung in background thread"
assert not errors, f"main() raised in a background thread: {errors!r}"
+276
View File
@@ -0,0 +1,276 @@
#!/usr/bin/env python3
"""Windows system-tray entry and state bridge for Clawdmeter — APP-01.
Provides:
TrayState — thread-safe scalar bridge (daemon loop writes, tray reads)
header_text — pure helper producing the D-05 status-header string
main() — tray entry: builds per-state icons, runs the daemon loop in a
bg thread, and runs pystray.Icon on the main thread
The daemon loop (claude_usage_daemon_windows.main) is UNCHANGED in logic;
this module injects only additive state-setter calls at existing branch points.
Usage::
python tray_windows.py
Run: python -m pytest daemon/tests/test_windows_tray.py -x -q
"""
import os
import sys
import threading
import time
# Repo root = the directory that CONTAINS the `daemon` package (this file is
# <repo>/daemon/tray_windows.py). Resolve it from __file__ so the package
# imports below and the brand-logo asset load work no matter what the current
# working directory is — critical for logon autostart, where the HKCU\Run entry
# starts with cwd = System32, not the repo (APP-01 / SC#1).
_REPO_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
if _REPO_ROOT not in sys.path:
sys.path.insert(0, _REPO_ROOT)
# Autostart launches us with the BASE interpreter's pythonw.exe, not the venv's
# (see autostart_windows._command — the venv pythonw redirector pops a console
# window). The base interpreter does NOT see the venv's site-packages, so add
# them here to resolve pystray/bleak/PIL. os.path.isdir guards the no-venv and
# already-inside-venv cases; site.addsitedir is a no-op on a missing dir anyway.
_VENV_SITE = os.path.join(_REPO_ROOT, ".venv", "Lib", "site-packages")
if os.path.isdir(_VENV_SITE):
import site
site.addsitedir(_VENV_SITE)
# ---------------------------------------------------------------------------
# TrayState — thread-safe scalar bridge (loop -> tray)
# ---------------------------------------------------------------------------
class TrayState:
"""Shared state object bridging the daemon asyncio loop to the tray.
The daemon loop writes state via the set_* methods; the tray reads the
scalar attributes. No lock is needed — writes are atomic attribute
assignments of simple Python scalars, and the tray only ever reads them.
The loop populates `loop` and `stop_event` at startup (inside
daemon_main()) so the tray's Quit handler can route through
loop.call_soon_threadsafe (RESEARCH Pitfall 2 / Anti-Pattern).
"""
def __init__(self) -> None:
self.state: str = "scanning" # "connected" | "scanning" | "error"
self.reason: str = "" # error reason string (D-04)
self.last_sync: float | None = None # time.time() of last successful write
# Populated by daemon main() at startup:
self.loop = None # asyncio running loop (for call_soon_threadsafe)
self.stop_event = None # asyncio.Event (the existing clean-shutdown hook)
def set_connected(self, ts: float) -> None:
"""Called after write_payload returns True. ts = time.time()."""
self.state = "connected"
self.reason = ""
self.last_sync = ts
def set_scanning(self) -> None:
"""Called in scan/reconnect branches. BLE churn stays Scanning (D-01)."""
self.state = "scanning"
self.reason = ""
def set_error(self, why: str) -> None:
"""Called on token-expired / API auth failure (D-01 Error = actionable only)."""
self.state = "error"
self.reason = why
# ---------------------------------------------------------------------------
# header_text — pure D-05 status header string
# ---------------------------------------------------------------------------
def header_text(ts: TrayState) -> str:
"""Return the D-05 menu status-header string for the current TrayState.
Shapes:
"Connected · last update HH:MM" (ts.last_sync is a float)
"Connected · last update never" (ts.last_sync is None)
"Scanning…"
"Error: {reason}"
"""
if ts.state == "connected":
if ts.last_sync is not None:
when = time.strftime("%H:%M", time.localtime(ts.last_sync))
else:
when = "never"
return f"Connected · last update {when}"
if ts.state == "scanning":
return "Scanning…" # "Scanning…"
return f"Error: {ts.reason}"
# ---------------------------------------------------------------------------
# single-instance guard (named kernel mutex — no stale-lock problem)
# ---------------------------------------------------------------------------
# Per-session mutex name. "Local\\" scopes it to the interactive logon, which is
# exactly the granularity we want: one tray per signed-in user. Both the headless
# autostart (HKCU\Run pythonw) and an ARSO-restored console instance live in the
# same session, so this name catches the duplicate-launch collision that produced
# the "mystery console window fighting the headless tray over BLE" field bug.
_SINGLETON_MUTEX_NAME = "Local\\Clawdmeter-tray-singleton"
_ERROR_ALREADY_EXISTS = 183
def _acquire_single_instance():
"""Acquire the process-wide single-instance lock.
Returns a truthy handle to keep alive for the process lifetime if this is
the first/only tray, or None if another Clawdmeter tray already owns the
lock (the caller must then exit immediately, before touching BLE).
Uses a named kernel mutex: Windows releases it automatically when the owning
process dies, so there is no stale-lock cleanup (unlike a pidfile). We never
CloseHandle it — the handle lives until process exit, which is precisely the
lock lifetime we want.
Off-Windows (Linux dev box / unit tests) this is a no-op that always
succeeds — the tray only ever runs on Windows, and the dev box must stay
importable for the pure-helper tests.
"""
if sys.platform != "win32":
return object() # no-op sentinel; never blocks off-Windows
import ctypes
from ctypes import wintypes
kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
kernel32.CreateMutexW.restype = wintypes.HANDLE
kernel32.CreateMutexW.argtypes = [wintypes.LPVOID, wintypes.BOOL, wintypes.LPCWSTR]
handle = kernel32.CreateMutexW(None, True, _SINGLETON_MUTEX_NAME)
if not handle:
# Couldn't create the mutex at all — fail OPEN so a kernel quirk never
# stops the tray from starting; single-instance is best-effort hardening.
return object()
if ctypes.get_last_error() == _ERROR_ALREADY_EXISTS:
return None # another instance already holds it
return handle
# ---------------------------------------------------------------------------
# main() — tray entry (pystray on main thread, daemon loop in bg thread)
# ---------------------------------------------------------------------------
def main() -> None:
"""Tray entry point: build icons, start daemon bg thread, run pystray.
`import pystray` is intentionally INSIDE this function (not at module top)
so the module can be imported on a GTK-less Linux dev box for unit tests
of the pure helpers (TrayState, header_text) without pystray failing.
"""
# Single-instance guard FIRST — before icons, the daemon thread, or any BLE
# work. If another tray already owns the session mutex (e.g. ARSO restored a
# console instance and the headless autostart also fired), exit silently.
# Under pythonw there is no console to print to, so this is a quiet return.
_instance_lock = _acquire_single_instance()
if _instance_lock is None:
return
import asyncio as _asyncio
import pystray
from pystray import Menu, MenuItem
import daemon.autostart_windows as autostart
from daemon.claude_usage_daemon_windows import main as daemon_main, log as daemon_log
from daemon.icon_assets import load_logo_rgba, build_state_icons
# Build per-state icons once at startup; swap icon.icon per tick (never recomposite).
base = load_logo_rgba(os.path.join(_REPO_ROOT, "firmware", "src", "logo.h"))
images = build_state_icons(base)
ts = TrayState()
icon = pystray.Icon("Clawdmeter", images["scanning"], "Clawdmeter")
# --- background thread: asyncio loop ---
def _run_daemon() -> None:
# daemon=True thread: an unhandled exception here would vanish silently
# and freeze the tray on its last state forever (the field "frozen tray"
# failure mode). Surface it instead — log the traceback to the rotating
# file and flip the tray to an actionable error state.
try:
_asyncio.run(daemon_main(tray_state=ts))
except Exception as e: # last-resort thread guard
import traceback
daemon_log(f"Daemon thread crashed: {e!r}")
daemon_log(traceback.format_exc())
ts.set_error(f"daemon crashed: {type(e).__name__}")
daemon_thread = threading.Thread(target=_run_daemon, daemon=True)
daemon_thread.start()
# --- menu ---
def _on_quit(icon_ref, _item) -> None:
# NEVER call ts.stop_event.set() directly from the tray thread;
# asyncio.Event is NOT thread-safe (RESEARCH Pitfall 2).
#
# After signalling, WAIT for the daemon thread to finish its graceful
# shutdown (the loop's finally: client.disconnect()) BEFORE we stop the
# icon and let the process exit. Without this join the daemon=True thread
# is killed mid-flight, the peer never gets a clean GATT disconnect, and
# the device sits frozen on stale data instead of returning to its waiting
# screen (SC#3 field report). The timeout caps the block so Quit can never
# hang if a WinRT disconnect wedges (rare) — we exit anyway as a fallback.
if ts.loop is not None and ts.stop_event is not None:
ts.loop.call_soon_threadsafe(ts.stop_event.set)
daemon_thread.join(timeout=6.0)
icon_ref.stop()
def _on_toggle(_icon_ref, _item) -> None:
if autostart.is_enabled():
autostart.disable()
else:
# Pass THIS file explicitly — without it enable() defaults the Run
# value to autostart_windows.py (which has no entry point and starts
# nothing), silently breaking menu-enabled autostart.
autostart.enable(tray_script=os.path.abspath(__file__))
icon.update_menu()
icon.menu = Menu(
# Non-clickable status header; text updates via update_menu() on state change.
MenuItem(lambda _item: header_text(ts), None, enabled=False),
# Start-at-login toggle: checked= is a CALLABLE for live query (Pitfall 6).
MenuItem("Start at login", _on_toggle, checked=lambda _item: autostart.is_enabled()),
MenuItem("Quit", _on_quit),
)
# --- setup callback (runs in pystray's setup thread, 1s poll) ---
prev_state: dict = {"state": None, "last_sync": None}
def _refresh(_icon: pystray.Icon) -> None:
_icon.visible = True
while _icon._running: # type: ignore[attr-defined]
current = ts.state
last_sync = ts.last_sync
state_changed = current != prev_state["state"]
# Refresh the tooltip/menu when last_sync advances too — not only on
# state change. A healthy "connected" daemon polling a flat usage
# value never changes state, so a transition-only refresh froze the
# "last update HH:MM" tooltip and read as a dead daemon (SC#2 field
# report: device + tooltip both looked stuck while polling was fine).
if state_changed or last_sync != prev_state["last_sync"]:
if state_changed:
_icon.icon = images[current] # icon image depends on state only
_icon.title = header_text(ts)
# D-04: toast ONLY on transition INTO error, not on every error tick.
if current == "error" and prev_state["state"] != "error":
_icon.notify(ts.reason or "Clawdmeter error", "Clawdmeter")
prev_state["state"] = current
prev_state["last_sync"] = last_sync
_icon.update_menu()
time.sleep(1.0)
# Blocks the main thread until icon.stop() is called from _on_quit.
icon.run(setup=_refresh)
if __name__ == "__main__":
main()
+125
View File
@@ -0,0 +1,125 @@
# install-windows.ps1 - Clawdmeter Windows turnkey bootstrap (D-09)
#
# Creates a Python virtual environment, installs dependencies from
# daemon\requirements-windows.txt, registers the tray app to launch at login
# (HKCU\...\Run, no admin required), and starts the tray app immediately.
#
# Usage:
# powershell -ExecutionPolicy Bypass -File install-windows.ps1
#
# Or, if you have already set a permissive execution policy:
# .\install-windows.ps1
#
# To disable autostart later: right-click the tray icon -> uncheck "Start at login"
# Or remove manually: reg delete "HKCU\Software\Microsoft\Windows\CurrentVersion\Run" /v Clawdmeter /f
#
# Security: this script downloads nothing from the internet. It installs only
# the packages listed in the in-repo daemon\requirements-windows.txt.
Set-StrictMode -Version Latest
$ErrorActionPreference = "Stop"
function Log {
param([string]$Msg)
$ts = Get-Date -Format "HH:mm:ss"
Write-Host "[$ts] $Msg"
}
$RepoRoot = $PSScriptRoot
if (-not $RepoRoot) {
$RepoRoot = (Get-Location).Path
}
Log "=== Clawdmeter Windows Install ==="
Log "Repository root: $RepoRoot"
# ------------------------------------------------------------------
# Guard: refuse to install from a WSL path (APP-02 / SC#4 / SC#5)
# ------------------------------------------------------------------
# If $RepoRoot lives on the WSL share (\\wsl$\... or \\wsl.localhost\...),
# the venv and the HKCU\Run autostart entry would both point at a path that
# disappears when WSL is shut down -- exactly the WSL-dependence this project
# exists to eliminate. Copy the repo to a native Windows path first.
if ($RepoRoot -match '\\\\wsl(\$|\.localhost)\\') {
throw @"
Refusing to install from a WSL path:
$RepoRoot
The Clawdmeter daemon must be WSL-independent. Installing from the WSL share
would make the virtual environment and login-autostart entry point at a path
that is unreachable once WSL shuts down.
Fix: copy this repository to a native Windows location and run the installer
there, e.g.
Copy-Item -Recurse '$RepoRoot' "$env:USERPROFILE\Clawdmeter"
cd "$env:USERPROFILE\Clawdmeter"
powershell -ExecutionPolicy Bypass -File install-windows.ps1
"@
}
# ------------------------------------------------------------------
# Step 1: Create virtual environment
# ------------------------------------------------------------------
$VenvDir = Join-Path $RepoRoot ".venv"
if (Test-Path $VenvDir) {
Log "Virtual environment already exists at .venv - skipping creation"
} else {
Log "Creating virtual environment at .venv ..."
& python -m venv $VenvDir
if ($LASTEXITCODE -ne 0) { throw "Failed to create virtual environment (exit $LASTEXITCODE)" }
Log "Virtual environment created"
}
# ------------------------------------------------------------------
# Step 2: Install dependencies
# ------------------------------------------------------------------
$PythonExe = Join-Path $VenvDir "Scripts\python.exe"
$PythonwExe = Join-Path $VenvDir "Scripts\pythonw.exe"
$RequirementsFile = Join-Path $RepoRoot "daemon\requirements-windows.txt"
Log "Installing dependencies from daemon\requirements-windows.txt ..."
& $PythonExe -m pip install --quiet -r $RequirementsFile
if ($LASTEXITCODE -ne 0) { throw "pip install failed (exit $LASTEXITCODE)" }
Log "Dependencies installed"
# ------------------------------------------------------------------
# Step 3: Register autostart (HKCU\Run, per-user, no admin needed)
# ------------------------------------------------------------------
# Derive all paths at install time - never hard-code an absolute path that
# breaks when the repository is moved (CLAUDE.md "repoint ExecStart" lesson,
# RESEARCH Anti-Pattern).
$TrayScript = Join-Path $RepoRoot "daemon\tray_windows.py"
Log "Registering autostart (HKCU\Software\Microsoft\Windows\CurrentVersion\Run) ..."
# Invoke the autostart helper via the just-created venv python so sys.executable
# resolves to the venv's pythonw.exe (the path that will be written to the registry).
& $PythonExe -c @"
import sys, os
sys.path.insert(0, r'$RepoRoot')
import daemon.autostart_windows as a
a.enable(tray_script=r'$TrayScript')
"@
if ($LASTEXITCODE -ne 0) { throw "Autostart registration failed (exit $LASTEXITCODE)" }
Log "Autostart registered - Clawdmeter will launch automatically at next logon"
# ------------------------------------------------------------------
# Step 4: Launch the tray app (headless - BASE pythonw.exe, no console window)
# ------------------------------------------------------------------
# Use the BASE interpreter's pythonw.exe, NOT the venv's Scripts\pythonw.exe.
# The venv pythonw is a redirector stub that re-launches the CONSOLE python.exe
# build as a child (a CPython venv-launcher bug), popping a black console window.
# tray_windows.py adds the venv site-packages to sys.path itself, so the venv's
# dependencies still resolve. (See autostart_windows._command - same rationale.)
$BasePrefix = & $PythonExe -c "import sys; print(sys.base_exec_prefix)"
$BasePythonw = Join-Path $BasePrefix "pythonw.exe"
Log "Launching tray app ..."
$StartArgs = @{
FilePath = $BasePythonw
ArgumentList = "`"$TrayScript`""
WorkingDirectory = $RepoRoot
}
Start-Process @StartArgs
Log "Tray app started - look for the Clawdmeter icon in your notification area"
Log "=== Install complete ==="