diff --git a/README.md b/README.md index 1f03645..25a1bb5 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..a9e5625 --- /dev/null +++ b/conftest.py @@ -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__)) diff --git a/daemon/README-windows.md b/daemon/README-windows.md new file mode 100644 index 0000000..e32fc1a --- /dev/null +++ b/daemon/README-windows.md @@ -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\\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 diff --git a/daemon/__init__.py b/daemon/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/daemon/autostart_windows.py b/daemon/autostart_windows.py new file mode 100644 index 0000000..3bded88 --- /dev/null +++ b/daemon/autostart_windows.py @@ -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", ) +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 diff --git a/daemon/claude_usage_daemon_windows.py b/daemon/claude_usage_daemon_windows.py new file mode 100644 index 0000000..85e6f32 --- /dev/null +++ b/daemon/claude_usage_daemon_windows.py @@ -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 + # ~5–10s 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) diff --git a/daemon/icon_assets.py b/daemon/icon_assets.py new file mode 100644 index 0000000..42231b8 --- /dev/null +++ b/daemon/icon_assets.py @@ -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} diff --git a/daemon/requirements-windows.txt b/daemon/requirements-windows.txt new file mode 100644 index 0000000..d333f8b --- /dev/null +++ b/daemon/requirements-windows.txt @@ -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 diff --git a/daemon/tests/__init__.py b/daemon/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/daemon/tests/fixtures/credentials_direct.json b/daemon/tests/fixtures/credentials_direct.json new file mode 100644 index 0000000..37ae883 --- /dev/null +++ b/daemon/tests/fixtures/credentials_direct.json @@ -0,0 +1,3 @@ +{ + "accessToken": "sk-ant-test-5678" +} diff --git a/daemon/tests/fixtures/credentials_nested.json b/daemon/tests/fixtures/credentials_nested.json new file mode 100644 index 0000000..90083db --- /dev/null +++ b/daemon/tests/fixtures/credentials_nested.json @@ -0,0 +1,8 @@ +{ + "claudeAiOauth": { + "accessToken": "sk-ant-test-1234", + "refreshToken": "sk-ant-ort-test-5678", + "expiresAt": 9999999999000, + "scopes": ["user:inference", "user:profile"] + } +} diff --git a/daemon/tests/test_windows_autostart.py b/daemon/tests/test_windows_autostart.py new file mode 100644 index 0000000..60c86e3 --- /dev/null +++ b/daemon/tests/test_windows_autostart.py @@ -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'(?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) diff --git a/daemon/tests/test_windows_no_wsl.py b/daemon/tests/test_windows_no_wsl.py new file mode 100644 index 0000000..73cfd0b --- /dev/null +++ b/daemon/tests/test_windows_no_wsl.py @@ -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}" + ) diff --git a/daemon/tests/test_windows_poll.py b/daemon/tests/test_windows_poll.py new file mode 100644 index 0000000..9b61957 --- /dev/null +++ b/daemon/tests/test_windows_poll.py @@ -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)" diff --git a/daemon/tests/test_windows_reconnect.py b/daemon/tests/test_windows_reconnect.py new file mode 100644 index 0000000..be763d7 --- /dev/null +++ b/daemon/tests/test_windows_reconnect.py @@ -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") diff --git a/daemon/tests/test_windows_token.py b/daemon/tests/test_windows_token.py new file mode 100644 index 0000000..73682fa --- /dev/null +++ b/daemon/tests/test_windows_token.py @@ -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}" + ) diff --git a/daemon/tests/test_windows_tray.py b/daemon/tests/test_windows_tray.py new file mode 100644 index 0000000..9daac97 --- /dev/null +++ b/daemon/tests/test_windows_tray.py @@ -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}" diff --git a/daemon/tray_windows.py b/daemon/tray_windows.py new file mode 100644 index 0000000..dac3195 --- /dev/null +++ b/daemon/tray_windows.py @@ -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 +# /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() diff --git a/install-windows.ps1 b/install-windows.ps1 new file mode 100644 index 0000000..6879bc2 --- /dev/null +++ b/install-windows.ps1 @@ -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 ==="