← All Apps
The Community Workbench
Busy Codex
Live Codex task status, account limits and weekly progress, with a reasoning effort dial and START-button Fast mode control for Codex Desktop and a native-control CLI fork.
Updated Sep 14, 2026 · Added Sep 6, 2026 · ↓ 0 package download starts
By wowlocal · ▲ 0 · Rate and Discuss
How It Works
Busy Codex
Codex task state, current account limits and elapsed quota-window progress on a BUSY Bar. Turn the dial to change the reasoning effort of the foreground Codex Desktop task or supported CLI session. The display reflects the confirmed setting; each effort level has its own native animation. Press the large START button to toggle Fast mode: gold warp ignition for Fast, blue deceleration for standard speed. Holding it does not repeat the toggle. Fast uses more plan allowance, just like enabling it inside Codex.
After a task completes, DONE animates for five seconds and then becomes IDLE. Status and quota refreshes do not restart this timer.
This is an independent community app, not an OpenAI or BUSY Bar product.
Run
Download this complete folder. Python 3.9 or later is the only runtime dependency; there are no pip packages to install.
python3 app.py # USB at 10.0.4.20
python3 app.py --host 192.168.1.50 # Wi-Fi
python3 app.py --host 127.0.0.1:8080 --demo # emulator, synthetic data
python3 app.py --test # one synthetic frame, then exit
For an authenticated Wi-Fi Bar, set BUSYBAR_TOKEN in your environment. For a
non-default Codex installation, set CODEX_HOME and/or BUSYBAR_CODEX_BIN.
Authentication stays with the installed Codex executable.
Normal startup uploads this app's native animation assets to the Bar. It does
not install an APPS-menu entry, a login item or a system service. Subsequent
starts may use --no-upload while the packaged assets remain on the device.
The app owns a loopback report endpoint on port 18765; use --port to choose
another. Stop any other busy-codex instance before giving this one the dial.
--no-effort leaves the app in read-only mode. Ctrl-C or SIGTERM stops the
owned workers, releases its canvas and does not restart either process.
Codex compatibility
- Codex Desktop on macOS: follows the task open in the foreground window. Effort uses the Desktop's local IPC interface, which is private and may change between Desktop releases.
- Codex CLI: account limits can be read with stock CLI. Reliable effort switching and terminal session targeting require the native-control fork. It exposes a local, same-user Unix socket with versioned status and confirmed effort changes. No terminal keystroke injection or config-file rewriting is used. Foreground app/terminal selection is currently macOS-specific.
- No Codex installed:
--demoand--testrun without opening Codex files, processes, credentials or network services.
Session state is read from local Codex metadata. Account limits come from
account/rateLimits/read through a short-lived local Codex app-server, refreshed
once a minute. No model inference or prompt submission is performed. Missing
or expired quota data remains unavailable instead of being shown as zero.
Fast switching requires the fork's fast/set endpoint; restart CLI sessions
after updating. The earlier v0.153.4-fork.1-native-control release supports
effort only. The controls preserve Plan mode and do not write global defaults.
Preview and conformance
In a busybar-apps checkout:
npm run check -- busy-codex --run -- --demo
python3 tools/busyrec.py apps/busy-codex --seconds 8 --steal-at 3 -- --demo
npm run preview -- busy-codex --seconds 7.2 --fps 9 -- --demo
npm run build
The preview is actual output from --demo, captured by the gallery recorder.
Demo mode streams PNGs generated by the same pixel renderer as the native
.anim files. Normal mode uploads the animations once and plays them on the
device at 25 fps, avoiding constant frame traffic. The PNG demo transport is
also compatible with the gallery renderer's more limited .anim decoder.
The adjacent Python modules and assets are a reproducible snapshot of
busy-codex, built with
python3 scripts/build_gallery.py --output /path/to/apps/busy-codex.
SOURCE.json records SHA-256 hashes for review. Generated files should be
updated from the upstream project instead of patched independently here.
License
MIT; see LICENSE, including the original Alpharius-003 copyright notice.
The project extends that MIT codebase; its pixel font and effort effects are
included as source. No account details, access tokens or personal configuration
are included.
Source Files
These are the mirrored source bytes. The complete package also includes every asset and supporting file.
README.md
# Busy Codex
Codex task state, current account limits and elapsed quota-window
progress on a BUSY Bar. Turn the dial to change the reasoning effort of the
foreground Codex Desktop task or supported CLI session. The display reflects
the confirmed setting; each effort level has its own native animation.
Press the large START button to toggle Fast mode: gold warp ignition for Fast,
blue deceleration for standard speed. Holding it does not repeat the toggle.
Fast uses more plan allowance, just like enabling it inside Codex.
After a task completes, DONE animates for five seconds and then becomes IDLE.
Status and quota refreshes do not restart this timer.
This is an independent community app, not an OpenAI or BUSY Bar product.
## Run
Download this **complete folder**. Python 3.9 or later is the only runtime
dependency; there are no pip packages to install.
```sh
python3 app.py # USB at 10.0.4.20
python3 app.py --host 192.168.1.50 # Wi-Fi
python3 app.py --host 127.0.0.1:8080 --demo # emulator, synthetic data
python3 app.py --test # one synthetic frame, then exit
```
For an authenticated Wi-Fi Bar, set `BUSYBAR_TOKEN` in your environment. For a
non-default Codex installation, set `CODEX_HOME` and/or `BUSYBAR_CODEX_BIN`.
Authentication stays with the installed Codex executable.
Normal startup uploads this app's native animation assets to the Bar. It does
not install an APPS-menu entry, a login item or a system service. Subsequent
starts may use `--no-upload` while the packaged assets remain on the device.
The app owns a loopback report endpoint on port 18765; use `--port` to choose
another. Stop any other busy-codex instance before giving this one the dial.
`--no-effort` leaves the app in read-only mode. Ctrl-C or SIGTERM stops the
owned workers, releases its canvas and does not restart either process.
## Codex compatibility
- **Codex Desktop on macOS:** follows the task open in the foreground window.
Effort uses the Desktop's local IPC interface, which is private and may
change between Desktop releases.
- **Codex CLI:** account limits can be read with stock CLI. Reliable effort
switching and terminal session targeting require the
[native-control fork](https://github.com/wowlocal/codex/tree/codex/native-tui-control).
It exposes a local, same-user Unix socket with versioned status and confirmed
effort changes. No terminal keystroke injection or config-file rewriting is
used. Foreground app/terminal selection is currently macOS-specific.
- **No Codex installed:** `--demo` and `--test` run without opening Codex files,
processes, credentials or network services.
Session state is read from local Codex metadata. Account limits come from
`account/rateLimits/read` through a short-lived local Codex app-server, refreshed
once a minute. No model inference or prompt submission is performed. Missing
or expired quota data remains unavailable instead of being shown as zero.
Fast switching requires the fork's `fast/set` endpoint; restart CLI sessions
after updating. The earlier `v0.153.4-fork.1-native-control` release supports
effort only. The controls preserve Plan mode and do not write global defaults.
## Preview and conformance
In a busybar-apps checkout:
```sh
npm run check -- busy-codex --run -- --demo
python3 tools/busyrec.py apps/busy-codex --seconds 8 --steal-at 3 -- --demo
npm run preview -- busy-codex --seconds 7.2 --fps 9 -- --demo
npm run build
```
The preview is actual output from `--demo`, captured by the gallery recorder.
Demo mode streams PNGs generated by the same pixel renderer as the native
`.anim` files. Normal mode uploads the animations once and plays them on the
device at 25 fps, avoiding constant frame traffic. The PNG demo transport is
also compatible with the gallery renderer's more limited `.anim` decoder.
The adjacent Python modules and assets are a reproducible snapshot of
[busy-codex](https://github.com/wowlocal/busy-codex), built with
`python3 scripts/build_gallery.py --output /path/to/apps/busy-codex`.
`SOURCE.json` records SHA-256 hashes for review. Generated files should be
updated from the upstream project instead of patched independently here.
## License
MIT; see `LICENSE`, including the original Alpharius-003 copyright notice.
The project extends that MIT codebase; its pixel font and effort effects are
included as source. No account details, access tokens or personal configuration
are included.SOURCE.json
{
"repository": "https://github.com/wowlocal/busy-codex",
"revision": "4d6c8dcde74df38f2bdef75bdb70383f9774ba99",
"working_tree_changes": false,
"generator": "scripts/build_gallery.py",
"files": {
"LICENSE": "41e9217958abde104076431eb88a0d406d26b9384f75cede965382e90d0a6d73",
"README.md": "9c5500357a9d21de0714db438ede69fba33ee132b28e38f2137b15b49b4ee3ac",
"adapters/codex_status.py": "e33f16e1384f37eef6f3b045ddcdc99350da18dfbc8e1a45a0b6da8aa9e25275",
"ai_status.py": "a1d355d545d3670cb2ad78608abb641f43a5e0dc8ef3d4b755aea8f8c1917469",
"app.py": "74cb8afdb1954239d27c7be9193d275b7e2eeeeb9d28b83117ade479ffe1f312",
"assets/astra.anim": "f8e036b59e2ce59e9cb16124a1dc4849b911b7210947e0407127ed7f4202b818",
"assets/done.anim": "aa9ac7ea9eb4ab490c7c8c9c72383d47ba2995429f5e1c361b3cbf8a8466b8fa",
"assets/effort_clear.anim": "72408efe98b924fc4497678eda1835f2239d6ffed019be916361c2d997244c3a",
"assets/effort_v4_high_down_change.anim": "cc14c1065aa90533cf702d4bfc418fbe18642902d839d6f758554dc4ccf15c12",
"assets/effort_v4_high_down_in.anim": "635d8e78f7eaf4d7f2a07dccb2f71365a08b29a94a9c34d96521e01dd299a114",
"assets/effort_v4_high_up_change.anim": "0145dff48713f3fe50e4f29d92d8a198b08bde2bbf6ff9943e74b656225074c3",
"assets/effort_v4_high_up_in.anim": "42d7c1e0571b0ed259a1472cf06d0fa3cc948da1079767699caf2773451a3aa4",
"assets/effort_v4_low_down_change.anim": "e455fac32d35f594990d346781aada814b3a1de97887ac741d6b2aac2cb64b3f",
"assets/effort_v4_low_down_in.anim": "4567b1d3ce3cb648db61f9f3b85a2d1fd3e4806e2c52672c884bc26552786d55",
"assets/effort_v4_low_up_change.anim": "4e685ef3e51f202ddd978019906bcca8e78a4a4cd882891e4efcac55df40cb3d",
"assets/effort_v4_low_up_in.anim": "a57cae863649fb0904cb4f8bb0e55ba7ad4b42b868bb435da5a30645b3f5cd03",
"assets/effort_v4_max_down_change.anim": "46f5e15a6e31b63086bbbe6aa90508fef563ce14c1a6780f373dad23f52c9bc3",
"assets/effort_v4_max_down_in.anim": "36c15cfee4e41074b8d16c65b825b5599966c8d31551ac5949489d4ce380dcde",
"assets/effort_v4_max_up_change.anim": "1fe19f3ff180b36e9e9b9ae8ebb5d0d651ea95a3543e1e4d2d5240fb8aba966e",
"assets/effort_v4_max_up_in.anim": "f3323e604eb354e568348a2848647f6ebfda639224735b0c90bd13c48d103a3f",
"assets/effort_v4_medium_down_change.anim": "84db12a5cdff6d686424c5476c2da61729b145974c5a4470ded0f4b803333b54",
"assets/effort_v4_medium_down_in.anim": "282a2139870ed780f226a0188702fe5a007b8fc9181267ab95e158970d840770",
"assets/effort_v4_medium_up_change.anim": "11e70e09680703e125f866c2caaa206420eb079315d2c94b74a46db12df2de74",
"assets/effort_v4_medium_up_in.anim": "afcc2f93942a037cf32a0a5bf99940c50b4518b757e9822d0befec1caf8a924e",
"assets/effort_v4_minimal_down_change.anim": "fab080e4df2ca48358f65b3312780f8a8b1318e1a443c54a5f984cf3ca7dc885",
"assets/effort_v4_minimal_down_in.anim": "1435d484ad23ddcdd1e2800af5ad53930aad9b75a0c99b009481ebea185a78af",
"assets/effort_v4_minimal_up_change.anim": "c27c01c6ea0edb9274f59f4542f479069d011b6806f7304fed5b04b302eca2fd",
"assets/effort_v4_minimal_up_in.anim": "297b19e475e6f07e9f435e1eec5a3d4d44851795ec412d98855ef7355aace053",
"assets/effort_v4_none_down_change.anim": "8ba4d9b0f97e776879c96d056855d202665bcfa0d5ae0e2f22ddd0a47eb819c0",
"assets/effort_v4_none_down_in.anim": "e1d61bc6342eee677502554a25e298c294c7aee2d6255ab9cf3ea948ef98d5c4",
"assets/effort_v4_none_up_change.anim": "68045b14e12056726bc01f4c2d8f95324889be23a26636dce408b0bc1bfe1b78",
"assets/effort_v4_none_up_in.anim": "f84d24a70d240d9738f81f12bf369f23e6a7ff273bf7fb397329cf11cdab3ded",
"assets/effort_v4_ultra_down_change.anim": "7909ade81ec0cefe6ec2c62d98e034dbbcd999139d38087148ca3a107ccac34b",
"assets/effort_v4_ultra_down_in.anim": "0a6bb9ddb1c5f8dd69c0922b6e7bf65b6d43d9bb3b02199ad137c70bde70b7f7",
"assets/effort_v4_ultra_up_change.anim": "96346b11ab01139a7bdf4cf2967f6361d0e73de772272c27e1ab3d6894c9f74d",
"assets/effort_v4_ultra_up_in.anim": "43ebdd7e5218fe7ee3e440cf272827fa177f2aa6fd21a5d8ef2705eee4478765",
"assets/effort_v4_xhigh_down_change.anim": "37d51e8521538867ac399bd1308d1bfafc52b18c62a084a4b924e60eb0818dc0",
"assets/effort_v4_xhigh_down_in.anim": "0b26e19a917e781c01b55e7d5d627c0a72564e763f8f18a36cf8e83a43659ddd",
"assets/effort_v4_xhigh_up_change.anim": "c2650346ec9a6c283d4572452a326d4ba7b7b50e13d63dd6d278224c08a8247b",
"assets/effort_v4_xhigh_up_in.anim": "e2469dd25f52d62f693f03bd1b75c922ee63fd9211774f81e8e70f2751c10610",
"assets/error.anim": "00d0d4f58804b0bedfdf2187ad82dbdb0f82207cae7e7b80447788a05a6774b7",
"assets/fast_v1_off_change.anim": "f21a3e387724860428e39958ee9efd15bfa457d66d0eb5201542554ee459ae26",
"assets/fast_v1_off_in.anim": "6d3450b7a52f147b2db36ba66895277c2926f013b622dd830abe4f1f1b934cc8",
"assets/fast_v1_on_change.anim": "773e974890d0db159c6df814651eb0a9966a54fd8c062bce2e740b0f943871fe",
"assets/fast_v1_on_in.anim": "c114d4a25cccfd775e72e7a6d8affa8214d2c4537d700e0db5edbbf524161189",
"assets/idle.anim": "527c0888e9fd8a4912bdbeaa9c9038a8365adfa295db85bde6bb15f56bb942c6",
"assets/think.anim": "a880d4d2483d98e3f800541fd410e12538523b269ebeea3c1afcf64be3317c97",
"assets/wait.anim": "434b3d31376de37f533a34718bdcdd9e8afd0aea80ec52331ad5e13f55b72aab",
"assets/work.anim": "d9c9023f71dd81886b76bd62cfd617351ca49ec54ab879946c42bc67b5a026bc",
"assets/work_fast.anim": "50d37abb3ac55219e1e9dd400790277917817f9ba08693e2587931a293dd141d",
"busybar_http.py": "b36766dc37c6dfbef010290c679d137ef36a5cfdff08675ff32d8e2f7ac3e362",
"busybar_input.py": "7ea6154d6f6b6d728e22fd3d33a2304aa513599aa8f5169644343c3956cd4c0d",
"codex_cli_client.py": "2a500fb474652c1b8d06517c2dbb072fe33bb08c8fe251ba64e1402a76c8ae76",
"codex_cli_native.py": "e2b24c28152d301967c14be2a48e4fa57d9ab7f7844631b4a1893c4a20c7edde",
"codex_effort.py": "10a51775184f7b9594dd273156e254301290de6443f49db5bab60944c9e8b0dc",
"codex_fast.py": "05edd7af9d34499abda32e424b8338268c341fa8d7fb80ebfad68315827bcb39",
"codex_focus.py": "7dc54bfbe66a2b592e7cea1ca9f926e81a4a5294e24dbefd0cb13cfb2f17c0bf",
"codex_target.py": "6f5cb4e86058cff29379867b4986f3200a88b5eb315846a0ce434d4adfad800a",
"codex_usage.py": "91f821bf535d76dcbff6180beb3cad334042b352ea9f91b7534827780a813321",
"daemon.py": "10ca999adee5d8568434b4ee6bfc408cc6f31376e19ec3f9cf0c6e5eb9e8541b",
"display_scene.py": "00a6c6356159366d7dd07d24e845e22b05e11226b8dc0337dccf79fe1410e812",
"effort_animation.py": "40b8fc16bedb86be1f5a5cb28d9100f4169ccd59818fff534cad1c9c57459d79",
"fast_animation.py": "a3caabe3619a392a9880d02c10a2dc2653e28ae0895718631b961a516cbee125",
"manifest.yaml": "cbf5b60260f22a5c9af8c64e58337b9a08cb5597ab0db605246ebca0a49d2063",
"pixel_fonts.py": "b95a90b0ab824088448e41b319e3f579dbbd31e97c4ff242b8957f3bf03a7bb2",
"pixel_ui.py": "3b7c7e992aba40f4ed5841f622e88d5726390805476bf272e1514966099614ea",
"preview.gif": "45d86a0d240bdb66e056ff1bd56b23da11b3fcac62bcec262ab00e190f0c4707",
"report.py": "2b579fa39940a9cdf567101f7d45fbc6ccded2a409d84095ebe4f5dd776f0729"
}
}adapters/codex_status.py
#!/usr/bin/env python3
"""Codex Desktop/CLI adapter for the busybar daemon.
Zero-config and rename-proof: everything is derived from what Codex
itself reports (config defaults, the selected task's rollout and the
account/rateLimits/read API). No model-name tables anywhere:
- label: the raw model id, prettified by GENERIC rules only
("gpt-5.6-sol" -> "5.6 Sol"; a future "gpt-7-luna" ->
"7 Luna" with zero changes here), plus the reasoning effort
- badges: service_tier other than default becomes a badge
("fast" selects the yellow high-speed working contour)
- context: last_token_usage.total_tokens / model_context_window
- quotas: fresh account windows, independent of selected task/activity
(600 -> "10h", 10080 -> "7d") - names survive plan changes
- state: Codex task_started / task_complete lifecycle events
Usage:
python3 adapters/codex_status.py # watch task + refresh quotas every minute
python3 adapters/codex_status.py --once -v # single probe, print it
"""
from __future__ import annotations
import json
import pathlib
import re
import signal
import sys
import threading
import time
import urllib.request
sys.path.insert(0, str(pathlib.Path(__file__).resolve().parent.parent))
import report # noqa: E402 (daemon/hub address + host headers from env.sh)
import codex_focus
import codex_target
import codex_usage
DAEMON = report.BASE + "/v1/report"
CODEX_HOME = pathlib.Path(report.ENV.get("CODEX_HOME", pathlib.Path.home() / ".codex"))
SESSIONS = CODEX_HOME / "sessions"
TARGET = codex_target.Target(CODEX_HOME)
QUIET_RECHECK_S = 8 # one final read after the rollout stops changing
COMPLETE_S = 120 # fallback for old rollouts without lifecycle events
POLL_S = 0.25
VENDOR_PREFIXES = ("gpt-", "chatgpt-", "openai-")
def _empty_snapshot() -> dict:
return {
"model": None, "effort": None, "tier": None,
"info": None, "state": None,
}
# The long-running adapter reads only newly appended JSONL records. A --once
# process starts with an empty cache and scans the current rollout once.
_ROLLOUT_CACHE = {
"path": None,
"offset": 0,
"snapshot": _empty_snapshot(),
}
def prettify_model(model: str) -> str:
"""Generic prettifier - strips vendor prefixes and title-cases word
tokens. Never matches specific model names."""
m = model.lower()
for p in VENDOR_PREFIXES:
if m.startswith(p):
model = model[len(p):]
break
parts = []
for tok in re.split(r"[-_]", model):
parts.append(tok if any(c.isdigit() for c in tok) else tok.capitalize())
return " ".join(p for p in parts if p)
def config_defaults() -> dict:
"""Minimal TOML pluck of the keys we need (no toml dependency)."""
out = {}
try:
for line in (CODEX_HOME / "config.toml").read_text().splitlines():
m = re.match(r'\s*(model|model_reasoning_effort|service_tier)\s*=\s*"([^"]*)"', line)
if m:
out[m.group(1)] = m.group(2)
except OSError:
pass
return out
_FOCUSED_ROLLOUTS = {}
def selected_target():
pinned = report.ENV.get("BUSYBAR_CODEX_THREAD_ID", "")
if pinned:
record = next((r for r in codex_target.cli_sessions(CODEX_HOME)
if r.get('thread_id') == pinned), None)
return {**record, 'kind': 'cli'} if record else {'kind': 'desktop', 'thread_id': pinned}
return TARGET.display()
def newest_rollout(selection=None) -> pathlib.Path | None:
target = (selection if selection is not None else selected_target()).get('thread_id')
if target:
path = _FOCUSED_ROLLOUTS.get(target)
if path and path.exists():
return path
path = next(SESSIONS.glob(f"*/*/*/rollout-*-{target}.jsonl"), None)
if path:
_FOCUSED_ROLLOUTS[target] = path
return path
return None
def _apply_event(snapshot: dict, event: dict) -> None:
"""Merge one structured rollout event into the latest known snapshot."""
payload = event.get("payload")
if not isinstance(payload, dict):
return
if event.get("type") == "turn_context":
for source_key, target_key in (("model", "model"), ("effort", "effort"),
("service_tier", "tier")):
value = payload.get(source_key)
if isinstance(value, str) and value:
snapshot[target_key] = value
return
if event.get("type") != "event_msg":
return
event_type = payload.get("type")
if event_type == "task_started":
snapshot["state"] = "WORKING"
elif event_type in ("task_complete", "turn_aborted", "task_cancelled", "task_canceled"):
snapshot["state"] = "COMPLETE"
elif event_type == "token_count":
if isinstance(payload.get("info"), dict):
snapshot["info"] = payload["info"]
def _rollout_snapshot(path: pathlib.Path) -> dict:
"""Incrementally parse JSONL and retain the latest structured fields."""
stat = path.stat()
if (_ROLLOUT_CACHE["path"] != path or stat.st_size < _ROLLOUT_CACHE["offset"]):
_ROLLOUT_CACHE.update({
"path": path,
"offset": 0,
"snapshot": _empty_snapshot(),
})
with path.open("rb") as stream:
stream.seek(_ROLLOUT_CACHE["offset"])
while True:
position = stream.tell()
raw = stream.readline()
if not raw:
break
try:
event = json.loads(raw)
except (json.JSONDecodeError, UnicodeDecodeError):
# Do not advance past a record while Codex is still appending it.
if not raw.endswith(b"\n"):
stream.seek(position)
break
_ROLLOUT_CACHE["offset"] = stream.tell()
continue
if isinstance(event, dict):
_apply_event(_ROLLOUT_CACHE["snapshot"], event)
_ROLLOUT_CACHE["offset"] = stream.tell()
return dict(_ROLLOUT_CACHE["snapshot"])
def probe(usage=None, selection=None) -> dict | None:
target = selected_target() if selection is None else selection
rollout = newest_rollout(target)
defaults = config_defaults()
if rollout is None and not defaults and not target.get('model'):
return None
model = defaults.get("model")
effort = defaults.get("model_reasoning_effort")
tier = defaults.get("service_tier")
context_pct = None
state = "IDLE"
session_id = target.get('thread_id') or 'config'
if rollout is not None:
age = time.time() - rollout.stat().st_mtime
snapshot = _rollout_snapshot(rollout)
state = snapshot["state"] or (
"WORKING" if age < QUIET_RECHECK_S else
("COMPLETE" if age < COMPLETE_S else "IDLE")
)
model = snapshot["model"] or model
effort = snapshot["effort"] or effort
tier = snapshot["tier"] or tier
info = snapshot["info"]
if info:
window = info.get("model_context_window")
last = (info.get("last_token_usage") or {}).get("total_tokens")
if window and last:
context_pct = round(min(100, last * 100 / window), 1)
if target.get('kind') == 'cli':
model, effort = target.get('model'), target.get('effort')
state, context_pct = target.get('state', 'IDLE'), target.get('context_pct')
tier = next(iter(target.get('badges') or []), None)
if not model:
return None
if tier == "priority":
tier = "fast"
label = prettify_model(model)
badges = None
if tier and tier not in ("default", "standard"):
badges = [tier]
if tier != "fast": # unknown tiers also get spelled out
label += f" {tier}"
if effort:
label += f" {effort}"
# A newly created Desktop task may not have a rollout yet. Its native
# owner and settings snapshot are verified by the controller; history is
# display data, not evidence that a task can accept settings.
control_id = None
if target.get('kind') == 'desktop' and re.fullmatch(codex_focus.UUID, target.get('thread_id') or ''):
control_id = target['thread_id']
elif target.get('kind') == 'cli' and target.get('ready'):
control_id = target['thread_id']
return {
"source": "codex", "session_id": session_id, "state": state,
"control_thread_id": control_id,
"label": label, "context_pct": context_pct,
"badges": badges, "ttl_s": 600,
**(usage or {}),
}
def report_headers() -> dict:
return dict(report.HEADERS)
def post(report: dict) -> bool:
try:
urllib.request.urlopen(urllib.request.Request(
DAEMON, data=json.dumps(report).encode(), method="POST",
headers=report_headers()), timeout=2).read()
return True
except OSError:
return False
def _emit(verbose: bool, usage=None, selection=None):
report = probe(usage, selection)
if report:
if verbose:
print(json.dumps(report, ensure_ascii=False), flush=True)
return post(report)
elif verbose:
print("no codex data found", flush=True)
return False
def main():
once = "--once" in sys.argv
verbose = "-v" in sys.argv
if once:
# Notify hooks leave account polling to the existing background adapter.
# A manual --once still provides a fresh, self-contained diagnostic.
usage = None
if "--no-usage-refresh" not in sys.argv:
monitor = codex_usage.Monitor(report.ENV)
monitor.refresh()
usage = monitor.snapshot()
_emit(verbose, usage)
return
monitor = codex_usage.Monitor(report.ENV)
stop = threading.Event()
signal.signal(signal.SIGTERM, lambda *_: stop.set())
signal.signal(signal.SIGINT, lambda *_: stop.set())
worker = threading.Thread(target=monitor.run, args=(stop,), daemon=True)
worker.start()
try:
watch(monitor, stop, verbose)
finally:
stop.set()
worker.join(timeout=4)
def watch(monitor, stop, verbose):
# Quota updates also trigger a report while the selected task is idle.
last_stamp = None
previous = None
retired = set()
last_report_at = 0
quiet_rechecked = False
while not stop.is_set():
target = selected_target()
rollout = newest_rollout(target)
stat = rollout.stat() if rollout else None
session_id = target.get('thread_id') or 'config'
stamp = (session_id, target.get('kind'), target.get('ready'), target.get('model'),
target.get('effort'), target.get('state'), tuple(target.get('badges') or []),
target.get('context_pct'), stat.st_mtime_ns if stat else None, stat.st_size if stat else None)
if (stamp != last_stamp or monitor.changed.is_set()
or time.time() - last_report_at >= 20):
# Publish the replacement before retiring the previous session.
# Otherwise the renderer sees an empty store and clears its canvas,
# exposing the firmware menu between Desktop and CLI tasks.
if _emit(verbose, monitor.snapshot(), target):
if previous and previous != session_id:
retired.add(previous)
retired.discard(session_id)
previous = session_id
monitor.changed.clear()
last_stamp = stamp
last_report_at = time.time()
quiet_rechecked = False
elif (not quiet_rechecked and stat is not None
and time.time() - stat.st_mtime > QUIET_RECHECK_S):
quiet_rechecked = _emit(verbose, monitor.snapshot(), target)
for session in tuple(retired):
if post({"source": "codex", "session_id": session, "ended": True}):
retired.remove(session)
stop.wait(POLL_S)
if __name__ == "__main__":
main()ai_status.py
#!/usr/bin/env python3
"""Network-only AI provider outage overlay for the BUSY Bar.
Health data comes from AIWatch's public, unauthenticated JSON API. The
monitor keeps its snapshot in memory and owns a separate high-priority
canvas, so the normal Codex/Claude dashboard remains visible while every
provider is healthy and reappears automatically after an incident clears.
"""
from __future__ import annotations
import json
import threading
import time
import urllib.request
from dataclasses import dataclass
from busybar_input import InputStream, input_stream_url, parse_input_events
API_URL = "https://aiwatch-worker.p2c2kbf.workers.dev/api/v1/status"
X_STATUS_URL = "https://isupmap.com/api/status"
GOOGLE_STATUS_URL = "https://www.google.com/generate_204"
APP_NAME = "ai_provider_status"
DRAW_PRIORITY = 80
POLL_S = 60.0
ROTATE_S = 4.0
KEEPALIVE_S = 8.0
ANIM_REFRESH_S = 60.0
SOURCE_MAX_AGE_S = 20 * 60.0
DIRECT_FAILURES_TO_ALERT = 2
MANUAL_HOLD_S = ROTATE_S
TEXT_TIMEOUT_S = 15
ANIM_TIMEOUT_S = 120
FONT = "small"
BLACK = "#000000FF"
WHITE = "#FFFFFFFF"
DIM = "#777777FF"
HEALTHY = "#20C040FF"
DEGRADED = "#FFB000FF"
DOWN = "#FF2020FF"
STATUS_RANK = {"degraded": 1, "down": 2}
@dataclass(frozen=True)
class Provider:
name: str
services: tuple[tuple[str, str], ...]
# Each provider is one alert even when several of its surfaces fail together.
# The short labels are deliberate: the physical display is only 72 pixels wide.
PROVIDERS = (
Provider("OPENAI", (("openai", "API"), ("chatgpt", "CHAT"), ("codex", "CODE"))),
Provider("ANTHROPIC", (("claude", "API"), ("claudeai", "CHAT"),
("claudecode", "CODE"))),
Provider("GEMINI", (("gemini", "API"),)),
Provider("OPENRTR", (("openrouter", "API"),)),
Provider("DEEPSEEK", (("deepseek", "API"),)),
Provider("MISTRAL", (("mistral", "API"),)),
Provider("PERPLEX", (("perplexity", "API"),)),
)
PINNED_AI_PROVIDER = "ANTHROPIC"
def _iso_timestamp(value) -> float | None:
if not isinstance(value, str) or not value:
return None
try:
# AIWatch emits UTC timestamps with a trailing Z.
from datetime import datetime
return datetime.fromisoformat(value.replace("Z", "+00:00")).timestamp()
except (ValueError, OverflowError):
return None
def provider_alerts(payload: dict, now: float | None = None) -> list[dict]:
"""Collapse service-level AIWatch data into provider-level alerts.
Unknown and stale observations never become outages: loss of the monitor
itself must not look like loss of an AI provider.
"""
now = time.time() if now is None else now
services = payload.get("services") if isinstance(payload, dict) else None
if not isinstance(services, list):
raise ValueError("AIWatch response has no services list")
by_id = {
str(service.get("id")): service
for service in services
if isinstance(service, dict) and service.get("id")
}
alerts = []
for order, provider in enumerate(PROVIDERS):
affected: list[tuple[str, str]] = []
for service_id, surface in provider.services:
service = by_id.get(service_id)
if not service:
continue
checked = _iso_timestamp(service.get("lastChecked"))
if checked is None or abs(now - checked) > SOURCE_MAX_AGE_S:
continue
status = str(service.get("status") or "").lower()
if status in STATUS_RANK:
affected.append((status, surface))
if not affected:
continue
worst = max((status for status, _ in affected), key=STATUS_RANK.__getitem__)
surfaces = []
for _, surface in affected:
if surface not in surfaces:
surfaces.append(surface)
alerts.append({
"provider": provider.name,
"status": worst,
"surfaces": surfaces,
"order": order,
})
return sorted(alerts, key=lambda a: (-STATUS_RANK[a["status"]], a["order"]))
def provider_status_item(payload: dict, provider_name: str,
now: float | None = None) -> dict | None:
"""Return one provider's current state, including a healthy state."""
now = time.time() if now is None else now
services = payload.get("services") if isinstance(payload, dict) else None
if not isinstance(services, list):
raise ValueError("AIWatch response has no services list")
provider = next((item for item in PROVIDERS if item.name == provider_name), None)
if provider is None:
raise ValueError(f"unknown provider {provider_name}")
by_id = {
str(service.get("id")): service
for service in services
if isinstance(service, dict) and service.get("id")
}
observations: list[tuple[str, str]] = []
for service_id, surface in provider.services:
service = by_id.get(service_id)
if not service:
continue
checked = _iso_timestamp(service.get("lastChecked"))
if checked is None or abs(now - checked) > SOURCE_MAX_AGE_S:
continue
status = str(service.get("status") or "").lower()
if status in STATUS_RANK or status in ("operational", "up"):
observations.append((status, surface))
if not observations:
return None
affected = [(status, surface) for status, surface in observations
if status in STATUS_RANK]
if affected:
status = max((state for state, _ in affected), key=STATUS_RANK.__getitem__)
surfaces = list(dict.fromkeys(surface for _, surface in affected))
else:
status, surfaces = "operational", ["ALL"]
return {
"provider": provider.name,
"status": status,
"surfaces": surfaces,
"order": PROVIDERS.index(provider),
}
def x_status_item(payload: dict, now: float | None = None) -> dict | None:
"""Read X.com health plus its Downdetector-style community surge signal."""
now = time.time() if now is None else now
services = payload.get("services") if isinstance(payload, dict) else None
updated_ms = payload.get("updatedAt") if isinstance(payload, dict) else None
if not isinstance(services, list) or not isinstance(updated_ms, (int, float)):
raise ValueError("isUpMap response has no services list or update time")
if abs(now - updated_ms / 1000.0) > SOURCE_MAX_AGE_S:
return None
service = next((item for item in services
if isinstance(item, dict) and item.get("id") == "x"), None)
if not service:
raise ValueError("isUpMap response has no X service")
status = str(service.get("status") or "").lower()
surge = service.get("surge") is True
# A community-report spike while the HTTP probe still passes is degraded,
# not down. Confirmed probe failures retain isUpMap's severity.
if status in STATUS_RANK:
display_status = status
surfaces = ["WEB"]
elif surge:
display_status = "degraded"
surfaces = ["REPORTS"]
elif status in ("up", "operational"):
display_status = "operational"
surfaces = ["WEB"]
else:
return None
return {
"provider": "X.COM",
"status": display_status,
"surfaces": surfaces,
"order": len(PROVIDERS),
}
def x_status_alerts(payload: dict, now: float | None = None) -> list[dict]:
item = x_status_item(payload, now=now)
return [item] if item and item["status"] in STATUS_RANK else []
def google_status_item(consecutive_failures: int) -> dict:
"""Represent Google's debounced direct-probe state for the carousel."""
return {
"provider": "GOOGLE",
"status": ("down" if consecutive_failures >= DIRECT_FAILURES_TO_ALERT
else "operational"),
"surfaces": ["WEB"],
"order": len(PROVIDERS) + 1,
}
def google_status_alerts(consecutive_failures: int) -> list[dict]:
"""Alert only after repeated direct failures; one timeout is just noise."""
item = google_status_item(consecutive_failures)
return [item] if item["status"] in STATUS_RANK else []
def carousel_items(alerts: list[dict], pinned: list[dict]) -> list[dict]:
"""Put incidents first, followed by pinned healthy services once each."""
active = {item["provider"] for item in alerts}
return [*alerts, *(item for item in pinned if item["provider"] not in active)]
def surface_label(surfaces: list[str]) -> str:
if not surfaces:
return ""
if len(surfaces) == 1:
return surfaces[0]
return f"{surfaces[0]}+{len(surfaces) - 1}"
def _rect(eid, x, y, w, h, color, timeout=TEXT_TIMEOUT_S):
return {"id": eid, "type": "rectangle", "display": "front",
"x": x, "y": y, "width": w, "height": h,
"border_width": 0, "fill": "solid", "fill_colors": [color],
"timeout": timeout}
def _text(eid, x, y, align, text, color):
return {"id": eid, "type": "text", "display": "front",
"x": x, "y": y, "align": align, "text": text,
"font": FONT, "color": color, "timeout": TEXT_TIMEOUT_S}
def animation_path(status: str) -> str:
if status == "down":
return "outage.anim"
if status == "degraded":
return "degraded.anim"
return "healthy.anim"
def alert_animation(status: str) -> list[dict]:
path = animation_path(status)
# The opaque background is created before the transparent contour and
# masks the lower-priority Codex canvas without repainting it.
return [
_rect("alert_bg", 0, 0, 72, 16, BLACK, ANIM_TIMEOUT_S),
{"id": "alert_ring", "type": "animation", "display": "front",
"x": 0, "y": 0, "path": path, "loop": True,
"timeout": ANIM_TIMEOUT_S},
]
def alert_elements(alert: dict, index: int, total: int) -> list[dict]:
status = alert["status"]
if status == "down":
color, marker, word = DOWN, "!", "DOWN"
elif status == "degraded":
color, marker, word = DEGRADED, "!", "DEGR"
else:
color, marker, word = HEALTHY, "+", "OK"
# Keep the warning word compact so the affected surface has a clean gap.
return [
# A colored marker reads more cleanly than a tiny raster lightning bolt.
_text("alert_mark", 3, 0, "top_left", marker, color),
_text("alert_provider", 9, 0, "top_left", alert["provider"], WHITE),
_text("alert_index", 69, 0, "top_right", f"{index + 1}/{total}", DIM),
_text("alert_state", 3, 15, "bottom_left", word, color),
_text("alert_surface", 69, 15, "bottom_right",
surface_label(alert.get("surfaces") or []), DIM),
]
class Monitor:
"""Poll AIWatch and maintain the independent outage canvas."""
def __init__(self, transport, render_lock: threading.Lock, *,
url: str = API_URL, poll_s: float = POLL_S,
x_url: str = X_STATUS_URL,
google_url: str = GOOGLE_STATUS_URL,
input_url: str = "",
input_source_address: str | None = None,
should_render=lambda: True, logger=lambda _msg: None,
opener=None):
self.transport = transport
self.render_lock = render_lock
self.url = url
self.x_url = x_url
self.google_url = google_url
self.input_url = input_url
self.input_source_address = input_source_address
self.poll_s = max(15.0, poll_s)
self.should_render = should_render
self.logger = logger
# Ignore proxy environment variables just like the device/LAN transport;
# they are commonly stale in desktop launch environments.
self.opener = opener or urllib.request.build_opener(urllib.request.ProxyHandler({}))
self.alerts: list[dict] = []
self.pinned: list[dict] = []
self.last_success: float | None = None
self.last_error = ""
self.drawn = False
self.google_failures = 0
self.google_last_ok: float | None = None
self.input_connected = False
self.input_error = ""
self.last_input = ""
self.selected_index: int | None = None
self.manual_until = 0.0
self.displayed_index: int | None = None
self.displayed_provider = ""
self.displayed_at: float | None = None
self.draw_attempts = 0
self.draw_rejections = 0
self.last_draw_ms: float | None = None
self.last_draw_status: int | None = None
self.refresh_requested = threading.Event()
self.redraw_requested = threading.Event()
self._lock = threading.Lock()
def handle_input_event(self, event: tuple, now: float | None = None) -> bool:
"""Apply one BUSY input event. Returns whether it changed this view."""
now = time.time() if now is None else now
direction = 0
action = ""
refresh = False
if event[0] == "encoder":
direction = 1 if event[1] > 0 else -1
action = "NEXT" if direction > 0 else "PREV"
elif event[0] == "button" and event[2] == 0: # PRESS only
if event[1] == 0: # OK
refresh, action = True, "REFRESH"
elif event[1] == 2: # START
direction, action = 1, "NEXT"
if not action:
return False
with self._lock:
items = carousel_items(self.alerts, self.pinned)
if not self.alerts or not items:
return False
if refresh:
self.selected_index = 0
else:
if self.selected_index is not None and now < self.manual_until:
current = self.selected_index % len(items)
elif self.displayed_index is not None:
current = self.displayed_index % len(items)
else:
current = 0
self.selected_index = (current + direction) % len(items)
self.manual_until = now + MANUAL_HOLD_S
self.last_input = action
if refresh:
self.refresh_requested.set()
self.redraw_requested.set()
return True
def _input_loop(self, stop: threading.Event):
def update_state(connected: bool, error: str):
with self._lock:
self.input_connected = connected
self.input_error = error
InputStream(
self.input_url, self.handle_input_event,
source_address=self.input_source_address,
on_state=update_state, logger=self.logger,
).run(stop)
def fetch(self, now: float | None = None) -> tuple[list[dict], list[str]]:
def get_json(url):
request = urllib.request.Request(
url,
headers={"Accept": "application/json",
"User-Agent": "busy-codex-ai-status/1"},
)
with self.opener.open(request, timeout=10) as response:
return json.loads(response.read())
alerts = []
pinned = []
errors = []
successful_sources = 0
try:
ai_payload = get_json(self.url)
alerts.extend(provider_alerts(ai_payload, now=now))
anthropic_item = provider_status_item(
ai_payload, PINNED_AI_PROVIDER, now=now,
)
except (OSError, ValueError, json.JSONDecodeError) as exc:
errors.append(f"AIWatch: {type(exc).__name__}: {exc}"[:160])
else:
successful_sources += 1
if anthropic_item:
pinned.append(anthropic_item)
try:
x_item = x_status_item(get_json(self.x_url), now=now)
except (OSError, ValueError, json.JSONDecodeError) as exc:
errors.append(f"isUpMap: {type(exc).__name__}: {exc}"[:160])
else:
successful_sources += 1
if x_item:
pinned.append(x_item)
if x_item["status"] in STATUS_RANK:
alerts.append(x_item)
if successful_sources == 0:
raise OSError("; ".join(errors))
google_ok = False
try:
request = urllib.request.Request(
self.google_url,
headers={"User-Agent": "busy-codex-ai-status/1"},
)
with self.opener.open(request, timeout=8) as response:
google_ok = 200 <= response.status < 400
except OSError:
pass
with self._lock:
if google_ok:
self.google_failures = 0
self.google_last_ok = time.time() if now is None else now
else:
self.google_failures += 1
google_failures = self.google_failures
google_item = google_status_item(google_failures)
self.pinned = [*pinned, google_item]
if google_item["status"] in STATUS_RANK:
alerts.append(google_item)
return (sorted(alerts,
key=lambda a: (-STATUS_RANK[a["status"]], a["order"])),
errors)
def status(self) -> dict:
with self._lock:
return {
"enabled": True,
"source": self.url,
"x_source": self.x_url,
"google_probe": {
"url": self.google_url,
"consecutive_failures": self.google_failures,
"last_ok": self.google_last_ok,
},
"last_success": self.last_success,
"error": self.last_error,
"alerts": [dict(alert) for alert in self.alerts],
"pinned": [dict(item) for item in self.pinned],
"controls": {
"connected": self.input_connected,
"error": self.input_error,
"last_action": self.last_input,
"manual_until": self.manual_until or None,
"selected_index": self.selected_index,
},
"view": ({
"provider": self.displayed_provider,
"index": self.displayed_index,
"shown_at": self.displayed_at,
} if self.displayed_index is not None else None),
"draw": {
"attempts": self.draw_attempts,
"rejections": self.draw_rejections,
"last_ms": self.last_draw_ms,
"last_http_status": self.last_draw_status,
},
"rendering": self.drawn,
}
def _draw(self, payload: dict) -> bool:
started = time.monotonic()
ok = self.transport.draw(payload)
elapsed_ms = (time.monotonic() - started) * 1000
with self._lock:
self.draw_attempts += 1
if not ok:
self.draw_rejections += 1
self.last_draw_ms = round(elapsed_ms, 1)
self.last_draw_status = getattr(self.transport, "last_http_status", None)
return ok
def _note_drawn(self, alert: dict, index: int, now: float):
with self._lock:
self.displayed_index = index
self.displayed_provider = alert["provider"]
self.displayed_at = now
def _clear(self):
if self.drawn:
with self.render_lock:
cleared = self.transport.clear(APP_NAME)
if cleared:
self.drawn = False
with self._lock:
self.displayed_index = None
self.displayed_provider = ""
self.displayed_at = None
def run(self, stop: threading.Event):
if self.input_url:
threading.Thread(target=self._input_loop, args=(stop,),
daemon=True).start()
next_poll = 0.0
last_view = None
last_text_ts = 0.0
last_anim = None
last_anim_ts = 0.0
item_signature = None
auto_index = 0
next_rotate_at = None
was_manual = False
error_logged = ""
try:
while not stop.is_set():
now = time.time()
if now >= next_poll or self.refresh_requested.is_set():
self.refresh_requested.clear()
try:
alerts, source_errors = self.fetch(now=now)
except (OSError, ValueError, json.JSONDecodeError) as exc:
error = f"{type(exc).__name__}: {exc}"[:160]
with self._lock:
self.last_error = error
if error != error_logged:
self.logger(f"AI status source unavailable: {error}")
error_logged = error
else:
with self._lock:
previous = {(a["provider"], a["status"])
for a in self.alerts}
self.alerts = alerts
self.last_success = now
self.last_error = "; ".join(source_errors)
current = {(a["provider"], a["status"]) for a in alerts}
if current != previous:
summary = ", ".join(
f'{a["provider"]} {a["status"]}' for a in alerts
) or "all providers operational"
self.logger(f"AI status: {summary}")
partial_error = "; ".join(source_errors)
if partial_error and partial_error != error_logged:
self.logger(f"AI status source unavailable: {partial_error}")
error_logged = partial_error
elif not partial_error and error_logged:
self.logger("AI status source reachable again")
error_logged = ""
next_poll = now + self.poll_s
with self._lock:
if (self.last_success is not None
and now - self.last_success > SOURCE_MAX_AGE_S):
self.alerts = []
self.pinned = []
alerts = [dict(alert) for alert in self.alerts]
pinned = [dict(item) for item in self.pinned]
selected_index = self.selected_index
manual_until = self.manual_until
manual_index = (selected_index
if now < manual_until else None)
force_redraw = self.redraw_requested.is_set()
if force_redraw:
self.redraw_requested.clear()
allowed = self.should_render()
if not alerts or not allowed:
self._clear()
last_view = last_anim = None
item_signature = None
auto_index = 0
next_rotate_at = None
was_manual = False
else:
items = carousel_items(alerts, pinned)
signature = tuple(
(item["provider"], item["status"],
tuple(item.get("surfaces") or []))
for item in items
)
if signature != item_signature:
item_signature = signature
auto_index = 0
next_rotate_at = None
last_view = None
if manual_index is not None:
index = manual_index % len(items)
was_manual = True
else:
if was_manual:
auto_index = ((selected_index or 0) % len(items))
next_rotate_at = now + ROTATE_S
was_manual = False
elif next_rotate_at is not None and now >= next_rotate_at:
auto_index = (auto_index + 1) % len(items)
# Do not start the next dwell until this item has
# actually been accepted by the device.
next_rotate_at = None
index = auto_index
alert = items[index]
view = (alert["provider"], alert["status"],
tuple(alert["surfaces"]), index, len(items))
anim_path = animation_path(alert["status"])
with self.render_lock:
if stop.is_set():
break # shutdown may have cleared the canvas while we waited
if (anim_path != last_anim
or now - last_anim_ts >= ANIM_REFRESH_S):
ok = self._draw({
"application_name": APP_NAME,
"priority": DRAW_PRIORITY,
"elements": [
*alert_animation(alert["status"]),
*alert_elements(alert, index, len(items)),
],
})
if ok:
rendered_at = time.time()
last_anim, last_anim_ts = anim_path, rendered_at
last_view, last_text_ts = view, rendered_at
self.drawn = True
self._note_drawn(alert, index, rendered_at)
if manual_index is None and next_rotate_at is None:
next_rotate_at = rendered_at + ROTATE_S
elif (force_redraw or view != last_view
or now - last_text_ts >= KEEPALIVE_S):
ok = self._draw({
"application_name": APP_NAME,
"priority": DRAW_PRIORITY,
"elements": alert_elements(alert, index, len(items)),
})
if ok:
rendered_at = time.time()
last_view, last_text_ts = view, rendered_at
self.drawn = True
self._note_drawn(alert, index, rendered_at)
if manual_index is None and next_rotate_at is None:
next_rotate_at = rendered_at + ROTATE_S
# A physical control wakes the renderer immediately; otherwise
# it ticks at the normal low-traffic cadence.
wake_at = next_poll
if next_rotate_at is not None:
wake_at = min(wake_at, next_rotate_at)
if manual_index is not None:
wake_at = min(wake_at, manual_until)
self.redraw_requested.wait(
timeout=min(0.5, max(0.05, wake_at - time.time())),
)
finally:
self._clear()app.py
#!/usr/bin/env python3
"""Busy Codex: task status, account quotas and tactile reasoning effort.
python3 app.py # USB BUSY Bar, local Codex
python3 app.py --host 192.168.1.50 # Wi-Fi; BUSYBAR_TOKEN if required
python3 app.py --host 127.0.0.1:8080 --demo # emulator, no Codex or secrets
Download the complete app folder, not this entrypoint alone. Python 3.9+;
standard library only. macOS foreground targeting follows Codex Desktop or
the active CLI terminal. CLI controls require the native-control fork:
https://github.com/wowlocal/codex/tree/codex/native-tui-control
Fast needs a build with fast/set; restart CLI sessions after updating.
Stock CLI remains usable for account limits, but cannot accept control changes.
Reads local Codex session metadata and invokes the installed `codex app-server`
only to read account limits. It does not send prompts, edit Codex config,
install background services, update executables or restart exited processes.
SIGINT/SIGTERM stops the owned processes and clears only this app's canvas.
"""
from __future__ import annotations
import argparse
import json
import math
import os
from pathlib import Path
import signal
import socket
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
APP = "busy-codex"
HERE = Path(__file__).resolve().parent
STOP = threading.Event()
def arguments(argv=None):
parser = argparse.ArgumentParser(description=__doc__,
formatter_class=argparse.RawDescriptionHelpFormatter)
parser.add_argument("--host", default="10.0.4.20", metavar="IP[:PORT]")
parser.add_argument("--port", type=int, default=18765,
help="local report listener, separate from installed busy-codex (18765)")
parser.add_argument("--demo", action="store_true",
help="cycle real effort animations without opening Codex data")
parser.add_argument("--test", action="store_true",
help="show a synthetic effort frame and exit; no Codex needed")
parser.add_argument("--no-effort", action="store_true", help="read-only status and limits")
parser.add_argument("--no-upload", action="store_true",
help="reuse this version's assets already uploaded to the Bar")
parser.add_argument("--seconds", type=float, default=0,
help="stop automatically after this many seconds (0: until stopped)")
args = parser.parse_args(argv)
if not 1 <= args.port <= 65535:
parser.error("--port must be between 1 and 65535")
if args.seconds < 0 or not math.isfinite(args.seconds):
parser.error("--seconds must be a finite, nonnegative number")
host = args.host.removeprefix("http://").rstrip("/")
parsed = urllib.parse.urlsplit("http://" + host)
try:
valid = bool(parsed.hostname) and parsed.port != 0
except ValueError:
valid = False
if not valid or parsed.path or parsed.query or parsed.fragment or parsed.username:
parser.error("--host must be an IP or hostname with an optional port")
args.host = host
return args
class Bar:
"""Small preview transport; normal mode uses the daemon's USB-aware transport."""
def __init__(self, host):
from busybar_http import local_opener
self.base = "http://" + host + "/api"
source = os.environ.get("BUSYBAR_USB_SOURCE_IP", "10.0.4.21") if host == "10.0.4.20" else None
self.opener = local_opener(source)
self.headers = {}
token = os.environ.get("BUSYBAR_TOKEN", "")
if token:
self.headers["X-API-Token"] = token
def request(self, method, path, body=None, content_type="application/json"):
headers = dict(self.headers)
if body is not None:
headers["Content-Type"] = content_type
req = urllib.request.Request(self.base + path, data=body, method=method, headers=headers)
try:
with self.opener.open(req, timeout=2) as response:
response.read()
return response.status
except urllib.error.HTTPError as exc:
if exc.code in (401, 403):
raise RuntimeError("BUSY Bar refused authentication; check BUSYBAR_TOKEN") from exc
return exc.code
def clear(self):
try:
self.request("DELETE", "/display/draw?application_name=" + APP)
except OSError:
pass
def demo(args):
"""Emit the same BGRA animation frames used to generate native .anim assets.
PNG transport makes those frames reproducible with the gallery recorder,
whose animation decoder does not support our native inter-frame encoding.
The clock selects a current frame after every send; delayed frames are
skipped rather than collected in a queue.
"""
import effort_animation
import fast_animation
from pixel_ui import encode_png
bar = Bar(args.host)
levels = ("high", "ultra", "fast", "normal")
started = time.monotonic()
deadline = started + (args.seconds or float("inf"))
retry_delay = .25
index = 0
print(f"{APP} demo -> {bar.base}; synthetic data only", flush=True)
try:
while not STOP.is_set() and time.monotonic() < deadline:
elapsed = time.monotonic() - started
level = "ultra" if args.test else levels[int(elapsed / effort_animation.DURATION_S) % len(levels)]
phase = elapsed % effort_animation.DURATION_S
frame_index = 18 if args.test else int(phase * effort_animation.FPS)
pixels = (fast_animation.frame(level == "fast", frame_index, entering=False)
if level in ("fast", "normal") else effort_animation.frame(level, frame_index, entering=False))
name = f"demo-{index % 4}.png"
index += 1
try:
status = bar.request("POST", f"/assets/upload?application_name={APP}&file={name}",
encode_png(pixels, 72, 16), "application/octet-stream")
if status == 200:
payload = {"application_name": APP, "priority": 30,
"elements": [{"id": "effort", "type": "image", "path": name,
"x": 0, "y": 0, "timeout": 5}]}
status = bar.request("POST", "/display/draw", json.dumps(payload).encode())
if status == 409:
STOP.wait(.5)
continue
if status == 508: # firmware still holds one of the rotating assets
STOP.wait(.1)
continue
if status != 200:
raise RuntimeError(f"BUSY Bar returned HTTP {status}")
retry_delay = .25
if args.test:
STOP.wait(.25)
break
except OSError as exc:
if args.test:
raise RuntimeError(f"BUSY Bar unreachable: {type(exc).__name__}") from exc
print(f"BUSY Bar unreachable: {type(exc).__name__}; retrying", file=sys.stderr)
STOP.wait(retry_delay)
retry_delay = min(5., retry_delay * 2)
continue
STOP.wait(1 / 12)
finally:
bar.clear()
return 0
def runtime_env(args):
env = dict(os.environ)
# This focused launcher owns its report listener and renderer. Settings for
# an independently installed multi-provider daemon must not leak into it.
for key in ("BUSYBAR_HUB", "BUSYBAR_STANDBY", "BUSYBAR_HUB_TOKEN",
"BUSYBAR_CODEX_THREAD_ID"):
env.pop(key, None)
env.update({"BUSYBAR_APP_NAME": APP, "BUSYBAR_DRAW_PRIORITY": "30", "BUSYBAR_MANAGED": "1",
"BUSYBAR_PORT": str(args.port), "BUSYBAR_LISTEN": "127.0.0.1",
"BUSYBAR_RENDER_MODE": "auto", "BUSYBAR_STYLE": "minimal",
"BUSYBAR_AI_STATUS": "0", "BUSYBAR_X_PULSE": "0",
"BUSYBAR_CODEX_EFFORT": "0" if args.no_effort else "1",
"BUSYBAR_ASTRA_STATE": str(HERE / "disabled-astra-state.json"),
"BUSYBAR_TRANSPORT": "usb" if args.host == "10.0.4.20" else "wifi",
"BUSYBAR_DEVICE": args.host, "PYTHONUNBUFFERED": "1"})
return env
def upload_assets(transport):
assets = HERE / "assets"
if not assets.is_dir():
raise RuntimeError("Packaged assets are missing. Run scripts/build_gallery.py, then its app.py.")
files = sorted(assets.glob("*.anim"))
if not files:
raise RuntimeError("Packaged animation assets are empty")
print(f"Uploading {len(files)} animations into {APP}'s assets (no app installation)", flush=True)
transport.clear(APP)
for path in files:
if STOP.is_set():
return
req = urllib.request.Request(
transport.base + f"/assets/upload?application_name={APP}&file=" + urllib.parse.quote(path.name),
data=path.read_bytes(), method="POST",
headers={**transport.headers, "Content-Type": "application/octet-stream"})
with transport.opener.open(req, timeout=8) as response:
response.read()
def stop_children(children):
for child in children:
if child.poll() is None:
child.terminate()
deadline = time.monotonic() + 3.5
for child in children:
try:
child.wait(timeout=max(.05, deadline - time.monotonic()))
except subprocess.TimeoutExpired:
child.kill()
child.wait(timeout=1)
def wait_ready(child, port):
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
deadline = time.monotonic() + 4
while not STOP.is_set() and time.monotonic() < deadline:
if child.poll() is not None:
raise RuntimeError("Display worker exited before becoming ready")
try:
with opener.open(f"http://127.0.0.1:{port}/health", timeout=.2) as response:
response.read()
if response.status == 200:
return
except OSError:
pass
STOP.wait(.05)
if not STOP.is_set():
raise RuntimeError("Display worker did not become ready")
def run(args):
# Fail before drawing or starting anything when a prior instance owns the
# local endpoint. It would be unsafe to feed an unrelated daemon's port.
with socket.socket() as probe:
try:
probe.bind(("127.0.0.1", args.port))
except OSError as exc:
raise RuntimeError(f"Local port {args.port} is in use; choose --port or stop that instance") from exc
env = runtime_env(args)
os.environ.update(env)
for key in ("BUSYBAR_HUB", "BUSYBAR_STANDBY", "BUSYBAR_HUB_TOKEN", "BUSYBAR_CODEX_THREAD_ID"):
if key not in env:
os.environ.pop(key, None)
import daemon
transport = daemon.make_transport()
if not args.no_upload:
upload_assets(transport)
children = []
deadline = time.monotonic() + (args.seconds or float("inf"))
try:
if STOP.is_set():
return 0
children.append(subprocess.Popen(
[sys.executable, str(HERE / "daemon.py"), "--port", str(args.port)], cwd=HERE, env=env))
wait_ready(children[0], args.port)
if STOP.is_set():
return 0
children.append(subprocess.Popen(
[sys.executable, str(HERE / "adapters" / "codex_status.py")], cwd=HERE, env=env))
print(f"{APP} -> {args.host}; Ctrl-C stops both workers", flush=True)
while not STOP.wait(.1) and time.monotonic() < deadline:
if any(child.poll() is not None for child in children):
raise RuntimeError("A worker exited; stopping the app (no automatic restart)")
finally:
stop_children(children)
transport.clear(APP)
return 0
def main(argv=None):
args = arguments(argv)
STOP.clear()
signal.signal(signal.SIGINT, lambda *_: STOP.set())
signal.signal(signal.SIGTERM, lambda *_: STOP.set())
try:
return demo(args) if args.demo or args.test else run(args)
except (OSError, RuntimeError) as exc:
print(f"{APP}: {exc}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())busybar_http.py
"""Device HTTP transport shared by the daemon, gallery runner and installers.
No Codex state or application configuration is loaded here. Interactive writes
are attempted once; the renderer retries the latest state on its next wake-up.
"""
from __future__ import annotations
import http.client
import json
import urllib.error
import urllib.parse
import urllib.request
def local_opener(source_address: str | None = None):
"""Bypass system proxies and optionally bind the Bar's USB interface."""
handlers = [urllib.request.ProxyHandler({})]
if source_address:
handlers.append(SourceAddressHTTPHandler(source_address))
return urllib.request.build_opener(*handlers)
class HttpTransport:
"""Busy Bar HTTP API over any of its three routes. Remembers whether
the device answers (device_ok, reported by GET /hub so a standby can
step in when the hub cannot draw) and logs only on transitions."""
TIMEOUT_S = 2.0
def __init__(self, base: str, headers: dict | None = None, opener=None, logger=None):
self.base = base
self.headers = headers or {}
self.opener = opener or local_opener() # cloud: proxy-aware; usb/wifi: never
self.logger = logger or (lambda _message: None)
self.device_ok: bool | None = None # None until the first draw/clear
self.last_error = ""
self.last_http_status: int | None = None
def _note(self, ok: bool, err: str = ""):
if ok:
if self.device_ok is False:
self.logger(f"device {self.base}: reachable again")
self.device_ok, self.last_error = True, ""
else:
if err != self.last_error:
self.logger(f"device {self.base}: {err}")
self.device_ok, self.last_error = False, err
def _request(self, method: str, path: str, body: bytes | None = None) -> bool:
headers = dict(self.headers)
if body:
headers["Content-Type"] = "application/json"
req = urllib.request.Request(self.base + path, data=body, method=method,
headers=headers)
try:
with self.opener.open(req, timeout=self.TIMEOUT_S) as response:
self.last_http_status = response.status
self._note(True)
return True
except urllib.error.HTTPError as e:
self.last_http_status = e.code
if e.code == 409: # reachable; an active focus session owns the screen
self._note(True)
else:
self._note(False, f"HTTP {e.code} on {method} {path}")
return False
except OSError as e:
self.last_http_status = None
self._note(False, f"{type(e).__name__}: {e}"[:120])
return False # unplugged / offline; retried on the normal cadence
def draw(self, payload: dict) -> bool:
return self._request("POST", "/display/draw", json.dumps(payload).encode())
def clear(self, app_name: str) -> bool:
return self._request(
"DELETE", "/display/draw?application_name=" + urllib.parse.quote(app_name)
)
def get_json(self, path: str) -> dict | None:
req = urllib.request.Request(self.base + path, headers=self.headers)
try:
with self.opener.open(req, timeout=self.TIMEOUT_S) as r:
return json.loads(r.read())
except (OSError, json.JSONDecodeError, urllib.error.HTTPError):
return None
def reachable(self) -> bool:
"""Cheap liveness check that never touches the canvas."""
return self.get_json("/version") is not None
class SourceAddressHTTPHandler(urllib.request.HTTPHandler):
"""Keep USB device traffic off VPNs that also advertise 10.0.4.0/24."""
def __init__(self, source_address: str):
super().__init__()
self.source_address = source_address
def http_open(self, request):
def connection(host, **kwargs):
return http.client.HTTPConnection(
host, source_address=(self.source_address, 0), **kwargs,
)
return self.do_open(connection, request)busybar_input.py
"""Small, dependency-free BUSY Bar input client shared by dashboard consumers.
The device sends BSB_State.State protobuf messages over its local WebSocket.
Input is delivered in order; unlike visual frames, dial steps must never be
coalesced or discarded. Rendering and event ownership stay with each caller.
"""
from __future__ import annotations
import base64
import hashlib
import os
import socket
import ssl
import struct
import threading
import time
import urllib.parse
from collections.abc import Callable
WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11"
MAX_MESSAGE_BYTES = 4 * 1024 * 1024
CONNECT_TIMEOUT_S = 3.0
READ_TIMEOUT_S = 1.0
PING_INTERVAL_S = 20.0
PONG_TIMEOUT_S = 10.0
RECONNECT_MIN_S = 2.0
RECONNECT_MAX_S = 30.0
def input_stream_url(api_base: str, token: str = "") -> str:
"""Turn a local device HTTP API base into its status WebSocket URL."""
parsed = urllib.parse.urlsplit(api_base)
scheme = "wss" if parsed.scheme == "https" else "ws"
path = parsed.path.rstrip("/") + "/status/ws"
query = ("x-api-token=" + urllib.parse.quote(token, safe="")) if token else ""
return urllib.parse.urlunsplit((scheme, parsed.netloc, path, query, ""))
def _read_varint(data: bytes, offset: int) -> tuple[int, int]:
value = 0
shift = 0
while offset < len(data) and shift < 70:
byte = data[offset]
offset += 1
if shift == 63 and byte > 1:
raise ValueError("protobuf varint exceeds 64 bits")
value |= (byte & 0x7F) << shift
if not byte & 0x80:
return value, offset
shift += 7
raise ValueError("invalid protobuf varint")
def _protobuf_fields(data: bytes):
"""Yield the protobuf wire fields needed by the BUSY input messages."""
offset = 0
while offset < len(data):
tag, offset = _read_varint(data, offset)
number, wire = tag >> 3, tag & 7
if not number:
raise ValueError("invalid protobuf field number")
if wire == 0:
value, offset = _read_varint(data, offset)
elif wire == 1:
if offset + 8 > len(data):
raise ValueError("truncated protobuf fixed64")
value, offset = data[offset:offset + 8], offset + 8
elif wire == 2:
length, offset = _read_varint(data, offset)
if offset + length > len(data):
raise ValueError("truncated protobuf message")
value, offset = data[offset:offset + length], offset + length
elif wire == 5:
if offset + 4 > len(data):
raise ValueError("truncated protobuf fixed32")
value, offset = data[offset:offset + 4], offset + 4
else:
raise ValueError(f"unsupported protobuf wire type {wire}")
yield number, wire, value
def parse_input_events(state: bytes) -> list[tuple]:
"""Decode button/encoder events from one BSB_State.State message.
The tiny wire decoder keeps the daemon dependency-free; field numbers come
from busy-app/busybar-protobuf's state.proto and input.proto.
"""
events = []
for number, wire, update in _protobuf_fields(state):
if number != 2 or wire != 2: # State.updates
continue
for update_number, update_wire, input_event in _protobuf_fields(update):
if update_number != 11 or update_wire != 2: # StateUpdate.input
continue
for event_number, event_wire, event in _protobuf_fields(input_event):
if event_wire != 2:
continue
if event_number == 1: # InputEvent.button_event
button = action = 0 # proto3 enum defaults: OK + PRESS
for field, field_wire, value in _protobuf_fields(event):
if field_wire == 0 and field == 1:
button = value
elif field_wire == 0 and field == 2:
action = value
events.append(("button", button, action))
elif event_number == 2: # InputEvent.switch_event
position = 0 # proto3 enum default: BUSY
for field, field_wire, value in _protobuf_fields(event):
if field == 1 and field_wire == 0:
position = value
events.append(("switch", position))
elif event_number == 3: # InputEvent.encoder_event
delta = 0
for field, field_wire, value in _protobuf_fields(event):
if field == 1 and field_wire == 0:
delta = (value >> 1) ^ -(value & 1) # sint32 zigzag
if delta:
events.append(("encoder", delta))
return events
def send_frame(sock: socket.socket, opcode: int, payload: bytes = b"") -> None:
"""Send a masked client frame; control frames cannot be fragmented."""
if opcode not in (0, 1, 2, 8, 9, 10):
raise ValueError("invalid WebSocket opcode")
length = len(payload)
if length > MAX_MESSAGE_BYTES or (opcode >= 8 and length > 125):
raise ValueError("WebSocket payload too large")
header = bytearray((0x80 | opcode,))
if length < 126:
header.append(0x80 | length)
elif length <= 0xFFFF:
header.append(0x80 | 126)
header.extend(struct.pack("!H", length))
else:
header.append(0x80 | 127)
header.extend(struct.pack("!Q", length))
mask = os.urandom(4)
header.extend(mask)
masked = bytes(byte ^ mask[index % 4] for index, byte in enumerate(payload))
sock.sendall(bytes(header) + masked)
class FrameReader:
"""Keep partial bytes across timeouts, including partial frame headers.
A timeout is a chance to check shutdown/keepalive, not a message boundary.
Reading each part with a fresh recv_exact would lose those bytes and turn
the rest of that frame into a corrupt header on the next call.
"""
def __init__(self, sock: socket.socket, max_bytes: int = MAX_MESSAGE_BYTES):
self.sock = sock
self.max_bytes = max_bytes
self.buffer = bytearray()
def _fill(self, length: int) -> None:
while len(self.buffer) < length:
chunk = self.sock.recv(min(65536, length - len(self.buffer)))
if not chunk:
raise ConnectionError("WebSocket closed")
self.buffer.extend(chunk)
def read(self) -> tuple[int, bool, bytes]:
self._fill(2)
first, second = self.buffer[:2]
opcode, final = first & 0x0F, bool(first & 0x80)
if first & 0x70 or opcode not in (0, 1, 2, 8, 9, 10):
raise ConnectionError("invalid WebSocket frame flags")
if second & 0x80:
raise ConnectionError("server WebSocket frame must not be masked")
length = second & 0x7F
offset = 2
if length == 126:
self._fill(4)
length = struct.unpack("!H", self.buffer[2:4])[0]
offset = 4
if length < 126:
raise ConnectionError("non-minimal WebSocket length")
elif length == 127:
self._fill(10)
length = struct.unpack("!Q", self.buffer[2:10])[0]
offset = 10
if length < 65536 or length >= 1 << 63:
raise ConnectionError("invalid WebSocket length")
if length > self.max_bytes:
raise ConnectionError(f"WebSocket frame exceeds {self.max_bytes} bytes")
if opcode >= 8 and (not final or length > 125):
raise ConnectionError("invalid WebSocket control frame")
self._fill(offset + length)
payload = bytes(self.buffer[offset:offset + length])
del self.buffer[:offset + length]
return opcode, final, payload
def connect_websocket(url: str, source_address: str | None = None) -> socket.socket:
"""Connect directly to the device, optionally binding its USB interface."""
parsed = urllib.parse.urlsplit(url)
if parsed.scheme not in ("ws", "wss") or not parsed.hostname:
raise ValueError("expected an absolute ws:// or wss:// device URL")
host = parsed.hostname
port = parsed.port or (443 if parsed.scheme == "wss" else 80)
source = (source_address, 0) if source_address else None
sock = socket.create_connection((host, port), timeout=CONNECT_TIMEOUT_S,
source_address=source)
try:
if parsed.scheme == "wss":
sock = ssl.create_default_context().wrap_socket(sock, server_hostname=host)
key = base64.b64encode(os.urandom(16)).decode()
target = urllib.parse.urlunsplit(("", "", parsed.path or "/", parsed.query, ""))
lines = [
f"GET {target} HTTP/1.1", f"Host: {parsed.netloc}",
"Upgrade: websocket", "Connection: Upgrade",
f"Sec-WebSocket-Key: {key}", "Sec-WebSocket-Version: 13",
]
sock.sendall(("\r\n".join(lines) + "\r\n\r\n").encode())
response = bytearray()
# Do not consume bytes from the first data frame after the HTTP headers.
while not response.endswith(b"\r\n\r\n"):
byte = sock.recv(1)
if not byte:
raise ConnectionError("WebSocket closed during handshake")
response.extend(byte)
if len(response) > 16384:
raise ConnectionError("oversized WebSocket handshake")
head = bytes(response).decode("iso-8859-1")
status = head.split("\r\n", 1)[0].split(" ")
if len(status) < 2 or status[:2] != ["HTTP/1.1", "101"]:
raise ConnectionError(head.split("\r\n", 1)[0])
headers = {}
for line in head.split("\r\n")[1:]:
if ":" in line:
name, value = line.split(":", 1)
headers[name.strip().lower()] = value.strip()
expected = base64.b64encode(hashlib.sha1((key + WS_GUID).encode()).digest()).decode()
if headers.get("sec-websocket-accept") != expected:
raise ConnectionError("invalid WebSocket accept header")
if (headers.get("upgrade", "").lower() != "websocket"
or "upgrade" not in {part.strip().lower() for part in
headers.get("connection", "").split(",")}):
raise ConnectionError("invalid WebSocket upgrade headers")
sock.settimeout(READ_TIMEOUT_S)
return sock
except Exception:
sock.close()
raise
class InputStream:
"""Reconnectable local device input stream with ordered callback delivery.
The caller supplies routing and connection-state callbacks. No worker or
event queue is created here: handlers should be quick and signal their own
workers when an action needs network I/O or rendering.
"""
def __init__(self, url: str, on_event: Callable[[tuple], object], *,
source_address: str | None = None,
on_state: Callable[[bool, str], object] = lambda _ok, _err: None,
logger: Callable[[str], object] = lambda _msg: None,
label: str = "BUSY controls", connect=None, clock=time.monotonic):
self.url = url
self.on_event = on_event
self.source_address = source_address
self.on_state = on_state
self.logger = logger
self.label = label
self.connect = connect or connect_websocket
self.clock = clock
def _consume(self, sock: socket.socket, stop: threading.Event) -> None:
reader = FrameReader(sock)
fragments = bytearray()
fragment_opcode = None
next_ping = self.clock() + PING_INTERVAL_S
pending_ping = None
pong_deadline = 0.0
while not stop.is_set():
now = self.clock()
if pending_ping is not None and now >= pong_deadline:
raise ConnectionError("WebSocket pong timed out")
if pending_ping is None and now >= next_ping:
pending_ping = os.urandom(4)
send_frame(sock, 9, pending_ping)
pong_deadline = now + PONG_TIMEOUT_S
next_ping = now + PING_INTERVAL_S
try:
opcode, final, payload = reader.read()
except socket.timeout:
continue
if opcode == 8:
raise ConnectionError("WebSocket closed")
if opcode == 9:
send_frame(sock, 10, payload)
continue
if opcode == 10:
if payload == pending_ping:
pending_ping = None
continue
if opcode in (1, 2):
if fragment_opcode is not None:
raise ConnectionError("new WebSocket message during fragments")
fragment_opcode = opcode
elif fragment_opcode is None:
raise ConnectionError("unexpected WebSocket continuation")
if len(fragments) + len(payload) > MAX_MESSAGE_BYTES:
raise ConnectionError(f"WebSocket message exceeds {MAX_MESSAGE_BYTES} bytes")
fragments.extend(payload)
if not final:
continue
if fragment_opcode == 2:
for event in parse_input_events(bytes(fragments)):
try:
self.on_event(event)
except Exception as exc:
# One failing route must not discard subsequent steps.
self.logger(f"{self.label} handler failed: {type(exc).__name__}: {exc}"[:240])
fragments.clear()
fragment_opcode = None
def run(self, stop: threading.Event) -> None:
delay = RECONNECT_MIN_S
logged_error = ""
while not stop.is_set():
sock = None
connected_at = None
try:
sock = self.connect(self.url, source_address=self.source_address)
send_frame(sock, 1, b'{"enable":true}')
connected_at = self.clock()
self.on_state(True, "")
if logged_error:
self.logger(f"{self.label} connected again")
logged_error = ""
self._consume(sock, stop)
except (OSError, ValueError) as exc:
error = f"{type(exc).__name__}: {exc}"[:160]
self.on_state(False, error)
if error != logged_error and not stop.is_set():
self.logger(f"{self.label} unavailable: {error}")
logged_error = error
finally:
if sock is not None:
try:
sock.close()
except OSError:
pass
if stop.is_set():
break
# A stable connection resets backoff; repeated handshake/close loops
# continue to back off rather than hammering an unavailable device.
if connected_at is not None and self.clock() - connected_at >= RECONNECT_MAX_S:
delay = RECONNECT_MIN_S
stop.wait(delay)
delay = min(delay * 2, RECONNECT_MAX_S)
self.on_state(False, "")codex_cli_client.py
"""Small settings-only client for a CLI launched through codex_cli.py."""
import json
import os
from pathlib import Path
import socket
import stat
import time
from codex_effort import CatalogError
class CLIIPC:
def __init__(self, path, on_change):
self.path = Path(path)
self.on_change = on_change
self.sock = None
self.stream = None
self.thread_id = None
self.owner = 'cli'
self.state = None
self.revision = 0
self.poll_after = 0
def connect(self, thread_id):
info = self.path.stat()
if info.st_uid != os.getuid() or not stat.S_ISSOCK(info.st_mode):
raise ValueError('CLI socket is not owned by this user')
self.thread_id = thread_id
self.sock = socket.socket(socket.AF_UNIX)
self.sock.settimeout(6)
self.sock.connect(str(self.path))
self.stream = self.sock.makefile('rb')
self.receive()
def close(self):
if self.stream:
self.stream.close()
if self.sock:
self.sock.close()
self.stream = self.sock = None
def rpc(self, value):
self.sock.sendall(json.dumps(value).encode() + b'\n')
line = self.stream.readline(16385)
if not line or len(line) > 16384:
raise ConnectionError('Invalid CLI control response')
result = json.loads(line)
if 'error' in result:
raise ValueError(result['error'])
state = result['result']
if state.get('thread_id') != self.thread_id or not state.get('ready'):
raise ValueError('CLI selected task changed')
if self.state != state:
self.state = state
self.revision += 1
self.on_change({'type': 'snapshot', 'revision': self.revision,
'conversationState': {'latestThreadSettings': {
'model': state['model'], 'effort': state['effort']}}})
return result
def receive(self):
self.poll_after = time.monotonic() + .2
return self.rpc({'method': 'status'})
def poll(self):
if time.monotonic() >= self.poll_after:
self.receive()
def levels_for(self, model):
levels = (self.state or {}).get('supported_efforts')
if (self.state or {}).get('model') != model or not levels:
raise CatalogError('CLI has not advertised effort levels for its current model')
return levels
def request(self, method, params, target=None):
if method != 'thread-follower-update-thread-settings':
raise ValueError('Unsupported CLI control operation')
thread_id, model = self.thread_id, self.state['model']
effort = params['threadSettings']['effort']
try:
return self.rpc({'method': 'set_effort', 'effort': effort,
'expected_thread_id': thread_id, 'expected_model': model,
'expected_effort': self.state['effort']})
except (ValueError, OSError) as error:
# An accepted operation can outlive the request/ack timeout. Read
# the live bridge again before reporting failure; never resend a
# dial step or infer success from the value we attempted to write.
uncertain = (isinstance(error, TimeoutError) or str(error) in (
'CLI settings request timed out', 'CLI did not confirm effort change'))
if not uncertain:
raise
self.close()
try:
self.connect(thread_id)
if self.state['model'] == model and self.state['effort'] == effort:
return {'result': self.state}
except (OSError, ValueError, KeyError):
pass
raise errorcodex_cli_native.py
"""Client for the fork's native, settings-only TUI control endpoint."""
import time
import uuid
from codex_cli_client import CLIIPC
def normalize(state):
if not isinstance(state, dict) or state.get('protocolVersion') != 1:
raise ValueError('Unsupported native CLI control version')
return {**state, 'native_control': True, 'thread_id': state.get('threadId'),
'supported_efforts': state.get('supportedEfforts'),
'context_pct': state.get('contextPct')}
class NativeCLIIPC(CLIIPC):
def raw_rpc(self, value):
import json
self.sock.sendall(json.dumps(value).encode() + b'\n')
line = self.stream.readline(16385)
if not line or len(line) > 16384:
raise ConnectionError('Invalid native CLI response')
result = json.loads(line)['result']
if 'error' in result:
raise ValueError(result['error'])
return result
def receive(self):
self.poll_after = time.monotonic() + .2
state = normalize(self.raw_rpc({'method': 'status/read'}))
if state['thread_id'] != self.thread_id or not state.get('ready'):
raise ValueError('Native CLI selected task changed or is not ready')
if self.state != state:
self.state = state
self.revision += 1
self.on_change({'type': 'snapshot', 'revision': self.revision,
'conversationState': {'latestThreadSettings': {
'model': state['model'], 'effort': state['effort'],
**({key: state.get(key) for key in ('serviceTier', 'fastServiceTier')}
if 'serviceTier' in state else {})}}})
return {'result': state}
def request(self, method, params, target=None):
if method != 'thread-follower-update-thread-settings':
raise ValueError('Unsupported native CLI operation')
settings = params['threadSettings']
if 'serviceTier' in settings:
tier = settings['serviceTier']
fast_tier = self.state.get('fastServiceTier')
if not fast_tier or tier not in (fast_tier, 'default'):
raise ValueError('Restart the updated native CLI with Fast mode support')
command = {'method': 'fast/set', 'enabled': tier == fast_tier}
key, expected = 'serviceTier', tier
else:
key, expected = 'effort', settings['effort']
command = {'method': 'effort/set', 'effort': expected}
request_id = str(uuid.uuid4())
instance, model = self.state['instanceId'], self.state['model']
try:
result = self.raw_rpc({**command, 'requestId': request_id,
'expectedRevision': self.state['revision'],
'expectedThreadId': self.thread_id})
except OSError:
# Recover the exact request result; never repeat a settings write.
self.close()
self.connect(self.thread_id)
if self.state['instanceId'] != instance:
raise ValueError('Native CLI instance changed')
result = self.raw_rpc({'method': 'request/read', 'requestId': request_id})
deadline = time.monotonic() + 8
while result['status'] == 'pending' and time.monotonic() < deadline:
time.sleep(.025)
result = self.raw_rpc({'method': 'request/read', 'requestId': request_id})
if result['status'] != 'applied':
raise ValueError((result.get('outcome') or {}).get('error') or 'Native settings confirmation is pending')
outcome = result['outcome']
if (outcome.get('threadId'), outcome.get('model'), outcome.get(key)) != (self.thread_id, model, expected):
raise ValueError('Native CLI confirmation does not match request')
# Let the native TUI consume its confirmed event and refresh the widget.
while time.monotonic() < deadline:
self.receive()
if self.state['model'] == model and self.state.get(key) == expected:
return {'result': self.state}
time.sleep(.025)
raise ValueError('Native CLI display has not caught up to confirmed settings')codex_effort.py
"""BUSY dial -> existing Codex Desktop or connected CLI settings (no new turns).
Desktop IPC is private/versioned. Fail closed if its owner or snapshot protocol
changes. Never resume a task, edit config.toml, or send a prompt as a fallback.
"""
from __future__ import annotations
import copy
from collections import deque
import json
import os
from pathlib import Path
import select
import socket
import stat
import struct
import threading
import time
import uuid
from effort_animation import DURATION_S
import codex_fast
LEVELS = ('none', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max', 'ultra')
STATE_KEYS = {'latestThreadSettings', 'latestModel', 'latestReasoningEffort',
'latestCollaborationMode'}
MAX_FRAME = 256 * 1024 * 1024
CATALOG_GRACE_S = 300
STEP_INTERVAL_S = .04
class CatalogError(ValueError):
"""A local catalog problem does not mean the Desktop connection failed."""
class ModelCatalog:
def __init__(self, home, clock=time.monotonic):
self.path = Path(home) / 'models_cache.json'
self.clock = clock
self.known = {}
self.current = {}
self.read_error = ''
self.refresh() # Warm before the first dial event, including every model.
def refresh(self):
try:
data = json.loads(self.path.read_text())
if not isinstance(data, dict) or not isinstance(data.get('models'), list):
raise ValueError('models must be an array')
current = {}
for entry in data['models']:
if not isinstance(entry, dict) or not isinstance(entry.get('slug'), str):
continue
supported = {r.get('effort') for r in entry.get('supported_reasoning_levels', [])
if isinstance(r, dict)}
current[entry['slug']] = [level for level in LEVELS if level in supported]
self.current = current
now = self.clock()
for model, levels in current.items():
self.known[model] = (levels, now)
self.read_error = ''
except (OSError, ValueError, TypeError) as error:
self.current = {}
self.read_error = type(error).__name__
def levels_for(self, model):
self.refresh()
if model in self.current:
levels = self.current[model]
else:
levels, seen_at = self.known.get(model, (None, float('-inf')))
if levels is None or self.clock() - seen_at > CATALOG_GRACE_S:
reason = self.read_error or 'model absent'
raise CatalogError(f'Codex catalog unavailable for model={model!r}: {reason}; '
f'catalog_models={",".join(self.current)}')
if not levels:
raise CatalogError(f'No supported effort levels for model={model!r}')
return list(levels)
def supported_efforts(model, home):
return ModelCatalog(home).levels_for(model)
def model_effort(state):
settings = state.get('latestThreadSettings') or {}
collab = settings.get('collaborationMode') or state.get('latestCollaborationMode') or {}
mode_settings = collab.get('settings') or {}
return (settings.get('model') or state.get('latestModel'),
mode_settings.get('reasoning_effort') or settings.get('effort')
or state.get('latestReasoningEffort'))
def effort_settings(state, effort):
result = {'effort': effort}
collab = ((state.get('latestThreadSettings') or {}).get('collaborationMode')
or state.get('latestCollaborationMode'))
if collab:
result['collaborationMode'] = copy.deepcopy(collab)
result['collaborationMode']['settings']['reasoning_effort'] = effort
return result
def apply_change(state, revision, change):
if change['type'] == 'snapshot':
return ({k: copy.deepcopy(v) for k, v in change['conversationState'].items()
if k in STATE_KEYS}, change['revision'])
if change['type'] != 'patches' or change['baseRevision'] != revision:
raise ValueError('Codex snapshot revision mismatch')
state = copy.deepcopy(state)
for patch in change['patches']:
path = patch['path']
if not isinstance(path, list):
raise ValueError('Unsupported Codex patch format')
if not path or path[0] not in STATE_KEYS:
continue
parent = state
for key in path[:-1]:
parent = parent[key]
if patch['op'] == 'remove':
parent.pop(path[-1], None)
elif patch['op'] in ('add', 'replace'):
parent[path[-1]] = copy.deepcopy(patch['value'])
else:
raise ValueError('Unsupported Codex patch operation')
return state, change['revision']
class DesktopIPC:
def __init__(self, path, on_change):
self.path = Path(path)
self.on_change = on_change
self.sock = None
self.client_id = None
self.thread_id = None
self.owner = None
def connect(self, thread_id):
info = self.path.stat()
if not stat.S_ISSOCK(info.st_mode) or info.st_uid != os.getuid():
raise ValueError('Codex IPC socket is not owned by this user')
self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
self.sock.settimeout(3)
self.sock.connect(str(self.path))
self.client_id = self.request('initialize', {'clientType': 'busybar'})['result']['clientId']
self.thread_id = thread_id
response = self.request('thread-owner-discovery',
{'hostId': 'local', 'conversationId': thread_id})
self.owner = response['handledByClientId']
self.send({'type': 'broadcast', 'method': 'thread-stream-following-changed',
'version': 1, 'sourceClientId': self.client_id,
'targetClientIds': [self.owner],
'params': {'hostId': 'local', 'conversationId': thread_id, 'following': True}})
def close(self):
if self.sock:
self.sock.close()
self.sock = None
def send(self, message):
data = json.dumps(message).encode()
self.sock.sendall(struct.pack('<I', len(data)) + data)
def receive(self):
def exact(length):
data = bytearray()
while len(data) < length:
part = self.sock.recv(length - len(data))
if not part:
raise ConnectionError('Codex Desktop disconnected')
data.extend(part)
return data
size = struct.unpack('<I', exact(4))[0]
if not 0 < size <= MAX_FRAME:
raise ValueError('Invalid Codex IPC frame length')
message = json.loads(exact(size))
if message['type'] == 'client-discovery-request':
self.send({'type': 'client-discovery-response', 'requestId': message['requestId'],
'response': {'canHandle': False}})
if message.get('method') == 'thread-stream-state-changed':
params = message['params']
if (params.get('conversationId') == self.thread_id
and params.get('hostId') == 'local'
and message.get('sourceClientId') == self.owner):
if message.get('version') != 11:
raise ValueError('Unsupported Codex Desktop snapshot protocol')
self.on_change(params['change'])
if (message.get('method') == 'client-status-changed'
and message.get('params', {}).get('clientId') == self.owner
and message['params'].get('status') == 'disconnected'):
raise ConnectionError('Codex task owner disconnected')
return message
def request(self, method, params, target=None):
rid = str(uuid.uuid4())
message = {'type': 'request', 'requestId': rid, 'version': 1,
'sourceClientId': self.client_id, 'method': method,
'params': params, 'timeoutMs': 2500}
if target:
message['targetClientId'] = target
self.send(message)
deadline = time.monotonic() + 3
while time.monotonic() < deadline:
response = self.receive()
if response.get('requestId') == rid and response['type'] == 'response':
if response.get('resultType') != 'success':
raise RuntimeError(response.get('error', 'Codex rejected settings'))
return response
raise TimeoutError('Codex settings request timed out')
class Controller:
def __init__(self, target, changed, home=None, logger=lambda _: None, allowed=lambda: True,
target_info=None):
self.target = target
self.changed = changed
self.allowed = allowed
self.target_info = target_info or (lambda: {'kind': 'desktop'})
self.target_key = None
self.kind = 'desktop'
self.home = Path(home or os.environ.get('CODEX_HOME', Path.home() / '.codex'))
self.catalog = ModelCatalog(self.home)
self.socket_path = os.environ.get('BUSYBAR_CODEX_IPC', str(self.home / 'ipc/ipc.sock'))
self.logger = logger
self.lock = threading.Lock()
self.wake = threading.Event()
self.thread_id = None
self.state = {}
self.display_settings = {}
self.revision = None
self.pending = 0
self.pending_fast = deque()
self.due = 0
self.last_applied_at = 0
self.pending_since = 0
self.feedback_input_at = 0
self.confirmation_ms = None
self.display_ms = None
self.error = ''
self.feedback = None
self.feedback_revision = 0
self.feedback_until = 0
self.direction = 1
self.connected = False
def selection(self):
info = self.target_info()
target = info['thread_id'] if 'thread_id' in info else self.target()
return target, info, (target, info.get('kind'), info.get('socket'))
def status(self, display_thread_id=None):
with self.lock:
model, effort = model_effort(self.state)
return {'enabled': True, 'connected': self.connected, 'thread_id': self.thread_id,
'kind': self.kind,
'model': model, 'effort': effort, 'error': self.error,
'service_tier': codex_fast.current_tier(self.state),
'fast': codex_fast.is_fast(self.state),
'display': dict(self.display_settings.get(display_thread_id or self.thread_id, {})),
'direction': self.direction,
'confirmation_ms': self.confirmation_ms,
'display_ms': self.display_ms,
'feedback_revision': self.feedback_revision,
'feedback': self.feedback if time.monotonic() < self.feedback_until else None}
def rotate(self, delta):
if not self.allowed():
return False
target, _, key = self.selection()
with self.lock:
if not target or target != self.thread_id or not self.connected:
return False
if self.target_key is not None and key != self.target_key:
return False
if not self.pending:
self.pending_since = time.monotonic()
# Leading-edge throttle: later detents never postpone a write.
self.due = max(self.pending_since, self.last_applied_at + STEP_INTERVAL_S)
self.pending = max(-32, min(32, self.pending + int(delta)))
self.wake.set()
return True
def toggle_fast(self):
if not self.allowed():
return False
target, _, key = self.selection()
with self.lock:
if (not target or target != self.thread_id or not self.connected
or (self.target_key is not None and key != self.target_key)
or len(self.pending_fast) >= 8):
return False
self.pending_fast.append(time.monotonic())
self.wake.set()
return True
def mark_drawn(self, revision):
with self.lock:
if revision == self.feedback_revision and self.feedback_input_at:
self.display_ms = round((time.monotonic() - self.feedback_input_at) * 1000, 1)
def on_change(self, change):
with self.lock:
self.state, self.revision = apply_change(self.state, self.revision, change)
self.connected = True
model, effort = model_effort(self.state)
if self.thread_id and model and effort:
# Keep confirmed labels/speed visible after Cmd-H or focus loss.
# This bounded cache is never used to authorize a settings write.
self.display_settings.pop(self.thread_id, None)
self.display_settings[self.thread_id] = {
'model': model, 'effort': effort,
'service_tier': codex_fast.current_tier(self.state),
'fast': codex_fast.is_fast(self.state),
}
if len(self.display_settings) > 16:
self.display_settings.pop(next(iter(self.display_settings)))
self.changed()
def run(self, stop):
ipc = None
retry_at = 0
try:
while not stop.is_set():
self.wake.clear()
target, info, target_key = self.selection()
if target_key != self.target_key:
if ipc:
ipc.close()
ipc = None
with self.lock:
self.thread_id = target
self.target_key, self.kind = target_key, info.get('kind', 'desktop')
self.state, self.revision, self.pending = {}, None, 0
self.pending_fast.clear()
self.connected, self.feedback, self.error = False, None, ''
self.confirmation_ms = self.display_ms = None
self.last_applied_at = self.feedback_input_at = 0
retry_at = 0
self.changed()
if not target or time.monotonic() < retry_at:
stop.wait(0.1)
continue
delta = 0
fast_at = None
requested = None
try:
if ipc is None:
with self.lock:
self.state, self.revision, self.connected = {}, None, False
if info.get('kind') == 'cli':
if info.get('native_control'):
from codex_cli_native import NativeCLIIPC
ipc = NativeCLIIPC(info['socket'], self.on_change)
else:
from codex_cli_client import CLIIPC
ipc = CLIIPC(info['socket'], self.on_change)
else:
ipc = DesktopIPC(self.socket_path, self.on_change)
ipc.connect(target)
deadline = time.monotonic() + 3
while self.revision is None and time.monotonic() < deadline:
ipc.receive()
if self.revision is None:
raise TimeoutError('No Codex task snapshot')
with self.lock:
self.error = ''
if hasattr(ipc, 'poll'):
ipc.poll()
readable = select.select([ipc.sock], [], [], 0)[0]
if readable:
ipc.receive()
with self.lock:
delta = self.pending if time.monotonic() >= self.due else 0
fast_at = (self.pending_fast.popleft() if self.pending_fast
and (not self.pending or self.pending_fast[0] < self.pending_since) else None)
if fast_at is not None:
delta = 0
input_at = fast_at if fast_at is not None else self.pending_since
if delta:
self.pending = 0
state = copy.deepcopy(self.state)
if (not delta and fast_at is None) or self.selection()[2] != target_key or not self.allowed():
with self.lock:
wait = min(.05, max(0, self.due - time.monotonic())) if self.pending else .05
if not readable:
self.wake.wait(wait)
continue
model, current = model_effort(state)
if fast_at is not None:
try:
settings, enabled = codex_fast.toggle_settings(state, self.home, self.kind)
except (OSError, ValueError, KeyError, TypeError) as error:
raise CatalogError(str(error)) from error
requested = settings['serviceTier']
feedback = 'FAST' if enabled else 'NORMAL'
def confirmed():
return (model_effort(self.state)[0] == model
and codex_fast.current_tier(self.state) == requested)
else:
levels = (ipc.levels_for(model) if hasattr(ipc, 'levels_for')
else self.catalog.levels_for(model))
if current not in levels:
raise CatalogError(f'Current effort={current!r} absent from catalog '
f'for model={model!r}; levels={levels}')
requested = levels[max(0, min(len(levels) - 1, levels.index(current) + delta))]
settings = effort_settings(state, requested)
feedback = requested.upper()
def confirmed():
return model_effort(self.state) == (model, requested)
if not confirmed():
ipc.request('thread-follower-update-thread-settings',
{'conversationId': target, 'threadSettings': settings}, target=ipc.owner)
deadline = time.monotonic() + 2
while not confirmed() and time.monotonic() < deadline:
ipc.receive()
if not confirmed():
raise ValueError('Codex did not confirm settings change')
# Focus may have moved while awaiting the owner's reply.
if self.selection()[2] != target_key or not self.allowed():
continue
with self.lock:
self.feedback = feedback
self.feedback_revision += 1
self.direction = 1 if delta > 0 else -1
self.last_applied_at = time.monotonic()
self.feedback_until = self.last_applied_at + DURATION_S
self.feedback_input_at = input_at
self.confirmation_ms = round((self.last_applied_at - input_at) * 1000, 1)
self.display_ms = None
if self.pending:
self.due = self.last_applied_at + STEP_INTERVAL_S
self.error = ''
self.logger(f'Codex settings: {target} -> {feedback}')
self.changed()
except CatalogError as error:
# Keep the valid subscription and any newly queued dial steps.
# A missing/partially rewritten shared cache is not an IPC error.
with self.lock:
self.confirmation_ms = self.display_ms = None
self.error = str(error)
self.feedback = 'ERR'
self.feedback_revision += 1
self.feedback_until = time.monotonic() + 2.5
self.logger(f'Codex effort catalog: thread={target} {error}')
self.changed()
except (OSError, ValueError, KeyError, TypeError, RuntimeError) as error:
if ipc:
ipc.close()
ipc = None
with self.lock:
self.confirmation_ms = self.display_ms = None
self.error = str(error)
self.connected = False
self.pending = 0
self.pending_fast.clear()
self.feedback = 'ERR' if delta or fast_at is not None else None
self.feedback_revision += 1
self.feedback_until = time.monotonic() + 2.5
self.logger(f'Codex effort unavailable: kind={self.kind} thread={target} '
f'requested={requested} {error}')
self.changed()
retry_at = time.monotonic() + 3
finally:
if ipc:
ipc.close()codex_fast.py
"""Resolve the selected model's advertised Fast tier, never a global default."""
import copy
import json
from pathlib import Path
def current_tier(state):
return (state.get('latestThreadSettings') or {}).get('serviceTier')
def is_fast(state):
settings = state.get('latestThreadSettings') or {}
if 'serviceTier' not in settings:
return None
tier = settings['serviceTier']
return tier is not None and tier in ('fast', 'priority', settings.get('fastServiceTier'))
def toggle_settings(state, home, kind):
settings = state.get('latestThreadSettings') or {}
if 'serviceTier' not in settings:
raise ValueError('Fast control needs a current settings snapshot; restart the updated native CLI')
current = current_tier(state)
if kind == 'cli':
fast_tier = settings.get('fastServiceTier')
else:
data = json.loads((Path(home) / 'models_cache.json').read_text())
model = settings.get('model') or state.get('latestModel')
entry = next((m for m in data['models'] if m.get('slug') == model), {})
if current is None:
current = entry.get('default_service_tier')
fast_tier = next((tier['id'] for tier in entry.get('service_tiers', [])
if str(tier.get('name', '')).strip().lower() in ('fast', 'priority')
or tier.get('id') in ('priority', 'fast')), None)
if not fast_tier:
raise ValueError('Fast mode is not advertised for this model or account')
enabled = current not in ('fast', 'priority', fast_tier)
# `default` explicitly disables tier routing, even on Fast-by-default models.
update = {'serviceTier': fast_tier if enabled else 'default'}
mode = settings.get('collaborationMode') or state.get('latestCollaborationMode')
if mode:
update['collaborationMode'] = copy.deepcopy(mode)
return update, enabledcodex_focus.py
"""Read Codex Desktop's task-view lifecycle log, without UI automation.
`thread_stream_view_activity_changed` comes from setActive() on the task view,
not from model activity. Only primary windows count; a task's background output
cannot select it. An inactive event clears the selection even on Settings/Home.
"""
from __future__ import annotations
import os
from pathlib import Path
import re
import threading
import time
UUID = r'[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}'
LOG_NAME = re.compile(r'(codex-desktop-(' + UUID + r')-(\d+)-t0)-.*\.log$')
EVENT = re.compile(r'^\S+ info \[electron-message-handler\] thread_stream_view_activity_changed ')
FIELDS = re.compile(r'\b(\w+)=([^\s]+)')
class FocusState:
def __init__(self):
self.windows = {}
self.last_focused = None
self.sequence = 0
def feed(self, line):
# Other logger messages can provide window focus/visibility evidence,
# but never change that window's selected task.
if ' [electron-message-handler] ' not in line:
return
fields = dict(FIELDS.findall(line))
if fields.get('rendererWindowAppearance') != 'primary':
return
window_id = fields.get('rendererWindowId')
if window_id is None:
return
window = self.windows.setdefault(window_id, {'thread_id': None, 'visible': False, 'seq': 0})
if 'rendererWindowVisible' in fields:
window['visible'] = fields['rendererWindowVisible'] == 'true'
if fields.get('rendererWindowFocused') == 'true':
self.last_focused = window_id
if not EVENT.match(line):
return
thread_id = fields.get('conversationId', '')
if not re.fullmatch(UUID, thread_id):
return
self.sequence += 1
if fields.get('active') == 'true':
window['thread_id'], window['seq'] = thread_id, self.sequence
elif fields.get('active') == 'false' and window['thread_id'] == thread_id:
window['thread_id'] = None
def current(self):
candidates = [w['thread_id'] for w in self.windows.values()
if w['visible'] and w['thread_id']]
if len(candidates) > 1:
# Window focus can change without a new log record. Never guess
# between multiple task windows; the usual one-window tabs work.
return None
if self.last_focused in self.windows:
window = self.windows[self.last_focused]
return window['thread_id'] if window['visible'] else None
# Startup before focus metadata: only an unambiguous visible task.
return candidates[0] if len(candidates) == 1 else None
def process_alive(pid):
try:
os.kill(pid, 0)
return True
except OSError:
return False
class DesktopFocus:
def __init__(self, log_root=None, alive=process_alive):
self.log_root = Path(log_root or os.environ.get('BUSYBAR_CODEX_LOG_DIR',
Path.home() / 'Library/Logs/com.openai.codex'))
self.alive = alive
self.lock = threading.RLock()
self.state = FocusState()
self.run_id = None
self.pid = None
self.files = {}
self.scan_after = 0
self.poll_after = 0
self.error = 'Waiting for a Codex Desktop task view'
def _discover(self, now):
self.scan_after = now + 2
candidates = []
for path in self.log_root.glob('*/*/*/codex-desktop-*-t0-*.log'):
match = LOG_NAME.fullmatch(path.name)
if match:
try:
candidates.append((path.stat().st_mtime_ns, path, match.group(1), int(match.group(3))))
except OSError:
pass
candidates.sort(reverse=True)
live = {}
selected = next((item for item in candidates
if live.setdefault(item[3], self.alive(item[3]))), None)
if selected is None:
self.state, self.files, self.pid, self.run_id = FocusState(), {}, None, None
self.error = 'No running Codex Desktop log'
return
_, _, run_id, pid = selected
if run_id != self.run_id:
self.state, self.files = FocusState(), {}
self.run_id, self.pid = run_id, pid
for _, path, run, _ in reversed(candidates):
if run == run_id:
self.files.setdefault(path, (0, None))
self.scan_after = now + 2
def current(self, force=False):
with self.lock:
now = time.monotonic()
if not force and now < self.poll_after:
return self.state.current()
self.poll_after = now + .05
try:
if now >= self.scan_after:
self._discover(now)
if self.pid is None or not self.alive(self.pid):
self.state = FocusState()
self.scan_after = 0
return None
for path, (offset, identity) in list(self.files.items()):
try:
info = path.stat()
new_identity = (info.st_dev, info.st_ino)
if identity is not None and (identity != new_identity or info.st_size < offset):
# Lost lifecycle events are not reconstructable by inference.
self.state, self.files, self.run_id = FocusState(), {}, None
self.scan_after = 0
self.error = 'Codex log changed; waiting for a fresh task view'
return None
if info.st_size == offset:
continue
with path.open('rb') as stream:
stream.seek(offset)
while True:
start = stream.tell()
raw = stream.readline()
if not raw or not raw.endswith(b'\n'):
offset = start
break
self.state.feed(raw.decode('utf-8', errors='replace'))
offset = stream.tell()
self.files[path] = (offset, new_identity)
except FileNotFoundError:
self.files.pop(path, None)
self.error = '' if self.state.current() else 'No local task selected in a visible Codex window'
return self.state.current()
except OSError as error:
self.state = FocusState()
self.error = str(error)
return None
def status(self):
thread_id = self.current()
return {'source': 'desktop-view-log', 'thread_id': thread_id, 'error': self.error}
FOCUS = DesktopFocus()codex_target.py
"""Choose the visible Desktop task or focused CLI; never follow background output."""
from __future__ import annotations
import ctypes
import json
import os
from pathlib import Path
import plistlib
import re
import sys
import threading
import time
import codex_focus
TERMINALS = {'com.mitchellh.ghostty': 'ghostty', 'com.apple.Terminal': 'Apple_Terminal',
'com.googlecode.iterm2': 'iTerm.app', 'dev.warp.Warp-Stable': 'WarpTerminal',
'org.wezfurlong.wezterm': 'WezTerm', 'net.kovidgoyal.kitty': 'kitty',
'org.alacritty': 'Alacritty'}
DESKTOP = {'com.openai.codex', 'com.openai.chat', 'com.openai.ChatGPT'}
_FOREGROUND_API = None
_FOREGROUND_API_LOCK = threading.Lock()
class ProcessSerialNumber(ctypes.Structure):
_fields_ = [('high', ctypes.c_uint32), ('low', ctypes.c_uint32)]
def frontmost_pid(processes=None):
# NSWorkspace.frontmostApplication caches activation until AppKit's main
# run loop runs. These headless workers have no AppKit loop: query the
# WindowServer's front process on every poll instead.
global _FOREGROUND_API
if processes is None:
with _FOREGROUND_API_LOCK:
if _FOREGROUND_API is None:
api = ctypes.CDLL('/System/Library/Frameworks/ApplicationServices.framework/'
'Frameworks/HIServices.framework/HIServices')
# Framework Python otherwise registers a Dock application when
# querying WindowServer. Transform only once: repeating returns
# paramErr even though the process is already background-only.
api.TransformProcessType.argtypes = [ctypes.POINTER(ProcessSerialNumber), ctypes.c_uint32]
api.TransformProcessType.restype = ctypes.c_int32
current = ProcessSerialNumber(0, 2) # kCurrentProcess
status = api.TransformProcessType(ctypes.byref(current), 2) # background application
if status != 0:
raise OSError(f'macOS background registration failed (status={status})')
_FOREGROUND_API = api
processes = _FOREGROUND_API
processes.GetFrontProcess.argtypes = [ctypes.POINTER(ProcessSerialNumber)]
processes.GetFrontProcess.restype = ctypes.c_int16 # OSErr
processes.GetProcessPID.argtypes = [ctypes.POINTER(ProcessSerialNumber), ctypes.POINTER(ctypes.c_int32)]
processes.GetProcessPID.restype = ctypes.c_int32 # OSStatus
front, pid = ProcessSerialNumber(), ctypes.c_int32()
status = processes.GetFrontProcess(ctypes.byref(front))
if status == 0:
status = processes.GetProcessPID(ctypes.byref(front), ctypes.byref(pid))
if status != 0 or pid.value <= 0:
raise OSError(f'macOS foreground process unavailable (status={status})')
return pid.value
def bundle_for_pid(pid):
# Read executable metadata without loading AppKit. Framework Python turns
# these headless workers into Dock applications when AppKit initializes.
proc = ctypes.CDLL('/usr/lib/libproc.dylib')
proc.proc_pidpath.argtypes = [ctypes.c_int, ctypes.c_void_p, ctypes.c_uint32]
proc.proc_pidpath.restype = ctypes.c_int
buffer = ctypes.create_string_buffer(4096) # PROC_PIDPATHINFO_MAXSIZE
if proc.proc_pidpath(pid, buffer, len(buffer)) <= 0:
return None
executable = Path(os.fsdecode(buffer.value))
for directory in executable.parents:
if directory.suffix != '.app':
continue
try:
with (directory / 'Contents/Info.plist').open('rb') as stream:
return plistlib.load(stream).get('CFBundleIdentifier')
except (OSError, ValueError, plistlib.InvalidFileException):
continue
return None
def frontmost_bundle():
return bundle_for_pid(frontmost_pid()) if sys.platform == 'darwin' else None
def cli_sessions(home):
records = []
paths = list((Path(home) / 'tui-control').glob('*/session.json'))
paths += list((Path(home) / 'busybar-cli').glob('*/session.json'))
for path in paths:
try:
if path.stat().st_uid != os.getuid():
continue
value = json.loads(path.read_text())
if path.parent.parent.name == 'tui-control':
from codex_cli_native import normalize
value = normalize(value)
if not isinstance(value, dict) or not isinstance(value.get('pid'), int):
continue
os.kill(value['pid'], 0)
if value['pid'] <= 0 or not re.fullmatch(codex_focus.UUID, value.get('thread_id') or ''):
continue
# Control must stay in the same private launch directory.
if value.get('socket') != str(path.parent / 'control.sock'):
continue
records.append(value)
except (OSError, ValueError, TypeError):
continue
return records
def choose_cli(records, terminal=None):
candidates = [record for record in records if record.get('focused')
and (terminal is None or record.get('terminal') == terminal)]
return candidates[0] if len(candidates) == 1 and candidates[0].get('ready') else None
class Target:
def __init__(self, home=None, foreground=frontmost_bundle, sessions=cli_sessions):
self.home = Path(home or os.environ.get('CODEX_HOME', Path.home() / '.codex'))
self.foreground = foreground
self.sessions = sessions
self.lock = threading.RLock()
self.next_poll = 0
self.selected = None
self.last_display = None
self.error = ''
self.bundle = None
def snapshot(self):
return {**(self.selected or {'source': None, 'thread_id': None, 'error': self.error}),
'foreground_bundle': self.bundle}
def status(self, force=False):
with self.lock:
now = time.monotonic()
if not force and now < self.next_poll:
return self.snapshot()
self.next_poll = now + .2
self.selected = None
self.error = ''
self.bundle = None
try:
bundle = self.bundle = self.foreground()
if bundle in DESKTOP:
thread_id = codex_focus.FOCUS.current(force=force)
if thread_id:
self.selected = {'source': 'desktop-view-log', 'thread_id': thread_id,
'kind': 'desktop', 'error': ''}
elif bundle in TERMINALS or (bundle is None and sys.platform != 'darwin'):
records = self.sessions(self.home)
record = choose_cli(records, TERMINALS.get(bundle))
if record:
self.selected = {**record, 'kind': 'cli', 'source': 'cli-focus', 'error': ''}
else:
matching = [r for r in records if r.get('focused') and
(bundle is None or r.get('terminal') == TERMINALS[bundle])]
self.error = ((matching[0].get('error') if len(matching) == 1 else None)
or 'No uniquely focused CLI with native TUI control')
else:
self.error = 'Codex Desktop or a CLI terminal is not in the foreground'
except (OSError, ValueError, TypeError) as error:
self.error = str(error)
if self.selected:
self.last_display = dict(self.selected)
return self.snapshot()
def current(self, force=False):
return self.status(force).get('thread_id')
def display(self):
with self.lock:
selected = self.status()
if selected.get('thread_id'):
return selected
# Hiding an app or waiting for terminal focus reports must not
# replace the last selected task with another Desktop view/config.
# This fallback is display-only; current() still rejects controls.
if self.last_display:
if self.last_display.get('kind') == 'cli':
records = self.sessions(self.home)
record = next((r for r in records
if r['socket'] == self.last_display['socket']
and r['thread_id'] == self.last_display['thread_id']), None)
if record and record.get('ready'):
self.last_display = {**record, 'kind': 'cli'}
return dict(self.last_display)
thread_id = codex_focus.FOCUS.current()
if thread_id:
self.last_display = {'kind': 'desktop', 'thread_id': thread_id}
return dict(self.last_display or {})
FOCUS = Target()codex_usage.py
"""Fresh account quotas via Codex's documented account/rateLimits/read API.
Only initialize and read account limits. No task is started, no model is called,
and authentication stays inside the installed Codex executable.
"""
from __future__ import annotations
import copy
import json
import math
import os
from pathlib import Path
import queue
import shutil
import subprocess
import threading
import time
POLL_S = 60
MAX_AGE_S = 180
REQUEST_TIMEOUT_S = 8
def number(value):
return (isinstance(value, (int, float)) and not isinstance(value, bool)
and math.isfinite(value))
def window_name(minutes):
hours = minutes / 60
return f'{round(hours)}h' if hours < 48 else f'{round(hours / 24)}d'
def account_quotas(result, observed_at, limit_id='codex'):
"""Select an explicit account bucket, never whichever model reported last."""
buckets = result.get('rateLimitsByLimitId')
if isinstance(buckets, dict):
bucket = buckets.get(limit_id)
else:
bucket = result.get('rateLimits')
if isinstance(bucket, dict) and bucket.get('limitId') not in (None, limit_id):
bucket = None
if not isinstance(bucket, dict):
raise ValueError(f'Account limit bucket {limit_id!r} unavailable')
quotas = []
for key in ('primary', 'secondary'):
window = bucket.get(key)
if not isinstance(window, dict):
continue
used, minutes, reset = (window.get(k) for k in
('usedPercent', 'windowDurationMins', 'resetsAt'))
if not number(minutes) or minutes <= 0 or not number(used):
continue # null is unavailable, not zero usage
reset = reset if number(reset) and reset > 0 else None
quotas.append({'name': window_name(minutes),
'left_pct': max(0., min(100., 100 - used)),
'window_minutes': minutes, 'resets_at': reset,
'observed_at': observed_at,
'valid_until': observed_at + MAX_AGE_S})
if not quotas:
raise ValueError(f'No quota windows available for {limit_id!r}')
return quotas
def codex_binary(env):
override = env.get('BUSYBAR_CODEX_BIN')
if override:
return override
for path in ('/Applications/ChatGPT.app/Contents/Resources/codex',
'/Applications/Codex.app/Contents/Resources/codex'):
if Path(path).is_file():
return path
path = shutil.which('codex', path=env.get('PATH'))
if not path:
raise FileNotFoundError('Codex executable not found; set BUSYBAR_CODEX_BIN')
return path
def read_rate_limits(env=None, stop=None):
env = dict(os.environ if env is None else env)
proc = subprocess.Popen(
[codex_binary(env), 'app-server', '--listen', 'stdio://'],
stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
env=env, text=True, bufsize=1,
)
replies = queue.Queue()
def read_stdout():
try:
for line in proc.stdout:
try:
value = json.loads(line)
if isinstance(value, dict):
replies.put(value)
except ValueError:
pass
finally:
replies.put(None)
reader = threading.Thread(target=read_stdout, daemon=True)
reader.start()
deadline = time.monotonic() + REQUEST_TIMEOUT_S
def send(value):
proc.stdin.write(json.dumps(value) + '\n')
proc.stdin.flush()
def request(request_id, method, params=None):
value = {'id': request_id, 'method': method}
if params is not None:
value['params'] = params
send(value)
while time.monotonic() < deadline:
if stop is not None and stop.is_set():
raise InterruptedError('Usage reader stopped')
try:
reply = replies.get(timeout=.1)
except queue.Empty:
continue
if reply is None:
raise ConnectionError('Codex account reader disconnected')
if reply.get('id') != request_id:
continue
if 'error' in reply:
error = reply['error']
code = error.get('code') if isinstance(error, dict) else None
raise RuntimeError(f'Codex account request failed (code={code})')
if not isinstance(reply.get('result'), dict):
raise ValueError('Invalid Codex account response')
return reply['result']
raise TimeoutError('Codex account request timed out')
try:
request(1, 'initialize', {'clientInfo': {
'name': 'busybar_usage', 'title': 'BUSY Bar usage display', 'version': '1.0.0'}})
send({'method': 'initialized'})
return request(2, 'account/rateLimits/read')
finally:
# A short-lived account reader picks up refreshed login state each poll
# and cannot leave an idle app-server or a blocked stdout reader behind.
try:
proc.stdin.close()
except OSError:
pass
try:
proc.wait(timeout=1)
except subprocess.TimeoutExpired:
proc.terminate()
try:
proc.wait(timeout=1)
except subprocess.TimeoutExpired:
proc.kill()
proc.wait()
reader.join(timeout=1)
proc.stdout.close()
class Monitor:
def __init__(self, env=None, fetch=None, clock=time.time):
self.env = dict(os.environ if env is None else env)
self.limit_id = self.env.get('BUSYBAR_CODEX_LIMIT_ID', 'codex')
self.fetch = fetch or (lambda stop: read_rate_limits(self.env, stop))
self.clock = clock
self.lock = threading.Lock()
self.changed = threading.Event()
self.quotas = []
self.observed_at = None
self.error = ''
def refresh(self, stop=None):
try:
result = self.fetch(stop)
observed_at = self.clock()
quotas = account_quotas(result, observed_at, self.limit_id)
with self.lock:
self.quotas, self.observed_at, self.error = quotas, observed_at, ''
except (OSError, ValueError, TypeError, RuntimeError) as error:
with self.lock:
self.error = str(error)[:180]
self.changed.set()
def snapshot(self):
with self.lock:
now = self.clock()
usable = any(q['valid_until'] > now and
(q['resets_at'] is None or q['resets_at'] > now)
for q in self.quotas)
state = ('cached' if self.error else 'fresh') if usable else 'unavailable'
return {'quotas': copy.deepcopy(self.quotas) or None,
'quota_status': {'source': 'codex-account', 'limit_id': self.limit_id,
'state': state, 'observed_at': self.observed_at,
'error': self.error}}
def next_delay(self):
with self.lock:
if self.error:
return 15
# Fetch shortly after an actual reset, even if the next regular
# minute has not arrived. Never manufacture a new seven-day window.
remaining = [q['resets_at'] - self.clock() for q in self.quotas
if q['resets_at'] is not None]
if any(seconds <= 0 for seconds in remaining):
return 15
return min([POLL_S] + [max(1., seconds + 1) for seconds in remaining])
def run(self, stop):
while not stop.is_set():
self.refresh(stop)
stop.wait(self.next_delay())daemon.py
#!/usr/bin/env python3
"""BusyBar agent-status daemon.
A provider-agnostic status display core with pluggable adapters and
transports:
claude-code (built-in adapter: /statusline + /state) --.
codex / cursor / anything (POST /v1/report) ----------+--> SessionStore
| |
GET /status <--------------+ Renderer
(device JS app, debugging) |
Transport
(usb / wifi / cloud)
The CORE understands only the normalized report schema (see
docs/EXTENDING.md): state, label, label_color, context_pct, quotas.
Everything Claude-specific — statusline JSON parsing, /effort palette
colors, the 5h/7d rate-limit windows — lives in the claude adapter
functions and never leaks into the renderer.
Layout (72x16 front display):
############################ 1px per-pixel animated ring (.anim)
# Fable 5 max [##----] # label | time-to-reset progress
# W [quota bar] A WORK # quota remaining | Astra | state word
############################
"""
from __future__ import annotations
import hmac
import json
import math
import os
import re
import secrets
import signal
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
import ai_status
import codex_effort
import codex_focus
import codex_target
import effort_animation
import fast_animation
from display_scene import DrawCache
from busybar_http import HttpTransport, local_opener
from busybar_input import InputStream, input_stream_url
# --------------------------------------------------------------------------
# Config
# --------------------------------------------------------------------------
# An explicit port supports isolated test rigs and the standalone gallery app.
# Installed hooks share BUSYBAR_PORT through report.py/report.sh (default 8765).
LISTEN_PORT = (int(sys.argv[sys.argv.index("--port") + 1]) if "--port" in sys.argv
else int(os.environ.get("BUSYBAR_PORT", "8765")))
# Bind addresses (env BUSYBAR_LISTEN, comma-separated). Default: loopback +
# the USB network (device side) only. "0.0.0.0" turns this daemon into the
# hub for the other computers on your LAN (README: "Several computers").
LISTEN_ADDRS = [a.strip() for a in
os.environ.get("BUSYBAR_LISTEN", "127.0.0.1,10.0.4.21").split(",")
if a.strip()] or ["127.0.0.1"]
# Optional shared secret for reports arriving from other machines (env
# BUSYBAR_HUB_TOKEN on the hub and on every client). Loopback never needs it.
HUB_TOKEN = os.environ.get("BUSYBAR_HUB_TOKEN", "")
# Standby role (README: "When the hub sleeps"): BUSYBAR_HUB names the hub;
# with BUSYBAR_STANDBY=1 this daemon mirrors every session there and
# renders only while the hub has been unreachable for HUB_DOWN_AFTER_S.
HUB_URL = os.environ.get("BUSYBAR_HUB", "").strip().rstrip("/")
STANDBY = os.environ.get("BUSYBAR_STANDBY", "").strip().lower() in ("1", "true", "yes", "on")
if STANDBY and not HUB_URL:
sys.exit("BUSYBAR_STANDBY needs BUSYBAR_HUB (the hub's URL)")
HUB_POLL_S = 3.0 # standby: probe cadence while nothing is queued
HUB_TIMEOUT_S = 2.0 # ... per request to the hub
HUB_DOWN_FAILURES = 3 # probes failing in a row (~10 s) before a standby takes over
MIRROR_LEASE_S = 90.0 # the hub forgets a mirrored session not refreshed within this
MIRROR_HEARTBEAT_S = 30.0 # ... so a standby re-mirrors everything live this often
SUSPEND_GAP_S = 30.0 # a loop iteration this late means the computer was asleep
INSTANCE = secrets.token_hex(8) # this process; GET /hub reports it
# LAN and loopback traffic never goes through a proxy (HTTP_PROXY, Windows
# system proxy): a proxy is never the route to 127.0.0.1, the USB link, the
# Bar's LAN address or the hub's .local name.
OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))
USB_SOURCE_IP = os.environ.get("BUSYBAR_USB_SOURCE_IP", "10.0.4.21").strip()
# How the daemon renders to the device (env BUSYBAR_RENDER_MODE overrides):
# "auto" - whenever any reporting agent is active (the always-on behavior)
# "theme" - only while "claude" is the device's currently selected
# BUSY/CUSTOM theme (on-device manual switch; see claude_card.py)
# "off" - data bridge only
RENDER_MODE = os.environ.get("BUSYBAR_RENDER_MODE", "auto")
# Display style (env BUSYBAR_STYLE): "minimal" (text state word + quota
# gauges) or "avatar" (a pixel companion acts out the state).
STYLE = os.environ.get("BUSYBAR_STYLE", "minimal")
AI_STATUS_ENABLED = os.environ.get("BUSYBAR_AI_STATUS", "").strip().lower() \
in ("1", "true", "yes", "on")
AI_STATUS_URL = os.environ.get("BUSYBAR_AI_STATUS_URL", ai_status.API_URL).strip()
X_STATUS_URL = os.environ.get("BUSYBAR_X_STATUS_URL", ai_status.X_STATUS_URL).strip()
GOOGLE_STATUS_URL = os.environ.get(
"BUSYBAR_GOOGLE_STATUS_URL", ai_status.GOOGLE_STATUS_URL,
).strip()
AI_STATUS_POLL_S = float(os.environ.get("BUSYBAR_AI_STATUS_POLL_S", ai_status.POLL_S))
ASTRA_STATE_PATH = os.path.expanduser(os.environ.get(
"BUSYBAR_ASTRA_STATE",
"~/.local/state/astra-watch/state.json",
))
ASTRA_STALE_S = float(os.environ.get("BUSYBAR_ASTRA_STALE_S", "1800"))
ASTRA_CHECK_INTERVAL_S = max(60.0, float(os.environ.get(
"BUSYBAR_ASTRA_CHECK_INTERVAL_S", "600",
)))
ASTRA_WATCH_SCRIPT = os.path.expanduser(os.environ.get(
"BUSYBAR_ASTRA_WATCH_SCRIPT",
"~/plugins/astra-watch/scripts/astra_watch.py",
))
X_PULSE_ENABLED = os.environ.get("BUSYBAR_X_PULSE", "").strip().lower() \
in ("1", "true", "yes", "on")
X_PULSE_BACKEND = os.environ.get("BUSYBAR_X_PULSE_BACKEND", "bird").strip().lower()
X_PULSE_SSH_HOST = os.environ.get("BUSYBAR_X_PULSE_SSH_HOST", "local").strip()
X_PULSE_APP = os.environ.get("BUSYBAR_X_PULSE_APP", "busy-codex").strip()
X_PULSE_USERNAME = os.environ.get(
"BUSYBAR_X_PULSE_USERNAME", "",
).strip()
X_PULSE_POLL_S = max(300.0, float(os.environ.get(
"BUSYBAR_X_PULSE_POLL_S", "21600",
)))
X_PULSE_MAX_RESULTS = max(10, min(100, int(os.environ.get(
"BUSYBAR_X_PULSE_MAX_RESULTS", "25",
))))
X_PULSE_MAX_PAGES = max(1, min(5, int(os.environ.get(
"BUSYBAR_X_PULSE_MAX_PAGES", "1",
))))
X_PULSE_BIRD_PATH = os.environ.get(
"BUSYBAR_X_PULSE_BIRD_PATH", "/usr/local/bin/bird",
).strip()
X_PULSE_STATE_PATH = os.path.expanduser(os.environ.get(
"BUSYBAR_X_PULSE_STATE",
"~/.local/state/astra-watch/x-pulse.json",
))
X_PULSE_CLASSIFIER_VALIDATED = os.environ.get(
"BUSYBAR_X_PULSE_CLASSIFIER_VALIDATED", "",
).strip().lower() in ("1", "true", "yes", "on")
X_PULSE_LLM_ENABLED = os.environ.get(
"BUSYBAR_X_PULSE_LLM", "",
).strip().lower() in ("1", "true", "yes", "on")
X_PULSE_LLM_MODEL = os.environ.get(
"BUSYBAR_X_PULSE_LLM_MODEL", "gpt-5.6-luna",
).strip()
X_PULSE_LLM_MAX_ITEMS = max(1, min(16, int(os.environ.get(
"BUSYBAR_X_PULSE_LLM_MAX_ITEMS", "12",
))))
X_PULSE_LLM_TIMEOUT_S = max(20.0, float(os.environ.get(
"BUSYBAR_X_PULSE_LLM_TIMEOUT_S", "90",
)))
X_PULSE_CODEX_PATH = os.path.expanduser(os.environ.get(
"BUSYBAR_X_PULSE_CODEX_PATH", "codex",
)).strip()
APP_NAME = os.environ.get("BUSYBAR_APP_NAME", "claude_status") # assets share this namespace
DRAW_PRIORITY = int(os.environ.get("BUSYBAR_DRAW_PRIORITY", "50"))
THEME_NAME = "claude" # installed in /ext/apps_assets/busy/themes/
SNAPSHOT_POLL_S = 2.0
TEXT_TIMEOUT_S = 15
ANIM_TIMEOUT_S = 120
ANIM_REFRESH_S = 60.0
KEEPALIVE_S = 8.0
COMPLETE_HOLD_S = 5.0
# Idle release: after this many seconds of IDLE the screen is handed
# back to the device (env BUSYBAR_IDLE_CLEAR_S; 0 = keep forever).
IDLE_CLEAR_AFTER_S = float(os.environ.get("BUSYBAR_IDLE_CLEAR_S", "600"))
DEFAULT_TTL_S = 6 * 3600
STATES = ("THINKING", "WORKING", "WAIT", "ERROR", "FAILED", "COMPLETE", "IDLE")
STATE_ANIMS = {
"THINKING": "think.anim", "WORKING": "work.anim", "WAIT": "wait.anim",
"ERROR": "error.anim", "FAILED": "error.anim", "COMPLETE": "done.anim",
"IDLE": "idle.anim",
}
STATE_WORDS = {
"THINKING": "THINK", "WORKING": "WORK", "WAIT": "WAIT",
"ERROR": "ERR", "FAIL": "FAIL", "FAILED": "FAIL", "COMPLETE": "DONE",
"IDLE": "IDLE",
}
STATE_WORDS_FULL = {
"THINKING": "thinking", "WORKING": "working", "WAIT": "waiting",
"ERROR": "error", "FAILED": "failed", "COMPLETE": "done", "IDLE": "idle",
}
AVATAR_ANIMS = {
"THINKING": "claw_think.anim", "WORKING": "claw_work.anim",
"WAIT": "claw_wait.anim", "ERROR": "claw_error.anim",
"FAILED": "claw_error.anim", "COMPLETE": "claw_done.anim",
"IDLE": "claw_idle.anim",
}
AVATAR_X, AVATAR_Y = 55, 1
STATE_COLORS = {
"THINKING": "#AF87FFFF", "WORKING": "#FFB000FF", "WAIT": "#FF6A00FF",
"ERROR": "#FF2020FF", "FAILED": "#FF2020FF", "COMPLETE": "#20C040FF",
"IDLE": "#808080FF",
}
LABEL_FALLBACK_COLOR = "#FFFFFFFF"
HOST_TAG_COLOR = "#8C8C8CFF" # text host tag (protocol `host_tag`)
# States that pull the display to a session: the user just spoke to it
# (THINKING) or it is asking for the user (WAIT).
ATTENTION_STATES = ("THINKING", "WAIT")
QUOTA_COLOR = "#A0A0A0FF"
FONT = "small"
BAR_X, BAR_Y, BAR_W, BAR_H = 50, 3, 20, 4
BAR_TRACK_COLOR = "#262626FF"
LABEL_MAX_PX = BAR_X - 2 - 3
QUOTA_BAR_X, QUOTA_BAR_Y, QUOTA_BAR_W, QUOTA_BAR_H = 10, 10, 35, 4
WEEK_SECONDS = 7 * 24 * 60 * 60
ASTRA_COLORS = {
"waiting": "#606060FF", "hidden": "#FFB000FF",
"available": "#20C040FF", "error": "#FF2020FF",
"stale": "#FF2020FF", "unknown": "#00000000",
}
ASTRA_REFRESH_LOCK = threading.Lock()
ASTRA_REFRESHED_AT = 0.0
ASTRA_APP_LOCK = threading.Lock()
ASTRA_APP_LAST_SEEN = 0.0
ASTRA_APP_REQUESTS = 0
ASTRA_APP_ACTIVE_S = 5.0
def _fetch_x_pulse() -> dict:
return x_pulse.fetch(
X_PULSE_SSH_HOST,
backend=X_PULSE_BACKEND,
app=X_PULSE_APP,
username=X_PULSE_USERNAME,
max_results=X_PULSE_MAX_RESULTS,
state_path=X_PULSE_STATE_PATH,
classifier_validated=X_PULSE_CLASSIFIER_VALIDATED,
max_pages=X_PULSE_MAX_PAGES,
bird_path=X_PULSE_BIRD_PATH,
llm_enabled=X_PULSE_LLM_ENABLED,
llm_model=X_PULSE_LLM_MODEL,
llm_max_items=X_PULSE_LLM_MAX_ITEMS,
llm_timeout_s=X_PULSE_LLM_TIMEOUT_S,
codex_path=X_PULSE_CODEX_PATH,
)
if X_PULSE_ENABLED:
import x_pulse # optional integration; absent from the focused gallery package
X_PULSE_MONITOR = x_pulse.Monitor(
_fetch_x_pulse,
interval_s=X_PULSE_POLL_S,
stale_s=max(3 * X_PULSE_POLL_S, 6 * 3600),
) if X_PULSE_ENABLED else None
if X_PULSE_MONITOR is not None:
try:
cached = x_pulse.cached_summary(
X_PULSE_STATE_PATH,
backend=X_PULSE_BACKEND,
classifier_validated=X_PULSE_CLASSIFIER_VALIDATED,
llm_enabled=X_PULSE_LLM_ENABLED,
llm_model=X_PULSE_LLM_MODEL,
)
if cached is not None:
X_PULSE_MONITOR.seed(cached[0], checked_at=cached[1])
except Exception:
# A damaged cache must never prevent the local status daemon starting.
pass
# --------------------------------------------------------------------------
# Transports (env BUSYBAR_TRANSPORT: usb | wifi | cloud; see docs/EXTENDING.md)
# --------------------------------------------------------------------------
def usb_opener():
return local_opener(USB_SOURCE_IP)
def make_transport() -> HttpTransport:
kind = os.environ.get("BUSYBAR_TRANSPORT", "usb")
if kind == "usb":
if STANDBY:
sys.exit("a standby cannot use the hub's USB link: set BUSYBAR_TRANSPORT=wifi "
"(BUSYBAR_DEVICE, BUSYBAR_TOKEN) - setup_claude.py install --standby ...")
return HttpTransport("http://10.0.4.20/api", opener=usb_opener(), logger=log)
if kind == "wifi":
host = os.environ.get("BUSYBAR_DEVICE", "")
legacy = os.environ.get("BUSYBAR_HOST", "")
if not host and re.fullmatch(r"[0-9.]+|.+\.local", legacy):
log("BUSYBAR_HOST as the Bar's address is deprecated: set BUSYBAR_DEVICE "
"(BUSYBAR_HOST now names this computer)")
host = legacy
host = re.sub(r"^https?://", "", host).strip().rstrip("/")
if not host:
sys.exit("wifi transport needs BUSYBAR_DEVICE (the Bar's LAN IP or name)")
headers = {}
if os.environ.get("BUSYBAR_TOKEN"):
headers["x-api-token"] = os.environ["BUSYBAR_TOKEN"]
t = HttpTransport(f"http://{host}/api", headers, logger=log)
t.TIMEOUT_S = 4.0
return t
if kind == "cloud":
token = os.environ.get("BUSYBAR_TOKEN")
if not token:
sys.exit("cloud transport needs BUSYBAR_TOKEN (API token from the BUSY app)")
t = HttpTransport("https://api.busy.app/busybar",
{"authorization": f"Bearer {token}"},
opener=urllib.request.build_opener(), logger=log) # the internet: proxies apply
t.TIMEOUT_S = 8.0
return t
if kind == "ble":
sys.exit("BLE transport is designed but not implemented yet - see docs/EXTENDING.md")
sys.exit(f"unknown BUSYBAR_TRANSPORT {kind!r} (usb|wifi|cloud)")
# --------------------------------------------------------------------------
# Session store (normalized records only)
# --------------------------------------------------------------------------
class Store:
def __init__(self):
self.lock = threading.Lock()
self.sessions: dict[str, dict] = {}
self.tombstones: dict = {} # key -> (rev, when): mirrored sessions that ended
self.ended_rev: dict = {} # key -> rev for the tombstone we mirror ourselves
self.dirty = threading.Event()
self.on_report = None # standby: HubLink.enqueue (called outside the lock)
def report(self, source: str, session_id: str, fields: dict,
mirrored: bool = False) -> bool:
"""Merge a normalized report. `fields` may contain: state, label,
label_color, context_pct, quotas, quota_status, badges, host, host_tag, ttl_s,
ended - and, from a standby's mirror only: rev, lease_s, focus_ts,
state_ts, last_active (the hub derives these from ages)."""
key = f"{source}:{session_id}"
now = time.time()
created = False
with self.lock:
for k in [k for k, (_, when) in self.tombstones.items() if now - when > 300]:
del self.tombstones[k]
self.ended_rev.pop(k, None)
rev = fields.get("rev")
if rev is not None:
# Mirrors can arrive late or twice; a record only moves forward.
dead = self.tombstones.get(key)
cur = self.sessions.get(key)
if (dead and rev <= dead[0]) or (cur and rev < cur["rev"]):
return False
if fields.get("ended"):
s = self.sessions.pop(key, None)
if rev is not None:
self.tombstones[key] = (rev, now)
elif s is not None:
self.ended_rev[key] = s["rev"] + 1
else:
created = key not in self.sessions
s = self.sessions.setdefault(key, {
"source": source, "state": "IDLE", "state_ts": 0.0,
"last_active": 0.0, "focus_ts": 0.0, "seen_ts": 0.0,
"label": None, "label_color": None,
"context_pct": None, "quotas": None, "quota_status": None, "badges": None,
"host": None, "host_tag": None, "ttl_s": DEFAULT_TTL_S,
"rev": 0, "mirrored": False, "lease_s": None,
})
if "state" in fields:
new, prev = fields["state"], s["state"]
# Attention events: the user spoke to this session, it
# wants the user, or a new task just started in it.
# Only these move the display between sessions.
if new in ATTENTION_STATES or (
new == "WORKING" and prev in ("IDLE", "COMPLETE")):
s["focus_ts"] = now
s["state"] = new
# Repeated snapshots are keepalives, not new completions.
# Mirrors may supply the origin's actual transition time.
if created or new != prev or "state_ts" in fields:
s["state_ts"] = fields.get("state_ts", now)
if "focus_ts" in fields:
s["focus_ts"] = fields["focus_ts"] # mirrored: the origin decided
for k in ("label", "label_color", "context_pct", "quotas", "quota_status",
"badges", "host", "host_tag", "ttl_s", "control_thread_id"):
if k in fields:
s[k] = fields[k]
s["lease_s"] = fields.get("lease_s") if mirrored else None
s["last_active"] = fields.get("last_active", now)
s["seen_ts"] = now
s["mirrored"] = mirrored
s["rev"] = rev if rev is not None else max(s["rev"] + 1, int(now * 1000))
if self.on_report is not None and not mirrored:
self.on_report(source, session_id)
self.dirty.set()
return created # a mirror that created a record must re-send its state
def live_keys(self) -> list:
"""(source, session_id) of every session that originated here."""
with self.lock:
return [(s["source"], k[len(s["source"]) + 1:])
for k, s in self.sessions.items() if not s["mirrored"]]
def drop_mirrored(self):
with self.lock:
for k in [k for k, s in self.sessions.items() if s["mirrored"]]:
del self.sessions[k]
def active_session(self) -> dict | None:
now = time.time()
with self.lock:
for key in [k for k, s in self.sessions.items()
if now - s["last_active"] > s.get("ttl_s", DEFAULT_TTL_S)
or (s.get("lease_s") and now - s["seen_ts"] > s["lease_s"])]:
del self.sessions[key]
if not self.sessions:
return None
# The display follows attention, not chatter: among the sessions
# doing something, the one the user last talked to wins; a
# background session surfaces only once the focused one is idle.
return dict(max(self.sessions.values(), key=lambda s: (
effective_state(s) != "IDLE", s["focus_ts"], s["last_active"])))
STORE = Store()
STOP = threading.Event() # set to make the daemon exit (signals, POST /shutdown)
REDRAW = threading.Event() # POST /redraw: repaint (or clear) the whole canvas
AI_MONITOR: ai_status.Monitor | None = None
DEVICE_INPUT_LOCK = threading.Lock()
DEVICE_MODE: str | None = None
DEVICE_INPUT_CONNECTED = False
DEVICE_INPUT_ERROR = ""
LAST_ENCODER = {}
LAST_START = {}
START_DOWN = False
EFFORT_CONTROLLER = None
def effort_target():
pinned = os.environ.get("BUSYBAR_CODEX_THREAD_ID", "")
if pinned:
return pinned if re.fullmatch(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}", pinned) else None
return codex_target.FOCUS.current(force=True)
def effort_target_info():
pinned = os.environ.get("BUSYBAR_CODEX_THREAD_ID", "")
if pinned:
record = next((r for r in codex_target.cli_sessions(codex_target.FOCUS.home)
if r.get("thread_id") == pinned), None)
return {**record, "kind": "cli"} if record else {"kind": "desktop", "thread_id": effort_target()}
return codex_target.FOCUS.status()
def effort_input_block_reason():
target = effort_target()
if not target:
return effort_target_info().get('error') or 'No Codex task selected'
session = STORE.active_session()
if session is None or session.get('control_thread_id') != target:
return 'Waiting for the display adapter to select the foreground task'
if not device_canvas_allowed() or DEVICE_MODE in ('OFF', 'UNKNOWN'):
return 'Device mode owns the dial'
if not DRAWN.is_set() or (AI_MONITOR and AI_MONITOR.drawn):
return 'Codex status is not being displayed'
if TRANSPORT is not None and TRANSPORT.last_http_status == 409:
return 'Another application owns the display'
if HUBLINK is not None and not HUBLINK.takeover:
return 'Another hub owns the display'
return ''
def effort_input_allowed():
return not effort_input_block_reason()
# --------------------------------------------------------------------------
# Standby role: mirror every session to the hub, render only while it is down
# --------------------------------------------------------------------------
MIRROR_FIELDS = ("label", "label_color", "context_pct", "quotas", "quota_status", "badges",
"host", "host_tag", "ttl_s")
RENDER_LOCK = threading.Lock() # one canvas transaction at a time (a yield waits on it)
DRAWN = threading.Event() # what is on the Bar right now was painted by us
class HubLink:
"""Standby role (README: "When the hub sleeps").
Mirrors every session that originates here to the hub as full
/v1/report records: coalesced per session, delivered in order, `state`
only when it changed since the hub last acknowledged it (so a hub of
any version sees exactly the transitions a direct client would have
sent), ages instead of timestamps (clock skew between computers does
not matter, LAN latency does not either), a lease so the hub forgets
us if we vanish, refreshed by a heartbeat.
Decides when this daemon may paint the Bar: after HUB_DOWN_FAILURES
probes fail in a row while the Bar itself still answers (evidence, not
wall clock - a laptop waking from sleep never takes over by mistake),
or when the hub says it cannot reach the Bar. When the hub is back:
resync, ask it to repaint, then yield - in that order and under
RENDER_LOCK, so no draw of ours lands after the hub's repaint."""
def __init__(self, hub: str, token: str, transport: HttpTransport):
self.hub = hub
self.transport = transport
self.headers = {"X-Busybar-Mirror": "1"}
if token:
self.headers["X-Busybar-Token"] = token
self.lock = threading.Lock()
self.queue: dict = {} # (source, session_id) -> seq, insertion-ordered
self.seq = 0
self.sent_state: dict = {} # key -> state the hub has acknowledged
self.wake = threading.Event()
self.failures = 0 # consecutive failed probes/deliveries
self.last_ok = time.time()
self.last_probe = 0.0
self.last_heartbeat = time.time()
self.last_error = ""
self.hub_info: dict = {} # last GET /hub body ({} for an older hub)
self.hub_instance = None
self.legacy_hub = False # answered 404 to GET /hub: probe /health instead
self.rejects: dict = {} # path -> last 4xx text, until that path succeeds
self.device_down = 0 # consecutive probes with the hub's device_ok false
self.takeover = False # True: this daemon paints the Bar
self.rendering = False
self.disabled = "" # why the link refuses to run, if it does
self._style_warned = False
self._no_net = False
def enqueue(self, source: str, session_id: str):
with self.lock:
self.seq += 1
self.queue[(source, session_id)] = self.seq
self.wake.set()
def hub_down(self) -> bool:
return max(self.failures, self.device_down) >= HUB_DOWN_FAILURES
def status(self) -> dict:
return {"hub": self.hub, "up": not self.hub_down(), "takeover": self.takeover,
"rendering": self.rendering, "queued": len(self.queue),
"failures": self.failures,
"last_ok_age_s": round(time.time() - self.last_ok, 1),
"hub_instance": self.hub_instance, "hub_style": self.hub_info.get("style"),
"hub_device_ok": self.hub_info.get("device_ok"),
"last_error": self.last_error, "rejected": dict(self.rejects),
"legacy_hub": self.legacy_hub, "disabled": self.disabled}
def _ok(self, path: str):
self.failures, self.last_ok, self.last_error = 0, time.time(), ""
self._no_net = False
self.rejects.pop(path, None)
def _request(self, method: str, path: str, body: bytes | None = None):
"""-> (status, json). status 0 = no answer at all (network)."""
headers = dict(self.headers)
if body is not None:
headers["Content-Type"] = "application/json"
req = urllib.request.Request(self.hub + path, data=body, method=method,
headers=headers)
try:
with OPENER.open(req, timeout=HUB_TIMEOUT_S) as r:
raw = r.read()
self._ok(path)
try:
return 200, (json.loads(raw) if raw else {})
except json.JSONDecodeError:
return 200, {}
except urllib.error.HTTPError as e:
# The hub answered: it is up even when it rejects us. Retrying a
# 4xx would not help, so the caller treats it as consumed. Said
# once per path, until that path succeeds again.
self.failures, self.last_ok, self._no_net = 0, time.time(), False
err = f"hub answered {e.code} on {path}"
if e.code == 401:
err += " - BUSYBAR_HUB_TOKEN differs from the hub's"
elif e.code == 404:
err += " - the hub runs an older daemon.py (update and restart it)"
if self.rejects.get(path) != err:
log(err)
self.rejects[path] = err
return e.code, {}
except OSError as e:
self.last_error = f"{type(e).__name__}: {e}"[:120]
return 0, {}
def _payload(self, source: str, session_id: str):
key = f"{source}:{session_id}"
now = time.time()
with STORE.lock:
s = STORE.sessions.get(key)
if s is None:
return ({"source": source, "session_id": session_id, "ended": True,
"rev": STORE.ended_rev.get(key, int(now * 1000))}, None)
p = {"source": source, "session_id": session_id, "rev": s["rev"],
"lease_s": MIRROR_LEASE_S,
"focus_age_s": max(0.0, now - s["focus_ts"]) if s["focus_ts"] else None,
"state_age_s": max(0.0, now - s["state_ts"]),
"active_age_s": max(0.0, now - s["last_active"]),
**{k: s[k] for k in MIRROR_FIELDS}}
if s["state"] != self.sent_state.get(key):
p["state"] = s["state"]
return p, s["state"]
def flush(self) -> bool:
"""Deliver queued records in order; False at the first network
failure (the record stays queued and is retried later)."""
while True:
with self.lock:
if not self.queue:
return True
key, seq = next(iter(self.queue.items()))
payload, state = self._payload(*key)
code, reply = self._request("POST", "/v1/report", json.dumps(payload).encode())
if code == 0 or code >= 500:
return False
skey = f"{key[0]}:{key[1]}"
if code < 300:
if state is None:
self.sent_state.pop(skey, None)
else:
self.sent_state[skey] = state
with self.lock:
if self.queue.get(key) == seq: # nothing newer arrived meanwhile
del self.queue[key]
if code < 300 and reply.get("created") and "state" not in payload:
# The hub had forgotten this session (lease, its own sleep):
# it now holds it as IDLE. Send the state again.
self.sent_state.pop(skey, None)
self.enqueue(*key)
def resync(self):
"""Re-mirror everything live, transitions included."""
self.sent_state.clear()
for source, sid in STORE.live_keys():
self.enqueue(source, sid)
def probe(self) -> bool:
self.last_probe = time.time()
info = {}
if self.legacy_hub:
code = self._request("GET", "/health")[0]
else:
code, info = self._request("GET", "/hub")
if code == 404: # older hub: alive, tells us nothing more
self.legacy_hub = True
code = self._request("GET", "/health")[0]
if code == 0:
return False
if info:
self._learn(info)
return True
def _learn(self, info: dict):
self.hub_info = info
inst = info.get("instance")
if inst == INSTANCE:
self.disabled = "BUSYBAR_HUB points at this very daemon"
elif info.get("role") == "standby":
self.disabled = "the hub is itself a standby: two standbys never paint anything"
if self.disabled:
log(f"standby link disabled: {self.disabled}")
return
if self.hub_instance is not None and inst != self.hub_instance:
log("hub restarted: resyncing every session")
self.resync()
self.hub_instance = inst
# The hub cannot reach its Bar: counts like unreachability, with the
# same debounce (one failed USB draw must not start a Wi-Fi takeover).
self.device_down = self.device_down + 1 if info.get("device_ok") is False else 0
style = info.get("style")
if style and style != STYLE and not self._style_warned:
self._style_warned = True
log(f"hub renders style {style!r}, this standby {STYLE!r}: "
f"set BUSYBAR_STYLE the same on both computers")
def _failed(self):
"""A probe or delivery got no answer. Count it only if the Bar still
answers; otherwise our own network is gone and we could not draw."""
if self.transport.reachable():
self.failures += 1
self._no_net = False
else:
self.failures = 0
if not self._no_net:
log("neither the hub nor the Bar answer: assuming our own network is down")
self._no_net = True
def loop(self, stop: threading.Event):
last_tick = time.time()
while not stop.is_set() and not self.disabled:
self.wake.clear()
now = time.time()
if now - last_tick > SUSPEND_GAP_S:
# We slept. Evidence from before does not count, a takeover
# from before is over (three fresh failures re-arm it), and
# the hub may have forgotten us meanwhile: re-mirror everything.
self.failures = self.device_down = 0
self.takeover = False
self.resync()
last_tick = now
# Once something failed, look again sooner: takeover and the
# hub's return are both decided by consecutive evidence.
poll = HUB_POLL_S if self.failures == 0 and not self.takeover else 1.0
if self.takeover:
if now - self.last_probe >= poll and self.probe() and not self.hub_down():
with RENDER_LOCK: # waits for any draw of ours in flight
self.resync()
if self.flush():
self._request("POST", "/redraw", b"{}")
self.takeover = False
log("hub is back: resynced, yielding the display")
STORE.dirty.set()
else:
if self.queue:
ok = self.flush() # deliveries double as probes
elif now - self.last_probe >= poll:
ok = self.probe()
else:
ok = True
if not ok:
self._failed()
elif now - self.last_heartbeat > MIRROR_HEARTBEAT_S:
self.last_heartbeat = now
for source, sid in STORE.live_keys():
self.enqueue(source, sid) # lease refresh
if self.hub_down():
self.takeover = True
log("hub unreachable: taking over the display" if self.failures
else "hub cannot reach the Bar: taking over the display")
REDRAW.set() # wipe the hub's stale frame before our first one
STORE.dirty.set()
self.wake.wait(timeout=poll)
HUBLINK: HubLink | None = None
TRANSPORT: HttpTransport | None = None
ROLE = "standby" if STANDBY else "hub" if "0.0.0.0" in LISTEN_ADDRS else "local"
def effective_state(sess: dict) -> str:
state = sess["state"]
if state == "COMPLETE" and time.time() - sess["state_ts"] >= COMPLETE_HOLD_S:
return "IDLE"
return state
def status_snapshot() -> dict:
"""Normalized active view: what renderers (and GET /status) consume."""
sess = STORE.active_session()
if sess is None:
return {"source": None, "state": "IDLE", "label": None,
"label_color": None, "context_pct": None, "quotas": None,
"age_s": None}
now = time.time()
quotas = []
for q in sess.get("quotas") or []:
q = dict(q)
if quota_expired(q, now):
q["left_pct"] = None # a passed reset is not a fresh account snapshot
quotas.append(q)
quota_status = dict(sess.get("quota_status") or {})
if quota_status:
if not any(q.get("left_pct") is not None for q in quotas):
quota_status["state"] = "unavailable"
observed_at = quota_status.get("observed_at")
if finite_number(observed_at):
quota_status["age_s"] = max(0, round(now - observed_at, 1))
label = sess.get("label")
badges = sess.get("badges")
if EFFORT_CONTROLLER:
control = EFFORT_CONTROLLER.status(display_thread_id=sess.get("control_thread_id"))
settings = (control if control.get("connected")
and control.get("thread_id") == sess.get("control_thread_id")
else control.get("display", {})) if sess.get("control_thread_id") else {}
if settings.get("model") and settings.get("effort"):
from adapters.codex_status import prettify_model
label = shorten_model_label(prettify_model(settings["model"]),
settings["effort"], LABEL_MAX_PX)
if settings.get("fast") is not None:
badges = [badge for badge in badges or [] if badge not in ("fast", "priority")]
badges += ["fast"] if settings["fast"] else []
return {
"source": sess["source"],
"state": effective_state(sess),
"label": label,
"label_color": sess.get("label_color"),
"context_pct": sess.get("context_pct"),
"quotas": quotas or None,
"quota_status": quota_status or None,
"week_progress_pct": week_progress_pct(weekly_quota(quotas), now),
"badges": badges,
"host": sess.get("host"),
"host_tag": sess.get("host_tag"),
"age_s": round(now - sess["last_active"], 1),
}
# --------------------------------------------------------------------------
# Claude Code adapter: statusline JSON + hook states -> normalized reports.
# All Claude-specific semantics live here.
# --------------------------------------------------------------------------
# Straight from the Claude Code theme palette:
# inactive / permission / warning / fastMode / effortUltra
CLAUDE_EFFORT_COLORS = {
"low": "#999999FF", "medium": "#99CCFFFF", "high": "#FFC107FF",
"xhigh": "#FF7814FF", "max": "#AF87FFFF",
}
def shorten_model_label(name: str, effort: str, max_px: int) -> str:
"""Fit "<model name> <effort>" into max_px: keep the effort word, chip
letters off the longest alphabetic word of the name first ("Fable 5"
-> "Fabl 5"), so version numbers survive as long as possible."""
words = name.split()
def label() -> str:
return f"{' '.join(words)} {effort}".strip()
while words and est_width(label()) > max_px:
# chip the longest alphabetic word down to 3 letters; then drop
# trailing words (versions, suffixes); then chip down to 1 letter
for floor in (3, 1):
alpha = [i for i, w in enumerate(words) if len(w) > floor and w[-1].isalpha()]
if alpha:
i = max(alpha, key=lambda i: len(words[i]))
words[i] = words[i][:-1]
break
if floor == 3 and len(words) > 1:
words.pop()
break
else:
words.pop()
return label()
def claude_statusline_report(data: dict, reserve_px: int = 0) -> dict:
model = (data.get("model") or {})
name = model.get("display_name") or model.get("id") or ""
effort = (data.get("effort") or {}).get("level") or ""
label = shorten_model_label(name, effort, LABEL_MAX_PX - reserve_px)
ctx = data.get("context_window") or {}
context_pct = ctx.get("used_percentage")
if context_pct is None and ctx.get("remaining_percentage") is not None:
context_pct = 100 - ctx["remaining_percentage"]
quotas = []
rl = data.get("rate_limits") or {}
for qname, key in (("5h", "five_hour"), ("7d", "seven_day")):
w = rl.get(key) or {}
if w.get("used_percentage") is not None:
quotas.append({
"name": qname,
"left_pct": max(0, round(100 - w["used_percentage"])),
"resets_at": w.get("resets_at"),
})
fields = {"context_pct": context_pct, "quotas": quotas or None}
if label:
fields["label"] = label
fields["label_color"] = CLAUDE_EFFORT_COLORS.get(effort, LABEL_FALLBACK_COLOR)
return fields
# --------------------------------------------------------------------------
# Renderer (normalized fields only)
# --------------------------------------------------------------------------
# The device's small font is proportional (measured on-device); errs wide.
_NARROW = set("iljI.,;:' ")
def est_width(text: str) -> int:
w = 0
for ch in text:
if ch in _NARROW:
w += 3
elif ch.isdigit():
w += 4
elif ch.isupper() or ch in "MWmw%":
w += 5
else:
w += 4
return w
def host_tag_kind(tag) -> tuple[str, str]:
"""Protocol `host_tag`: "#RRGGBB[AA]" -> ("flag", color): a 2x5 flag in
the free columns left of the label (costs no text space); up to two
printable ASCII chars -> ("text", chars): drawn after the label (the
model name is shortened to make room); anything else -> ("", "")."""
if not isinstance(tag, str):
return "", ""
tag = "".join(ch for ch in tag.strip() if 0x21 <= ord(ch) <= 0x7E)
if re.fullmatch(r"#[0-9A-Fa-f]{6}(?:[0-9A-Fa-f]{2})?", tag):
return "flag", _norm_color(tag, HOST_TAG_COLOR)
return ("text", tag[:2]) if tag else ("", "")
def bar_color(used: float) -> str:
if used >= 90:
return "#FF2020FF"
if used >= 80:
return "#FF6A00FF"
if used >= 50:
return "#FFB000FF"
return "#20C040FF"
def quota_bar_color(left: float) -> str:
if left <= 10:
return "#FF2020FF"
if left <= 25:
return "#FF6A00FF"
if left <= 50:
return "#FFB000FF"
return "#20C040FF"
def finite_number(value):
return (isinstance(value, (int, float)) and not isinstance(value, bool)
and math.isfinite(value))
def parse_quotas(value):
quotas = []
for q in value[:4] if isinstance(value, list) else []:
if not isinstance(q, dict):
continue
left = q.get("left_pct")
item = {"name": str(q.get("name", ""))[:6],
"left_pct": max(0., min(100., left)) if finite_number(left) else None}
for field in ("resets_at", "window_minutes", "observed_at", "valid_until"):
number = q.get(field)
if finite_number(number) and number > 0:
item[field] = number
quotas.append(item)
return quotas or None
def quota_expired(quota, now):
return any(finite_number(quota.get(key)) and quota[key] <= now
for key in ("resets_at", "valid_until"))
def weekly_quota(quotas):
for quota in quotas or []:
minutes = quota.get("window_minutes")
if minutes == WEEK_SECONDS / 60 or (minutes is None and
str(quota.get("name", "")).lower() in ("7d", "week", "weekly", "wk")):
return quota
return None # a lone 5h window must never be rendered as a week
def week_progress_pct(quota: dict | None, now: float | None = None) -> float | None:
"""How far the current seven-day window has advanced toward its reset."""
if not quota or not finite_number(quota.get("resets_at")):
return None
now = time.time() if now is None else now
if quota_expired(quota, now) or ("left_pct" in quota and quota["left_pct"] is None):
return None
minutes = quota.get("window_minutes", WEEK_SECONDS / 60)
if minutes != WEEK_SECONDS / 60:
return None
seconds_left = quota["resets_at"] - now
return max(0., min(100., 100. * (1. - seconds_left / (minutes * 60))))
def astra_availability(path: str | None = None, now: float | None = None) -> dict:
"""Read the small, non-secret state written by the Astra Watch plugin."""
path = path or ASTRA_STATE_PATH
now = time.time() if now is None else now
try:
with open(path, encoding="utf-8") as stream:
payload = json.load(stream)
modified_at = os.path.getmtime(path)
except (OSError, json.JSONDecodeError):
state = "unknown"
payload = {}
modified_at = None
else:
state = payload.get("last_state")
if state not in ("waiting", "hidden", "available", "error"):
state = "unknown"
elif ASTRA_STALE_S > 0 and now - modified_at > ASTRA_STALE_S:
state = "stale"
age_s = max(0.0, now - modified_at) if modified_at is not None else None
return {
"target": payload.get("target", "gpt-6-astra"),
"state": state,
"color": ASTRA_COLORS[state],
"last_checked_at": payload.get("last_checked_at"),
"catalog_fetched_at": payload.get("catalog_fetched_at"),
"client_version": payload.get("client_version"),
"model_count": payload.get("model_count"),
"visibility": payload.get("visibility"),
"age_s": round(age_s, 1) if age_s is not None else None,
"next_check_s": (max(0, round(ASTRA_CHECK_INTERVAL_S - age_s))
if age_s is not None else None),
"check_interval_s": round(ASTRA_CHECK_INTERVAL_S),
"stale": state == "stale",
}
def request_astra_refresh(now: float | None = None) -> tuple[bool, str]:
"""Start a non-inference catalog refresh, with a short debounce."""
global ASTRA_REFRESHED_AT
now = time.time() if now is None else now
with ASTRA_REFRESH_LOCK:
if now - ASTRA_REFRESHED_AT < 15:
return True, "already refreshing"
if not os.path.isfile(ASTRA_WATCH_SCRIPT):
return False, "Astra Watch script is not installed"
try:
subprocess.Popen(
[sys.executable, ASTRA_WATCH_SCRIPT, "check", "--json"],
stdin=subprocess.DEVNULL,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
close_fds=True,
)
except OSError as exc:
return False, f"could not start Astra Watch: {exc}"
ASTRA_REFRESHED_AT = now
return True, "refresh started"
def x_pulse_status(now: float | None = None) -> dict:
if X_PULSE_MONITOR is None:
return {
"enabled": False, "state": "disabled", "official": False,
"label": "X community pulse",
}
return X_PULSE_MONITOR.status(now)
def request_x_pulse_refresh(*, force: bool = False, now: float | None = None) -> bool:
if X_PULSE_MONITOR is None:
return False
return X_PULSE_MONITOR.request_refresh(force=force, now=now)
def astra_watch_status() -> dict:
status = astra_availability()
status["x_pulse"] = x_pulse_status()
return status
def mark_astra_app_seen(now: float | None = None):
"""Record a status poll from the on-device app for button routing/debug."""
global ASTRA_APP_LAST_SEEN, ASTRA_APP_REQUESTS
with ASTRA_APP_LOCK:
ASTRA_APP_LAST_SEEN = time.monotonic() if now is None else now
ASTRA_APP_REQUESTS += 1
STORE.dirty.set()
def astra_app_status(now: float | None = None) -> dict:
now = time.monotonic() if now is None else now
with ASTRA_APP_LOCK:
age = now - ASTRA_APP_LAST_SEEN if ASTRA_APP_LAST_SEEN else None
requests = ASTRA_APP_REQUESTS
return {
"active": age is not None and age <= ASTRA_APP_ACTIVE_S,
"last_seen_age_s": round(max(0.0, age), 1) if age is not None else None,
"requests": requests,
}
def handle_astra_app_poll(now: float | None = None):
"""Refresh once when the on-device app opens, then serve cheap polls."""
was_active = astra_app_status(now)["active"]
mark_astra_app_seen(now)
if not was_active:
request_astra_refresh()
request_x_pulse_refresh()
def _norm_color(c, fallback: str) -> str:
if not isinstance(c, str) or not c.startswith("#") or len(c) not in (7, 9):
return fallback
return (c + "FF") if len(c) == 7 else c
def _rect(eid, x, y, w, h, color):
return {"id": eid, "type": "rectangle", "display": "front",
"x": x, "y": y, "width": w, "height": h,
"border_width": 0, # undocumented; defaults to 1px white border
"fill": "solid", "fill_colors": [color], "timeout": TEXT_TIMEOUT_S}
def _text(eid, x, y, align, text, color):
return {"id": eid, "type": "text", "display": "front",
"x": x, "y": y, "align": align,
"text": text, "font": FONT, "color": color, "timeout": TEXT_TIMEOUT_S}
def anim_element(state: str, badges=None, astra_state: str | None = None) -> dict:
path = STATE_ANIMS.get(state, "idle.anim")
if astra_state == "available":
path = "astra.anim"
elif state == "WORKING" and "fast" in (badges or []):
path = "work_fast.anim"
return {"id": "ring", "type": "animation", "display": "front",
"x": 0, "y": 0, "path": path,
"loop": True, "timeout": ANIM_TIMEOUT_S}
def avatar_element(state: str) -> dict:
return {"id": "avatar", "type": "animation", "display": "front",
"x": AVATAR_X, "y": AVATAR_Y,
"path": AVATAR_ANIMS.get(state, "claw_idle.anim"),
"loop": True, "timeout": ANIM_TIMEOUT_S}
def device_canvas_allowed() -> bool:
with DEVICE_INPUT_LOCK:
mode_allowed = DEVICE_MODE not in ("APPS", "SETTINGS")
return mode_allowed and not astra_app_status()["active"]
def handle_device_input_event(event: tuple) -> bool:
"""Route START to Fast, the dial to effort, and OK to Astra Watch."""
global DEVICE_MODE, LAST_ENCODER, LAST_START, START_DOWN
if not event:
return False
if event[0] == "encoder":
handled = False
reason = ''
if EFFORT_CONTROLLER and effort_input_allowed():
handled = EFFORT_CONTROLLER.rotate(event[1])
if not handled:
reason = EFFORT_CONTROLLER.status().get('error') or 'Codex settings connection is not ready'
else:
reason = effort_input_block_reason() if EFFORT_CONTROLLER else 'Effort controller is disabled'
with DEVICE_INPUT_LOCK:
LAST_ENCODER = {'at': time.time(), 'delta': event[1], 'handled': bool(handled), 'reason': reason}
if not handled:
log(f'Codex dial ignored: {reason}')
return handled
if event[0] == "button":
if len(event) >= 3 and event[1] == 2: # START; PRESS=0, RELEASE=1
with DEVICE_INPUT_LOCK:
if event[2] == 1:
START_DOWN = False
return False
if event[2] != 0 or START_DOWN:
return False
START_DOWN = True
handled = False
reason = ''
if EFFORT_CONTROLLER and effort_input_allowed():
handled = EFFORT_CONTROLLER.toggle_fast()
if not handled:
reason = EFFORT_CONTROLLER.status().get('error') or 'Codex settings connection is not ready'
else:
reason = effort_input_block_reason() if EFFORT_CONTROLLER else 'Codex controls are disabled'
with DEVICE_INPUT_LOCK:
LAST_START = {'at': time.time(), 'handled': bool(handled), 'reason': reason}
if not handled:
log(f'Codex START ignored: {reason}')
return handled
if event[1:3] == (0, 0) and astra_app_status()["active"]:
request_astra_refresh()
request_x_pulse_refresh(force=True)
return True
return False
if event[0] != "switch":
return False
mode = {0: "BUSY", 1: "CUSTOM", 2: "OFF", 3: "APPS", 4: "SETTINGS"}.get(
event[1], "UNKNOWN",
)
with DEVICE_INPUT_LOCK:
changed = mode != DEVICE_MODE
DEVICE_MODE = mode
if changed:
STORE.dirty.set()
return changed
def device_input_loop(input_url: str, stop: threading.Event,
source_address: str | None = None):
"""Observe hardware events through the shared buffered input stream."""
def update_state(connected, error):
global DEVICE_INPUT_CONNECTED, DEVICE_INPUT_ERROR, START_DOWN
with DEVICE_INPUT_LOCK:
DEVICE_INPUT_CONNECTED = connected
DEVICE_INPUT_ERROR = error
if not connected:
START_DOWN = False
InputStream(input_url, handle_device_input_event,
source_address=source_address, on_state=update_state,
logger=log, label="BUSY mode observer").run(stop)
def info_elements(status: dict, astra: dict | None = None) -> list[dict]:
"""Text rows + weekly gauges for a normalized snapshot (ring/avatar are
separate animation elements)."""
avatar = STYLE == "avatar"
label_max = (AVATAR_X - 4 - 5) if avatar else LABEL_MAX_PX
elements = []
state = status["state"]
kind, tag = host_tag_kind(status.get("host_tag"))
if kind == "text":
label_max -= est_width(tag) + 2 # the tag sits right after the label
label = status.get("label") or ""
label = "".join(ch for ch in label if 0x20 <= ord(ch) <= 0x7E) # ASCII-only font
while label and est_width(label) > label_max:
label = label[:-1]
if label:
elements.append(_text("model", 3, 0, "top_left", label,
_norm_color(status.get("label_color"), LABEL_FALLBACK_COLOR)))
x = 3 + est_width(label)
# Host tag (which computer this session lives on). Both variants are
# always sent - blank/transparent when unused - so a stale tag never
# outlives a switch to another session (element ids can't change type).
elements.append(_text("htag", x + 2, 0, "top_left",
tag if kind == "text" else " ", HOST_TAG_COLOR))
if kind == "text":
x += 2 + est_width(tag)
elements.append(_rect("hflag", 1, 2, 2, 5, tag if kind == "flag" else "#00000000"))
quotas = status.get("quotas") or []
weekly = weekly_quota(quotas)
week_progress = week_progress_pct(weekly)
if avatar:
# Vertical week-progress gauge between the text column and avatar.
elements.append(_rect("ctrack", AVATAR_X - 4, 1, 2, 14, BAR_TRACK_COLOR))
if week_progress is not None and week_progress > 0:
fh = max(1, min(14, round(14 * week_progress / 100)))
elements.append(_rect("cfill", AVATAR_X - 4, 15 - fh, 2, fh,
bar_color(week_progress)))
else:
elements.append(_rect("cfill", AVATAR_X - 4, 1, 2, 14, "#00000000"))
else:
elements.append(_rect("ctrack", BAR_X, BAR_Y, BAR_W, BAR_H, BAR_TRACK_COLOR))
if week_progress is not None and week_progress > 0:
fill = max(1, min(BAR_W, round(BAR_W * week_progress / 100)))
elements.append(_rect("cfill", BAR_X, BAR_Y, fill, BAR_H,
bar_color(week_progress)))
else:
elements.append(_rect("cfill", BAR_X, BAR_Y, BAR_W, BAR_H, "#00000000"))
if weekly and weekly.get("left_pct") is not None and not quota_expired(weekly, time.time()):
left = max(0.0, min(100.0, float(weekly["left_pct"])))
color = quota_bar_color(left)
fill = max(1, min(QUOTA_BAR_W, round(QUOTA_BAR_W * left / 100)))
elements.append(_text("usage", 3, 15, "bottom_left", "W", color))
elements.append(_rect("qtrack", QUOTA_BAR_X, QUOTA_BAR_Y,
QUOTA_BAR_W, QUOTA_BAR_H, BAR_TRACK_COLOR))
elements.append(_rect("qfill", QUOTA_BAR_X, QUOTA_BAR_Y,
fill, QUOTA_BAR_H,
color if left > 0 else "#00000000"))
else:
# Keep element ids and types stable while clearing a previous quota.
unknown = bool(quotas or status.get("quota_status"))
elements.append(_text("usage", 3, 15, "bottom_left", "?" if unknown else " ", QUOTA_COLOR))
elements.append(_rect("qtrack", QUOTA_BAR_X, QUOTA_BAR_Y,
QUOTA_BAR_W, QUOTA_BAR_H, BAR_TRACK_COLOR if unknown else "#00000000"))
elements.append(_rect("qfill", QUOTA_BAR_X, QUOTA_BAR_Y,
1, QUOTA_BAR_H, "#00000000"))
# Tiny 3x5 pixel "A" between the quota gauge and state word. The plugin
# state maps to: waiting gray, server-visible-but-hidden amber, available
# green, and stale/error red. It is transparent when Astra Watch is absent.
astra_color = (astra or astra_availability())["color"]
elements.extend((
_rect("astra_top", 47, 10, 1, 1, astra_color),
_rect("astra_left", 46, 11, 1, 4, astra_color),
_rect("astra_right", 48, 11, 1, 4, astra_color),
_rect("astra_cross", 46, 12, 3, 1, astra_color),
))
if not avatar:
elements.append(_text("state", 69, 15, "bottom_right",
STATE_WORDS.get(state, state[:5]),
STATE_COLORS.get(state, STATE_COLORS["IDLE"])))
return elements
THEME_ACTIVE = threading.Event()
def snapshot_watch_loop(transport: HttpTransport, stop: threading.Event):
"""Poll the device's BUSY snapshot: the currently selected theme acts
as the on-device manual switch for the status display."""
while not stop.is_set():
active = False
snap = transport.get_json("/busy/snapshot")
if snap:
s = snap.get("snapshot") or {}
active = (s.get("busy_bar_settings") or {}).get("theme") == THEME_NAME
if active != THEME_ACTIVE.is_set():
log(f"claude theme {'selected' if active else 'deselected'} on device")
(THEME_ACTIVE.set if active else THEME_ACTIVE.clear)()
STORE.dirty.set()
stop.wait(SNAPSHOT_POLL_S)
def effort_feedback_elements(feedback):
return [{"id": "effort", "type": "text", "display": "front",
"x": 36, "y": 8, "align": "center", "text": feedback,
"font": "large", "color": "#EF5555FF" if feedback == "ERR"
else "#60BFFFFF", "timeout": TEXT_TIMEOUT_S}]
def effort_overlay_elements(feedback, direction=1, entering=True):
error = feedback == "ERR"
if feedback in ('FAST', 'NORMAL'):
path = fast_animation.filename(feedback == 'FAST', entering)
else:
path = (effort_animation.filename(feedback.lower(), direction, entering)
if feedback and not error else "effort_clear.anim")
return [
{"id": "effort_transition", "type": "animation", "display": "front",
"x": 0, "y": 0, "path": path, "loop": False, "timeout": 4, "z_index": 100},
{**_rect("effort_error_bg", 0, 0, 72, 16, "#000000FF" if error else "#00000000"),
"z_index": 101, "timeout": 4},
{**effort_feedback_elements("ERR" if error else " ")[0],
"id": "effort_error_text", "z_index": 102, "timeout": 4},
]
def render_loop(transport: HttpTransport, stop: threading.Event):
scene = DrawCache(APP_NAME, DRAW_PRIORITY)
last_tick = time.time()
ai_overlay_was_active = False
last_overlay_key = None
overlay_drawn_at = 0.0
while not stop.is_set():
STORE.dirty.clear()
now = time.time()
if now - last_tick > SUSPEND_GAP_S:
# This computer slept. Mirrors we hold are stale (their standby
# resyncs within seconds); the canvas is repainted from scratch.
log(f"resumed after {now - last_tick:.0f}s: repainting")
STORE.drop_mirrored()
REDRAW.set()
last_tick = now
# Do not queue lower-priority USB draw calls underneath the provider
# overlay. The firmware rejects or defers them, holding the shared
# render lock long enough to make a four-second carousel look stuck.
# Once the overlay clears, repaint the normal canvas from scratch.
ai_overlay_active = AI_MONITOR is not None and AI_MONITOR.drawn
if ai_overlay_active:
ai_overlay_was_active = True
STORE.dirty.wait(timeout=0.25)
continue
if ai_overlay_was_active:
scene.reset()
last_overlay_key = None
ai_overlay_was_active = False
with RENDER_LOCK:
if stop.is_set():
break # shutdown may have cleared the canvas while we waited
sess = STORE.active_session()
want = False
if sess is not None:
idle_expired = (
IDLE_CLEAR_AFTER_S > 0
and effective_state(sess) == "IDLE"
and now - max(sess["state_ts"], sess["last_active"]) > IDLE_CLEAR_AFTER_S
)
gate_ok = THEME_ACTIVE.is_set() if RENDER_MODE == "theme" else True
gate_ok = gate_ok and device_canvas_allowed()
if HUBLINK is not None:
gate_ok = gate_ok and HUBLINK.takeover # standby: only while the hub is out
want = gate_ok and not idle_expired
force = REDRAW.is_set()
if force:
REDRAW.clear()
scene.reset()
last_overlay_key = None
if HUBLINK is not None:
HUBLINK.rendering = want
if not want:
if DRAWN.is_set() or force:
# A standby yielding leaves the canvas to the hub, which
# repaints it (POST /redraw); everyone else wipes their own.
if HUBLINK is None or HUBLINK.takeover:
transport.clear(APP_NAME)
DRAWN.clear()
scene.reset()
last_overlay_key = None
else:
if force:
# Leftovers first: a sleeping hub's frame, another style's ids.
transport.clear(APP_NAME)
status = status_snapshot()
control = EFFORT_CONTROLLER.status() if EFFORT_CONTROLLER else {}
feedback = (control.get("feedback")
if control.get("thread_id") == sess.get("control_thread_id") else None)
# Detent feedback goes out before quota text/ring refreshes.
# Those can each occupy a USB round trip, even while hidden.
overlay_key = ((control.get("thread_id"), feedback, control.get("feedback_revision"))
if feedback else None)
if overlay_key != last_overlay_key:
elapsed = time.monotonic() - overlay_drawn_at
finishing = (not feedback and last_overlay_key and last_overlay_key[1] != "ERR"
and last_overlay_key[0] == control.get("thread_id")
and elapsed < effort_animation.DURATION_S)
if not finishing:
entering = not (last_overlay_key and last_overlay_key[1] != "ERR"
and last_overlay_key[0] == control.get("thread_id")
and elapsed < (effort_animation.FRAMES - effort_animation.FADE_FRAMES)
/ effort_animation.FPS)
elements = effort_overlay_elements(feedback, control.get("direction", 1), entering)
if transport.draw({"application_name": APP_NAME, "priority": DRAW_PRIORITY,
"elements": elements}):
DRAWN.set()
last_overlay_key = overlay_key
overlay_drawn_at = time.monotonic()
if feedback and feedback != "ERR":
EFFORT_CONTROLLER.mark_drawn(control.get("feedback_revision"))
# A detent may arrive during a draw. Return to the newest
# confirmed overlay before sending any background refresh.
if stop.is_set() or STORE.dirty.is_set():
continue
astra = astra_availability()
anims = [anim_element(status["state"], status.get("badges"), astra["state"])]
if STYLE == "avatar":
anims.append(avatar_element(status["state"]))
anims = [{**element, "z_index": 0} for element in anims]
texts = [{**element, "z_index": 2 if element["id"] in ("cfill", "qfill") else 1}
for element in info_elements(status, astra)]
# One request updates all changed dashboard groups. Each group
# keeps its own keepalive, and failed draws never enter the cache.
draw_now = time.monotonic()
if scene.draw(transport,
scene.pending("animations", anims, draw_now, ANIM_REFRESH_S),
scene.pending("information", texts, draw_now, KEEPALIVE_S)):
DRAWN.set()
STORE.dirty.wait(timeout=0.5)
# --------------------------------------------------------------------------
# Report/status server
# --------------------------------------------------------------------------
class Handler(BaseHTTPRequestHandler):
def log_message(self, *args):
pass
def _reply(self, code: int, body: bytes = b"{}"):
self.send_response(code)
self.send_header("Content-Type", "application/json")
self.send_header("Content-Length", str(len(body)))
self.end_headers()
self.wfile.write(body)
def _authorized(self) -> bool:
if not HUB_TOKEN or self.client_address[0] in ("127.0.0.1", "::1"):
return True
return hmac.compare_digest(self.headers.get("X-Busybar-Token", ""), HUB_TOKEN)
def _host_fields(self) -> dict:
"""Which computer a report comes from (report.py / report.sh headers)."""
fields = {}
host = (self.headers.get("X-Busybar-Host") or "").strip()[:64]
if host:
fields["host"] = host
tag = (self.headers.get("X-Busybar-Host-Tag") or "").strip()[:9]
if tag:
fields["host_tag"] = tag
return fields
def do_GET(self):
if self.path in ("/status", "/v1/status"):
self._reply(200, json.dumps(status_snapshot()).encode())
elif self.path == "/astra":
if self.client_address[0] == "10.0.4.20":
handle_astra_app_poll()
self._reply(200, json.dumps(astra_watch_status()).encode())
elif self.path == "/hub":
self._reply(200, json.dumps({
"ok": True, "instance": INSTANCE, "role": ROLE, "style": STYLE,
"render_mode": RENDER_MODE,
"device_mode": DEVICE_MODE,
"device_input": {
"connected": DEVICE_INPUT_CONNECTED,
"error": DEVICE_INPUT_ERROR,
"last_encoder": LAST_ENCODER or None,
"last_start": LAST_START or None,
},
"codex_effort": (EFFORT_CONTROLLER.status() if EFFORT_CONTROLLER
else {"enabled": False}),
"codex_focus": codex_focus.FOCUS.status(),
"codex_target": codex_target.FOCUS.status(),
"device_ok": TRANSPORT.device_ok if TRANSPORT else None,
"device_error": TRANSPORT.last_error if TRANSPORT else "",
"rendering": DRAWN.is_set(),
"astra": astra_availability(),
"x_pulse": x_pulse_status(),
"astra_app": astra_app_status(),
"ai_status": (AI_MONITOR.status() if AI_MONITOR else
{"enabled": False}),
}).encode())
elif self.path == "/standby":
if HUBLINK is None:
self._reply(404, b'{"error":"not a standby daemon"}')
else:
self._reply(200, json.dumps(HUBLINK.status()).encode())
elif self.path == "/health":
with STORE.lock:
snapshot = {
key: {k: s[k] for k in ("source", "state", "state_ts",
"last_active", "focus_ts", "host", "host_tag")}
for key, s in STORE.sessions.items()
}
self._reply(200, json.dumps(snapshot).encode())
else:
self._reply(404)
def do_POST(self):
parsed = urllib.parse.urlparse(self.path)
length = int(self.headers.get("Content-Length") or 0)
raw = self.rfile.read(length) if length else b""
try:
data = json.loads(raw) if raw else {}
except json.JSONDecodeError:
data = {}
if not self._authorized():
self._reply(401, b'{"error":"missing or wrong X-Busybar-Token"}')
return
host_fields = self._host_fields()
if parsed.path == "/astra/refresh":
# The on-device Astra app lives at 10.0.4.20. This action only
# refreshes the personalized model catalog; it never runs a model.
if self.client_address[0] not in ("127.0.0.1", "::1", "10.0.4.20"):
self._reply(403, b'{"error":"local device only"}')
return
started, message = request_astra_refresh()
self._reply(202 if started else 503, json.dumps({
"ok": started, "message": message,
}).encode())
elif parsed.path == "/v1/report":
# The provider-agnostic reporting endpoint. See docs/EXTENDING.md.
source = str(data.get("source") or "")[:32]
session_id = str(data.get("session_id") or "")[:128]
if not source or not session_id:
self._reply(400, b'{"error":"source and session_id are required"}')
return
fields: dict = {}
if data.get("ended"):
fields["ended"] = True
state = data.get("state")
if state is not None:
if state not in STATES:
self._reply(400, json.dumps(
{"error": f"state must be one of {list(STATES)}"}).encode())
return
fields["state"] = state
if "label" in data:
fields["label"] = str(data["label"] or "")[:64] or None
if "label_color" in data:
fields["label_color"] = str(data["label_color"] or "") or None
if "context_pct" in data:
v = data["context_pct"]
try:
fields["context_pct"] = (max(0.0, min(100.0, float(v)))
if v is not None else None)
except (TypeError, ValueError, OverflowError):
pass
if "quotas" in data:
fields["quotas"] = parse_quotas(data["quotas"])
if "quota_status" in data:
qs = data["quota_status"]
fields["quota_status"] = ({
**{key: str(qs.get(key) or "")[:180]
for key in ("source", "limit_id", "state", "error")},
"observed_at": qs.get("observed_at") if finite_number(qs.get("observed_at")) else None,
} if isinstance(qs, dict) else None)
if "badges" in data:
bs = data["badges"] or []
fields["badges"] = [str(b)[:12] for b in bs[:4]] or None
if "ttl_s" in data and data["ttl_s"]:
try:
fields["ttl_s"] = max(10.0, min(30 * 86400.0, float(data["ttl_s"])))
except (TypeError, ValueError, OverflowError):
pass
# Mirror fields (standby daemons only, header X-Busybar-Mirror).
# Ages, never timestamps: the other computer's clock may be
# seconds off, the LAN is not.
mirrored = self.headers.get("X-Busybar-Mirror") == "1"
if (source == "codex" and self.client_address[0] in ("127.0.0.1", "::1")
and not mirrored):
thread_id = data.get("control_thread_id")
fields["control_thread_id"] = (thread_id if isinstance(thread_id, str)
and re.fullmatch(r"[0-9a-f]{8}(?:-[0-9a-f]{4}){3}-[0-9a-f]{12}", thread_id)
else None)
now = time.time()
if mirrored:
for age_key, ts_key in (("focus_age_s", "focus_ts"),
("state_age_s", "state_ts"),
("active_age_s", "last_active")):
if data.get(age_key) is not None:
try:
fields[ts_key] = now - max(0.0, float(data[age_key]))
except (TypeError, ValueError, OverflowError):
pass
if data.get("rev") is not None:
try:
fields["rev"] = int(data["rev"])
except (TypeError, ValueError, OverflowError):
pass
if data.get("lease_s"):
try:
fields["lease_s"] = max(10.0, min(10 * MIRROR_LEASE_S,
float(data["lease_s"])))
except (TypeError, ValueError, OverflowError):
pass
fields.update(host_fields)
if data.get("host"):
fields["host"] = str(data["host"])[:64]
if "host_tag" in data:
fields["host_tag"] = str(data["host_tag"] or "")[:9] or None
created = STORE.report(source, session_id, fields, mirrored=mirrored)
self._reply(200, json.dumps({"ok": True, "created": created}).encode())
elif parsed.path == "/state":
# claude adapter: hook events
sid = data.get("session_id") or "unknown"
state = urllib.parse.parse_qs(parsed.query).get("state", ["WORKING"])[0]
if state == "ENDED":
STORE.report("claude-code", sid, {"ended": True})
elif state in STATES:
STORE.report("claude-code", sid, {"state": state, **host_fields})
self._reply(200)
elif parsed.path == "/shutdown":
# Loopback only. Exit cleanly; the next Claude Code activity
# respawns the daemon with a fresh env.sh (setup_claude.py uses it).
if self.client_address[0] not in ("127.0.0.1", "::1"):
self._reply(403)
return
self._reply(200, b'{"ok":true}')
STOP.set()
elif parsed.path == "/redraw":
# A standby yielded the canvas back: repaint everything, or clear
# it when there is nothing to show (the ids are shared).
REDRAW.set()
STORE.dirty.set()
self._reply(200, b'{"ok":true}')
elif parsed.path == "/statusline":
# claude adapter: statusline payload
sid = data.get("session_id") or "unknown"
kind, tag = host_tag_kind(host_fields.get("host_tag"))
reserve = est_width(tag) + 2 if kind == "text" else 0
STORE.report("claude-code", sid,
{**claude_statusline_report(data, reserve), **host_fields})
self._reply(200)
else:
self._reply(404)
def log(msg: str):
print(f"[{time.strftime('%H:%M:%S')}] {msg}", file=sys.stderr, flush=True)
class _Server(ThreadingHTTPServer):
# On Windows SO_REUSEADDR lets a second process bind the same port,
# which would break the single-instance guarantee.
allow_reuse_address = os.name != "nt"
def handle_error(self, request, client_address):
# Clients cap their wait at about a second; while we were asleep
# they gave up, and on wake their replies hit closed sockets.
if not isinstance(sys.exc_info()[1], (BrokenPipeError, ConnectionResetError)):
super().handle_error(request, client_address)
def serve_on(addr: str) -> ThreadingHTTPServer | None:
try:
server = _Server((addr, LISTEN_PORT), Handler)
except OSError:
return None
threading.Thread(target=server.serve_forever, daemon=True).start()
return server
def bind_loop(addr: str, stop: threading.Event, servers: list):
"""Secondary listeners. The USB interface (10.0.4.21) exists only while
the device is plugged in - keep retrying so the device app can always
reach us."""
while not stop.is_set():
server = serve_on(addr)
if server:
servers.append(server)
log(f"listener up on {addr}:{LISTEN_PORT}")
return
stop.wait(timeout=30)
def main():
global HUBLINK, TRANSPORT, AI_MONITOR, EFFORT_CONTROLLER
stop = STOP
servers = []
TRANSPORT = transport = make_transport()
if STANDBY:
HUBLINK = HubLink(HUB_URL, HUB_TOKEN, transport)
STORE.on_report = HUBLINK.enqueue
primary = serve_on(LISTEN_ADDRS[0])
if primary is None:
return 0 # another instance already owns the port
servers.append(primary)
for addr in LISTEN_ADDRS[1:]:
threading.Thread(target=bind_loop, args=(addr, stop, servers), daemon=True).start()
if STANDBY:
threading.Thread(target=HUBLINK.loop, args=(stop,), daemon=True).start()
def shutdown(*_):
log('shutdown requested' + (f' by signal {_[0]}' if _ else ''))
stop.set()
for s in servers:
threading.Thread(target=s.shutdown, daemon=True).start()
signal.signal(signal.SIGTERM, shutdown)
signal.signal(signal.SIGINT, shutdown)
if RENDER_MODE != "off":
if os.environ.get("BUSYBAR_CODEX_EFFORT", "1").lower() not in ("0", "false", "off"):
EFFORT_CONTROLLER = codex_effort.Controller(effort_target, STORE.dirty.set,
logger=log, allowed=effort_input_allowed,
target_info=effort_target_info)
threading.Thread(target=EFFORT_CONTROLLER.run, args=(stop,), daemon=True).start()
local_input_url = ""
if os.environ.get("BUSYBAR_TRANSPORT", "usb") != "cloud":
local_input_url = input_stream_url(
transport.base,
transport.headers.get("x-api-token", ""),
)
threading.Thread(
target=device_input_loop,
args=(local_input_url, stop,
USB_SOURCE_IP if transport.base.startswith("http://10.0.4.20/")
else None),
daemon=True,
).start()
threading.Thread(target=render_loop, args=(transport, stop), daemon=True).start()
if AI_STATUS_ENABLED:
AI_MONITOR = ai_status.Monitor(
transport,
RENDER_LOCK,
url=AI_STATUS_URL,
x_url=X_STATUS_URL,
google_url=GOOGLE_STATUS_URL,
input_url=local_input_url,
input_source_address=(USB_SOURCE_IP
if transport.base.startswith("http://10.0.4.20/")
else None),
poll_s=AI_STATUS_POLL_S,
should_render=lambda: (
(HUBLINK is None or HUBLINK.takeover)
and device_canvas_allowed()
),
logger=log,
)
threading.Thread(target=AI_MONITOR.run, args=(stop,), daemon=True).start()
if RENDER_MODE == "theme":
threading.Thread(target=snapshot_watch_loop, args=(transport, stop), daemon=True).start()
log(f"listening on {LISTEN_ADDRS[0]}:{LISTEN_PORT}, render_mode={RENDER_MODE}, "
f"style={STYLE}, transport={os.environ.get('BUSYBAR_TRANSPORT', 'usb')}, "
f"ai_status={'on' if AI_STATUS_ENABLED else 'off'}, "
f"x_pulse={X_PULSE_BACKEND if X_PULSE_ENABLED else 'off'}, "
f"x_llm={X_PULSE_LLM_MODEL if X_PULSE_LLM_ENABLED else 'off'}, "
f"hub_token={'on' if HUB_TOKEN else 'off'}, "
f"role={ROLE}{' for ' + HUB_URL if STANDBY else ''}")
while not stop.is_set():
stop.wait(timeout=3600)
if RENDER_MODE != "off":
# Wait for any final monitor draw, then remove only our two canvases.
with RENDER_LOCK:
if AI_MONITOR is not None:
transport.clear(ai_status.APP_NAME)
AI_MONITOR.drawn = False
if DRAWN.is_set():
transport.clear(APP_NAME)
return 0
if __name__ == "__main__":
sys.exit(main())display_scene.py
"""Batch native display elements and remember only acknowledged screen state.
Callers provide a fresh scene on each wake-up. Failed or superseded updates
are never queued, so reconnecting draws the current state instead of a backlog.
The device continues playing native animations between these small updates.
"""
from __future__ import annotations
from dataclasses import dataclass
import json
@dataclass(frozen=True)
class DrawUpdate:
group: str
elements: list[dict]
signature: str
observed_at: float
class DrawCache:
def __init__(self, application_name: str, priority: int):
self.application_name = application_name
self.priority = priority
self._sent: dict[str, tuple[str, float]] = {}
def reset(self):
"""Invalidate the cache after clearing, yielding or waking from sleep."""
self._sent.clear()
def pending(self, group: str, elements: list[dict], now: float,
refresh_after: float) -> DrawUpdate | None:
signature = json.dumps(elements, sort_keys=True, separators=(',', ':'))
previous = self._sent.get(group)
if previous and previous[0] == signature and now - previous[1] < refresh_after:
return None
return DrawUpdate(group, elements, signature, now)
def draw(self, transport, *updates: DrawUpdate | None) -> bool:
changes = [update for update in updates if update is not None]
if not changes:
return False
elements = [element for update in changes for element in update.elements]
if not transport.draw({'application_name': self.application_name,
'priority': self.priority, 'elements': elements}):
return False
for update in changes:
self._sent[update.group] = (update.signature, update.observed_at)
return Trueeffort_animation.py
"""Pixel-exact 72x16 effort labels over increasingly energetic native effects."""
import math
from pixel_fonts import EFFORT_BOLD
from pixel_ui import Canvas, SlideFade
W, H, FPS, FRAMES = 72, 16, 25, 45
TEXT_H = EFFORT_BOLD.height
DURATION_S = FRAMES / FPS
FADE_FRAMES = 8
LEVELS = ('none', 'minimal', 'low', 'medium', 'high', 'xhigh', 'max', 'ultra')
TRANSITION = SlideFade(FRAMES, FADE_FRAMES)
# Color, speed, and motion geometry all change, rather than just tinting a wave.
PALETTES = (
((9, 15, 24), (42, 60, 80)), # none: quiet slate
((3, 22, 24), (14, 95, 90)), # minimal: teal drift
((3, 28, 21), (20, 145, 95)), # low: green aurora
((2, 25, 42), (15, 145, 210)), # medium: cyan streams
((4, 16, 48), (35, 100, 245)), # high: blue racing currents
((24, 7, 45), (155, 50, 245)), # xhigh: violet double helix
((37, 14, 2), (245, 135, 20)), # max: gold shockwaves
((32, 3, 40), (225, 45, 200)), # ultra: magenta plasma + ice sparks
)
SPEEDS = (.2, .4, .65, 1., 1.4, 2., 2.7, 3.6)
def word_pixels(word):
"""Compatibility helper for tools that inspect the effort label geometry."""
mask = EFFORT_BOLD.layout(word)
return mask.pixels, mask.width
def background(rank, x, y, t):
speed = SPEEDS[rank]
phase = t * speed
edge = (abs(y - 7.5) / 7.5) ** 1.6
wave = .5 + .5 * math.sin(x * .085 - phase * 2 + y * .18)
base, accent = PALETTES[rank]
intensity = .12 + .26 * wave
# The center is smoothly quieter over the entire width, including gaps
# between letters. No rectangular backing plate or luminous glyph outline.
field = .38 + .62 * edge
spark = 0.
if rank <= 2:
crest = math.exp(-((y - (7 + 4 * math.sin(x * .065 - phase))) / 3) ** 2)
intensity += (.04 + .08 * rank) * crest
elif rank <= 4:
streams = (.5 + .5 * math.sin(x * .16 + y * .45 - phase * 4)) ** 5
intensity += (.2 + .15 * (rank - 3)) * streams
elif rank == 5:
# Two counter-running strands have a recognisable braided silhouette.
strand = 5.5 * math.sin(x * .11 - phase * 2.2)
braid = max(math.exp(-((y - 7.5 - strand) / 1.4) ** 2),
math.exp(-((y - 7.5 + strand) / 1.4) ** 2))
intensity += .7 * braid
elif rank == 6:
# Golden pressure rings expand out of the centre, with ember trails.
radius = math.hypot((x - 35.5) * .5, (y - 7.5) * 1.7)
ring = (.5 + .5 * math.cos(radius * .8 - phase * 4)) ** 9
intensity += .95 * ring
else:
# Layered plasma and paired outward shock fronts; no full-screen flash.
plasma = (.5 + .5 * math.sin(x * .19 + math.sin(y * .8 + phase * 3)
- phase * 4)) ** 3
front = (.5 + .5 * math.cos(abs(x - 35.5) * .24 - phase * 3)) ** 12
intensity += .75 * plasma + .55 * front
# Higher levels get more, faster comets, concentrated in the free margins.
for k in range(max(0, rank - 1)):
lane = (0, 15, 1, 14, 0, 15)[k]
travel = t * (10 + speed * 17 + k * 3)
head = (travel + k * 19) % 92 - 10
if k % 2:
head = W - 1 - head
distance = (head - x) * (-1 if k % 2 else 1)
if y == lane and 0 <= distance < 8:
spark = max(spark, (1 - distance / 8) ** 2 * (.2 + rank * .11))
rgb = [a + b * intensity * field for a, b in zip(base, accent)]
if rank == 7:
ice = (.5 + .5 * math.sin(x * .15 + phase * 4)) ** 12 * edge
rgb = [c + v * ice for c, v in zip(rgb, (20, 130, 160))]
return tuple(min(255, round(c + s * spark)) for c, s in zip(rgb, (160, 190, 210)))
def _render(rank, mask, frame, direction, entering):
alpha, shift = TRANSITION.at(frame, direction, entering)
canvas = Canvas(W, H)
if alpha:
canvas.paint(lambda x, y: background(rank, x, y, frame / FPS), alpha)
canvas.draw_mask(mask, (W - mask.width) // 2 + shift,
(H - mask.height) // 2, (245, 250, 255), alpha)
return canvas.to_bgra()
def frame(effort, index, direction=1, entering=True):
"""Render one deterministic frame without computing preceding frames."""
return _render(LEVELS.index(effort), EFFORT_BOLD.layout(effort.upper()),
index, direction, entering)
def frames(effort, direction=1, n=FRAMES, entering=True):
mask = EFFORT_BOLD.layout(effort.upper())
rank = LEVELS.index(effort)
# Fully readable after 80 ms; subsequent detents replace the label
# immediately at its resting position without replaying its entrance.
return [_render(rank, mask, f, direction, entering) for f in range(n)]
def filename(effort, direction, entering=True):
return f'effort_v4_{effort}_{"up" if direction > 0 else "down"}_{"in" if entering else "change"}.anim'fast_animation.py
"""START button scenes: golden warp ignition and a blue deceleration wake."""
import math
from effort_animation import W, H, FPS, FRAMES, FADE_FRAMES, DURATION_S
from pixel_fonts import EFFORT_BOLD
from pixel_ui import Canvas, PixelMask, SlideFade
TRANSITION = SlideFade(FRAMES, FADE_FRAMES, enter_distance=12, leave_distance=8)
BOLT = PixelMask(8, 12, frozenset((x, y) for y, row in enumerate((
'00001110', '00011100', '00111000', '01110000', '11111111', '11111110',
'00011100', '00111000', '01110000', '01100000', '11000000', '10000000'))
for x, value in enumerate(row) if value == '1'))
def background(enabled, x, y, t):
edge = (abs(y - 7.5) / 7.5) ** 1.7
if enabled:
# Accelerating, perspective-stretched stars leave the central core.
travel = 22 * t + 45 * t * t
distance = abs(x - 35.5)
ray = (.5 + .5 * math.sin(distance * .31 - travel * .6 + y * 1.7)) ** 12
shock = math.exp(-((math.hypot((x - 35.5) * .4, (y - 7.5) * 1.5)
- t * 40) / 2.8) ** 2)
base, accent = (28, 8, 1), (245, 126, 12)
energy = .18 + ray * .8 + shock * 1.4
else:
# The same kinetic energy compresses into cool rings, then settles.
travel = 45 * (1 - math.exp(-3.3 * t))
radius = math.hypot((x - 35.5) * .45, (y - 7.5) * 1.7)
ring = (.5 + .5 * math.cos(radius * .7 + travel)) ** 10
base, accent = (2, 12, 29), (22, 145, 245)
energy = .17 + ring * math.exp(-t * 1.5)
rgb = [a + b * energy * (.28 + .72 * edge) for a, b in zip(base, accent)]
# Bright warp trails stay outside the letters, not underneath a text slab.
for k, lane in enumerate((0, 15, 1, 14)):
head = (travel * (1 + k * .19) + k * 23) % 98 - 13
if not enabled:
head = 71 - head
tail = (head - x) * (1 if enabled else -1)
if y == lane and 0 <= tail < 13:
strength = (1 - tail / 13) ** 2 * (1 if enabled else math.exp(-1.6 * t))
rgb = [c + s * strength for c, s in zip(rgb, (170, 205, 240))]
return tuple(min(255, round(value)) for value in rgb)
def frame(enabled, index, entering=True):
alpha, shift = TRANSITION.at(index, 1 if enabled else -1, entering)
canvas = Canvas(W, H)
if alpha:
canvas.paint(lambda x, y: background(enabled, x, y, index / FPS), alpha)
mask = EFFORT_BOLD.layout('FAST' if enabled else 'NORMAL')
width = mask.width + (12 if enabled else 0)
x = (W - width) // 2 + shift
if enabled:
canvas.draw_mask(BOLT, x, 2, (255, 220, 90), alpha)
x += 12
canvas.draw_mask(mask, x, 2, (248, 252, 255), alpha)
return canvas.to_bgra()
def frames(enabled, entering=True):
return [frame(enabled, index, entering) for index in range(FRAMES)]
def filename(enabled, entering=True):
return f'fast_v1_{"on" if enabled else "off"}_{"in" if entering else "change"}.anim'manifest.yaml
name: Busy Codex
author: wowlocal
description: Live Codex task status, account limits and weekly progress, with a reasoning effort dial and START-button Fast mode control for Codex Desktop and a native-control CLI fork.
tags:
- ai
- monitor
- info
- animation
preview: ./preview.gif
repo: https://github.com/wowlocal/busy-codexpixel_fonts.py
"""Project-owned bitmap fonts; no resampling or runtime font dependency.
EFFORT_BOLD contains the uppercase letters used by the effort labels. Add
explicitly designed glyphs here as new screens need them; unsupported letters
raise an error rather than silently becoming blank pixels.
"""
from pixel_ui import BitmapFont
# Draw at the actual LED resolution. Two-pixel stems and open counters need
# neither resampling nor dilation, both of which closed up the previous font.
_EFFORT_GLYPHS = {
'A': ('0011100', '0111110', '1100011', '1100011', '1100011', '1111111',
'1111111', '1100011', '1100011', '1100011', '1100011', '1100011'),
'D': ('1111100', '1111110', '1100111', '1100011', '1100011', '1100011',
'1100011', '1100011', '1100011', '1100111', '1111110', '1111100'),
'E': ('1111111', '1111111', '1100000', '1100000', '1100000', '1111110',
'1111110', '1100000', '1100000', '1100000', '1111111', '1111111'),
'F': ('1111111',) * 2 + ('1100000',) * 3 + ('1111110',) * 2 + ('1100000',) * 5,
'S': ('0011110', '0111111', '1110000', '1100000', '1110000', '0111110',
'0011111', '0000011', '0000011', '1000111', '1111110', '0111100'),
'G': ('0011110', '0111111', '1110000', '1100000', '1100000', '1101111',
'1101111', '1100011', '1100011', '1110011', '0111111', '0011110'),
'H': ('1100011',) * 5 + ('1111111',) * 2 + ('1100011',) * 5,
'I': ('1111',) * 2 + ('0110',) * 8 + ('1111',) * 2,
'L': ('1100000',) * 10 + ('1111111',) * 2,
'M': ('1100011', '1110111', '1111111', '1111111', '1101011', '1100011',
'1100011', '1100011', '1100011', '1100011', '1100011', '1100011'),
'N': ('1100011', '1110011', '1110011', '1111011', '1111011', '1101111',
'1101111', '1100111', '1100111', '1100011', '1100011', '1100011'),
'O': ('0011100', '0111110', '1110111', '1100011', '1100011', '1100011',
'1100011', '1100011', '1100011', '1110111', '0111110', '0011100'),
'R': ('1111100', '1111110', '1100111', '1100011', '1100111', '1111110',
'1111100', '1101100', '1100110', '1100110', '1100011', '1100011'),
'T': ('1111111',) * 2 + ('0011000',) * 10,
'U': ('1100011',) * 9 + ('1110111', '0111110', '0011100'),
'W': ('1100011',) * 6 + ('1101011', '1101011', '1111111', '1111111',
'1110111', '1100011'),
'X': ('1100011', '1100011', '0110110', '0110110', '0011100', '0011100',
'0011100', '0011100', '0110110', '0110110', '1100011', '1100011'),
}
EFFORT_BOLD = BitmapFont(_EFFORT_GLYPHS, spacing=2)pixel_ui.py
"""Small, dependency-free primitives for native-resolution display scenes.
Colors are RGB at the drawing boundary; frames are BGRA8888 for ``animgen``.
There is no device I/O or scheduling here, so a scene can be rendered offline
or used by another integration without starting the daemon.
"""
from __future__ import annotations
from dataclasses import dataclass
import struct
from typing import Callable, Mapping
import zlib
RGB = tuple[int, int, int]
def encode_png(bgra: bytes, width: int, height: int) -> bytes:
"""Encode a native display frame as lossless RGBA PNG without Pillow.
This is also useful for gallery/emulator previews, which may not implement
every on-device .anim codec. Alpha is preserved for overlay compositing.
"""
if width <= 0 or height <= 0 or len(bgra) != width * height * 4:
raise ValueError('Frame length does not match its dimensions')
def chunk(kind, data):
return (struct.pack('>I', len(data)) + kind + data
+ struct.pack('>I', zlib.crc32(kind + data) & 0xffffffff))
raw = bytearray()
stride = width * 4
for y in range(height):
raw.append(0) # PNG filter: none
for i in range(y * stride, (y + 1) * stride, 4):
b, g, r, a = bgra[i:i + 4]
raw.extend((r, g, b, a))
return (b'\x89PNG\r\n\x1a\n'
+ chunk(b'IHDR', struct.pack('>IIBBBBB', width, height, 8, 6, 0, 0, 0))
+ chunk(b'IDAT', zlib.compress(raw, 6)) + chunk(b'IEND', b''))
@dataclass(frozen=True)
class PixelMask:
width: int
height: int
pixels: frozenset[tuple[int, int]]
class BitmapFont:
"""Lay out proportional binary glyphs without scaling or smoothing."""
def __init__(self, glyphs: Mapping[str, tuple[str, ...]], spacing: int = 1):
if not glyphs or spacing < 0:
raise ValueError('A font needs glyphs and nonnegative spacing')
self.spacing = spacing
self.height = len(next(iter(glyphs.values())))
self._glyphs = {}
for letter, rows in glyphs.items():
if (len(letter) != 1 or not rows or not rows[0]
or len(rows) != self.height
or any(len(row) != len(rows[0]) or set(row) - {'0', '1'}
for row in rows)):
raise ValueError(f'Invalid bitmap glyph: {letter!r}')
self._glyphs[letter] = PixelMask(
len(rows[0]), self.height,
frozenset((x, y) for y, row in enumerate(rows)
for x, bit in enumerate(row) if bit == '1'),
)
def layout(self, text: str) -> PixelMask:
pixels = set()
offset = 0
for letter in text:
try:
glyph = self._glyphs[letter]
except KeyError:
raise ValueError(f'Font has no glyph for {letter!r}') from None
pixels.update((offset + x, y) for x, y in glyph.pixels)
offset += glyph.width + self.spacing
return PixelMask(max(0, offset - self.spacing), self.height,
frozenset(pixels))
class Canvas:
"""A clipped pixel canvas whose output can be passed directly to animgen."""
def __init__(self, width: int, height: int):
if width <= 0 or height <= 0:
raise ValueError('Canvas dimensions must be positive')
self.width, self.height = width, height
self._pixels = bytearray(width * height * 4)
def paint(self, color_at: Callable[[int, int], RGB], alpha: int = 255):
"""Fill the scene, including spaces between foreground glyphs."""
for y in range(self.height):
for x in range(self.width):
r, g, b = color_at(x, y)
index = (y * self.width + x) * 4
self._pixels[index:index + 4] = bytes((b, g, r, alpha))
def draw_mask(self, mask: PixelMask, x: int, y: int,
color: RGB, alpha: int = 255):
"""Replace covered pixels; off-screen pixels are clipped, not wrapped."""
r, g, b = color
pixel = bytes((b, g, r, alpha))
for mx, my in mask.pixels:
px, py = x + mx, y + my
if 0 <= px < self.width and 0 <= py < self.height:
index = (py * self.width + px) * 4
self._pixels[index:index + 4] = pixel
def to_bgra(self) -> bytes:
return bytes(self._pixels)
def smoothstep(value: float) -> float:
value = max(0., min(1., value))
return value * value * (3 - 2 * value)
@dataclass(frozen=True)
class SlideFade:
"""An overlay envelope that can skip its entrance on repeated input.
Timings use frame indices, so file generation and previews use the same
deterministic clock. The last frame is completely transparent.
"""
frames: int
fade_frames: int
reveal_frames: int = 2
slide_frames: int = 3
enter_distance: int = 6
leave_distance: int = 3
def __post_init__(self):
if (self.frames < 2 or not 2 <= self.fade_frames <= self.frames
or self.reveal_frames <= 0 or self.slide_frames <= 0):
raise ValueError('Transition timing must have a positive duration')
def at(self, frame: int, direction: int = 1,
entering: bool = True) -> tuple[int, int]:
"""Return (alpha, horizontal offset) for this display frame."""
frame = max(0, frame)
outro = smoothstep((frame - (self.frames - self.fade_frames))
/ (self.fade_frames - 1))
intro = max(0, 1 - frame / self.slide_frames) ** 3 if entering else 0
reveal = smoothstep(frame / self.reveal_frames) if entering else 1
alpha = round(255 * reveal * (1 - outro))
offset = round(direction * (self.enter_distance * intro
- self.leave_distance * outro))
return alpha, offsetreport.py
#!/usr/bin/env python3
"""Cross-platform forwarder + daemon supervisor (Python twin of report.sh).
report.py ensure # just make sure the daemon is running
report.py state <STATE> # forward a hook event (JSON on stdin)
report.py statusline # forward statusline JSON (on stdin)
Used by setup_claude.py on Windows (and usable everywhere); report.sh
remains for existing POSIX installs. Must never slow Claude Code down:
sub-second timeouts, failures ignored, daemon spawned detached.
Reads env.sh next to this file for configuration (plain `KEY=VALUE` or
`export KEY=VALUE` lines — the same file bash sources on macOS/Linux):
BUSYBAR_HUB=http://<hub>:8765 forward to the computer that owns the
Bar instead of running a daemon here
BUSYBAR_STANDBY=1 ... but do run a local daemon: it mirrors
to the hub and takes over the Bar (over
Wi-Fi) while the hub is asleep
BUSYBAR_HUB_TOKEN=... shared secret, if the hub requires one
BUSYBAR_HOST_TAG=W | #00A4EF how this computer's sessions are marked
on the display (letter(s) or a flag color)
BUSYBAR_HOST=studio-pc name shown in /status (default: hostname)
"""
from __future__ import annotations
import os
import pathlib
import platform
import re
import subprocess
import sys
import threading
import time
import urllib.error
import urllib.request
HERE = pathlib.Path(__file__).resolve().parent
LOG = pathlib.Path.home() / ".claude" / "busybar-daemon.log"
DOWN_MARK = HERE / ".hub_down" # touched when the hub is unreachable
DOWN_BACKOFF_S = 20 # ... then skip forwarding for this long
FORWARD_CAP_S = 1.2 # hard cap per hook, DNS included
# Loopback and LAN requests never go through a proxy (HTTP_PROXY, Windows
# system proxy) - a proxy is never the route to 127.0.0.1 or to the hub.
OPENER = urllib.request.build_opener(urllib.request.ProxyHandler({}))
PROC = None # the daemon this call started (it may still be binding)
def load_env() -> dict:
env = dict(os.environ)
if env.get("BUSYBAR_MANAGED") == "1":
return env # the gallery launcher owns configuration and process lifetime
try:
for line in (HERE / "env.sh").read_text().splitlines():
m = re.match(r'\s*(?:export\s+)?([A-Za-z_][A-Za-z0-9_]*)=("?)(.*)\2\s*$', line)
if m and not line.lstrip().startswith("#"):
env[m.group(1)] = m.group(3)
except OSError:
pass
return env
ENV = load_env()
PORT = int(ENV.get("BUSYBAR_PORT", "8765"))
HUB = ENV.get("BUSYBAR_HUB", "").strip().rstrip("/")
STANDBY = ENV.get("BUSYBAR_STANDBY", "").strip().lower() in ("1", "true", "yes", "on")
# Forwarder mode talks to the hub and runs nothing here; standby mode talks
# to the local daemon, which mirrors to the hub itself (daemon.py HubLink).
FORWARD_TO_HUB = bool(HUB) and not STANDBY
BASE = HUB if FORWARD_TO_HUB else f"http://127.0.0.1:{PORT}"
HEADERS = {"Content-Type": "application/json",
"X-Busybar-Host": (ENV.get("BUSYBAR_HOST") or platform.node().split(".")[0])[:64] or "?"}
if ENV.get("BUSYBAR_HOST_TAG"):
HEADERS["X-Busybar-Host-Tag"] = ENV["BUSYBAR_HOST_TAG"].strip()
if ENV.get("BUSYBAR_HUB_TOKEN"):
HEADERS["X-Busybar-Token"] = ENV["BUSYBAR_HUB_TOKEN"].strip()
def daemon_alive() -> bool:
try:
OPENER.open(BASE + "/health", timeout=0.4)
return True
except OSError:
return False
def ensure_daemon():
if FORWARD_TO_HUB or daemon_alive():
return # a hub runs the daemon for us
try:
if LOG.exists() and LOG.stat().st_size > 1 << 20:
LOG.write_text("")
LOG.parent.mkdir(parents=True, exist_ok=True)
log = open(LOG, "ab")
except OSError:
log = subprocess.DEVNULL
kwargs: dict = {"stdout": log, "stderr": log, "stdin": subprocess.DEVNULL,
"env": ENV, "cwd": str(HERE)}
global PROC
flags = 0
if os.name == "nt":
flags = subprocess.DETACHED_PROCESS | subprocess.CREATE_NEW_PROCESS_GROUP
# CREATE_BREAKAWAY_FROM_JOB: outlive a job object the hook runs in.
kwargs["creationflags"] = flags | 0x01000000
else:
kwargs["start_new_session"] = True
# sys.executable sidesteps the python vs python3 naming mess entirely.
try:
PROC = subprocess.Popen([sys.executable, str(HERE / "daemon.py")], **kwargs)
except OSError:
if os.name != "nt":
raise
kwargs["creationflags"] = flags # breakaway not permitted here
PROC = subprocess.Popen([sys.executable, str(HERE / "daemon.py")], **kwargs)
def _hub_down() -> bool:
try:
return time.time() - DOWN_MARK.stat().st_mtime < DOWN_BACKOFF_S
except OSError:
return False
def forward(path: str, body: bytes) -> bool:
"""POST to the daemon/hub. Runs in a daemon thread so a stalled DNS
lookup or an asleep hub can never hold a Claude Code hook longer than
FORWARD_CAP_S; an unreachable hub is then backed off for a while."""
if FORWARD_TO_HUB and _hub_down():
return False
ok = [False]
def _post():
deadline = time.time() + FORWARD_CAP_S
while True:
try:
OPENER.open(urllib.request.Request(BASE + path, data=body, headers=HEADERS),
timeout=1)
ok[0] = True
return
except urllib.error.URLError as e:
# A daemon we just started may still be binding its port -
# but one that already exited (bad env.sh) never will.
if (PROC is not None and PROC.poll() is None
and isinstance(e.reason, ConnectionRefusedError)
and time.time() < deadline):
time.sleep(0.05)
continue
return
except OSError:
return
t = threading.Thread(target=_post, daemon=True)
t.start()
t.join(FORWARD_CAP_S)
if FORWARD_TO_HUB:
try:
if ok[0]:
DOWN_MARK.unlink(missing_ok=True)
else:
DOWN_MARK.touch()
except OSError:
pass
return ok[0]
def main():
mode = sys.argv[1] if len(sys.argv) > 1 else "ensure"
ensure_daemon()
if mode == "state":
state = sys.argv[2] if len(sys.argv) > 2 else "WORKING"
forward(f"/state?state={state}", sys.stdin.buffer.read() or b"{}")
elif mode == "statusline":
forward("/statusline", sys.stdin.buffer.read() or b"{}")
return 0
if __name__ == "__main__":
sys.exit(main())Source and Provenance
View This Version on GitHub → · Author’s Project →
Source commit: 26b19ba9ce2ba192c45b4c0eadcfad53df42b153
Content SHA-256: 4f90057d292ebbd8020da32c459adc921a6f5c17d43a6530711c9c865869405a
Related Apps
Claude Limits
Live Claude Code usage — 5-hour session and weekly limit bars with time to reset, an animated Clawd mascot, and a session-activity readout.
Community mirror▲ 0↓ 0
Apple Music Winamp
macOS Apple Music controls with song, artist, progress, a time counter, decorative Winamp-style visualizers, and animated lightning credits.
Community mirror▲ 0↓ 0
BusyBar Codex
Tactile Codex controller with live task status, actions, navigation, reasoning control, and push-to-talk dictation.
Community mirror▲ 0↓ 0
Rate and Discuss
Comments and reactions live in this site’s GitHub repository. Sign in with GitHub to participate.