app
KiCad - the KiCad Bridge
Public Made by Adomby adom
Drive your own KiCad from an AI, in the background: open every editor, read the whole board as data (the EDA Bridge Standard board model), stack-up, part heights, netlists, place, route, pour, DRC and exports.
master
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803
#!/usr/bin/env python3
"""kicad-dashboard - the KiCad bridge's Launch Page.
Parity sibling of fusion-dashboard (adom/fusion-bridge dashboard/src/main.rs):
same LED-with-_hint model, same live channel, same AI-drivability contract, same
visual system. Status comes from shelling `adom-bridge`, so the BRIDGE stays the
single source of truth and this server owns no state the bridge could contradict.
Serves:
GET / the app (dashboard/ui/index.html, read per request
so a ralph loop only needs a browser reload)
GET /favicon.svg
GET /events SSE live channel: hello / state / toast / click
GET /api/targets every connected box, lightly probed
GET /api/status[/<t>[/force]] the LED payload (also /state)
GET /console the server's own log lines (what the AI reads back)
GET /health liveness + sseClients (live viewer count)
POST /api/action {action,target,args}
POST /ui/toast {message,kind} narrate OVER the running app
POST /ui/click {target} the PAGE clicks its own control
CLI:
kicad-dashboard serve [--port 8931] [--target BOX] [--print-url] [--dev]
kicad-dashboard status [--target BOX]
kicad-dashboard ui toast "msg" [--kind ok|err|info]
kicad-dashboard ui click <setup|launch|close|demo|part-tour|settings|refresh>
kicad-dashboard ui state | ui console
kicad-dashboard open [--pup | --webview] [--target BOX]
kicad-dashboard health
--dev (or KICAD_DASHBOARD_DEV=1) adds the maintainer-only Daily suite LED. Normal
Adom users never run the nightly verb suite, so it is off by default.
PROXY NOTE (paid for live): adom.cloud STRIPS query strings on sub-paths, so the
target rides the PATH (/api/status/<target>/force), never ?target=. The SSE route
also needs the padding + no-transform headers or the proxy buffers every byte.
"""
import json
import os
import queue
import re
import shutil
import subprocess
import sys
import tempfile
import threading
import time
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import unquote, urlparse
VERSION = "4.8.1"
AI_THREAD = "kicad-dashboard"
DEFAULT_PORT = 8931
UI_DIR = Path(__file__).resolve().parent / "ui"
# SUPERSEDED, and worth stating so nobody re-litigates it. John, 2026-08-18,
# told me to ship the REAL KiCad logo ("go with the logo the user is familiar
# with and not try to reinvent the wheel") after I had invented a teal mark.
# John, 2026-08-24, looking at ah's webview tab strip where this blue filled
# tile sat beside white line icons: "this is awful. all of these should be our
# standard type of icon which is just a line drawing in white on a transparent
# background." The later ruling is narrower and explicit about THIS artifact,
# and adom-ui-design 1.10.0's `icons` skill made it fleet law (rule 4: never
# ship the wrapped vendor's mark - off-palette, filled, trademarked, and it
# turns the tab strip into a logo salad). So the identity icon is now our own
# line drawing of what a KiCad user works on; the 08-18 rule survives only in
# its real scope, which was "do not invent a random abstract mark".
LOGO_PATH = UI_DIR / "dock-kicad.svg"
# "Say it" (John via the dock thread, 2026-08-23): teach users how to ASK for
# this app by its noun, first and foremost, at the top of the dashboard. The
# list lives in dockbar.json (`say`) so the dock's interstitial and the
# dashboard teach the same phrases; the built-in copy is the fallback for an
# installed dashboard that ships without the manifest beside it.
# Every phrase names REAL wiki content a user can open right now (John,
# 2026-09-03: "give them real content to open or try in each example").
# ray/xl555 is a full part page (.kicad_sym, .kicad_mod, .step); the adom/
# boards carry .kicad_pcb + .kicad_sch; Adom Basic Parts is the PCM library on
# adom/kicad-pcm. If one of these pages disappears, replace it with another
# real one; never fall back to a placeholder phrase.
_SAY_FALLBACK = [
"open the XL555 symbol from wiki part ray/xl555 in kicad",
"send the footprint of ray/xl555 to kicad",
"show me adom/drone-motor-carrier in kicad 3d",
"open the wiki project adom/esc-g431 in kicad",
"install the Adom Basic Parts library into kicad",
"export gerbers for adom/carrier-fc-rp2350a-mini from kicad"
]
def say_phrases():
for cand in (Path(__file__).resolve().parent.parent / "dockbar.json",
Path.home() / ".local" / "share" / "adom" / "kicad-bridge" / "dockbar.json"):
try:
lst = json.loads(cand.read_text(encoding="utf-8")).get("say")
if isinstance(lst, list) and lst:
return [str(x) for x in lst][:6]
except Exception:
continue
return list(_SAY_FALLBACK)
HYDROGEN_CONTROL = "http://127.0.0.1:47084"
def _hydrogen_control_url():
"""The workspace control API base, from the well-known file the panel
writes (~/.adom/hydrogen-control-url). None when this machine has no
panel - the caller then falls back to copy."""
try:
u = (Path.home() / ".adom" / "hydrogen-control-url").read_text(
encoding="utf-8").strip()
return u.rstrip("/") or None
except Exception:
return None
LAST_VIEWPORT = {}
VIEWPORT_LOG = []
MIN_SHELL_VERSION = (1, 0, 142) # the build that fixed the run-prompt kill
def _ver_tuple(v):
"""Numeric version tuple. A LEXICAL compare passes 1.0.99 as >= 1.0.142
(issue #59), so every comparison in this file goes through here."""
out = []
for part in str(v or "").strip().lstrip("v").split("."):
digits = "".join(c for c in part if c.isdigit())
out.append(int(digits) if digits else 0)
return tuple(out[:3]) or (0,)
# ---------------------------------------------------------------------------
# ROUTE TABLE. Two catch-alls used to live here and each one made a check pass
# for the wrong reason (fleet sweep, 2026-08-25):
# * do_GET rendered index.html for EVERY unmatched path, so a missing settings
# PNG, demo mp3 or favicon came back as HTML 200. The launch-console 404
# watch is the one thing watched on a virgin install, so the check that
# exists to catch a broken fresh install could never fire. It also made
# proxy-fidelity row 3 ("not upstream's 418") pass on blanket-SERVED paths.
# * do_OPTIONS answered 204 for EVERY path, so the module's checkSend probes
# the send route with OPTIONS and its "send route wired: not 404" row went
# green for routes that do not exist at all.
# A permissive default does not fail a check, it ANSWERS it. Both now consult
# this one table, so a route cannot be live for OPTIONS and dead for GET.
#
# Matching mirrors each handler EXACTLY, endswith included: the adom.cloud proxy
# mounts this app under a sub-path and forwards the prefix, so "/apps/kicad/health"
# is the same route as "/health".
_EXACT_ROUTES = ("/health", "/console", "/events", "/api/targets",
"/api/action", "/ui/toast", "/ui/click")
_SUFFIX_ROUTES = ("/newbie-panel.js", "/favicon.svg", "/favicon.png", "/favicon.ico",
"/dock-kicad.svg", "/kicad-logo.png", "/updater/status",
"/api/viewport", "/ai-threads/run-prompt", "/ai-threads/run-prompt/status",
"/api/say-send")
_REGEX_ROUTES = (r"/assets/(.+)$",
r"^/api/verb_times(?:/([^/]+))?$",
r"^/(?:api/status|state)(?:/([^/]+))?(/force)?$")
# Namespaces we own: an unmatched path under one of these is a WRONG route, never
# an app URL, so it 404s even for a browser navigation.
_OWNED_PREFIXES = ("/api/", "/ui/", "/updater/", "/assets/", "/ai-threads/")
_ASSET_SUFFIXES = (".png", ".jpg", ".jpeg", ".webp", ".svg", ".ico", ".gif",
".mp3", ".wav", ".ogg", ".mp4", ".webm",
".js", ".mjs", ".css", ".map", ".json", ".woff", ".woff2", ".ttf")
def _route_exists(path):
"""True if `path` reaches a real handler. Used by do_OPTIONS so a preflight
cannot claim a route that GET/POST would not serve."""
if path in _EXACT_ROUTES or path.endswith(_SUFFIX_ROUTES):
return True
return any(re.search(rx, path) for rx in _REGEX_ROUTES)
def _wants_app_shell(handler, path):
"""The app is a single page, and under the proxy its root arrives WITH the
mount prefix ("/apps/kicad"), which is shape-identical to a typo. So the
shell is served only for a real browser navigation, and never for a path
that is asset-shaped or inside a namespace we own.
Stated honestly because it is the load-bearing caveat: a browser typing a
wrong extensionless URL still gets the shell. Every case the 404 watch
actually watches - a missing png, mp3, favicon, or js - carries an extension
or an owned prefix and 404s regardless of who asks."""
if path == "/":
return True # the app's own root, whatever client asks for it
if path.endswith(_ASSET_SUFFIXES):
return False
if any(pre in path + "/" for pre in _OWNED_PREFIXES):
return False
# A browser NAVIGATION always sends text/html in Accept; curl and every
# other tool send */* or nothing. That is the only signal separating the
# proxy's mounted root from a typo, and it is why the one-line fleet test
# (curl <base>/nonsense) gets an honest 404.
return "text/html" in (handler.headers.get("Accept") or "")
def _proxy_control(handler, path, body=None, timeout=20):
"""Forward one control-API call VERBATIM: same status, same bytes.
Both halves matter. Re-serialising the JSON (json.loads then json.dumps)
changes key order and whitespace, and hardcoding 200 hides an upstream 207
or 404 from the caller - the module decides what to do from the status and
the body it actually gets, so a proxy that "helpfully" normalises either is
lying to it. Non-2xx arrives as HTTPError and is forwarded, not swallowed.
(Proven with assets/tools/proxy-fidelity.py, which reds a re-serialising
proxy in its own self-test before it judges yours.)"""
import urllib.request, urllib.error
base = _hydrogen_control_url()
if not base:
return handler._json({"ok": False, "why": "no control api on this machine"}, 503)
req = urllib.request.Request(
base + path, method="POST" if body is not None else "GET",
data=json.dumps(body).encode("utf-8") if body is not None else None,
headers={"Content-Type": "application/json"} if body is not None else {})
try:
with urllib.request.urlopen(req, timeout=timeout) as r:
return handler._send(r.status, r.headers.get("Content-Type", "application/json"),
r.read())
except urllib.error.HTTPError as e:
return handler._send(e.code, e.headers.get("Content-Type", "application/json"),
e.read())
except Exception as e:
return handler._json({"ok": False, "why": f"{e.__class__.__name__}: {e}"}, 502)
def say_send(phrase):
"""Send-to-AI (un-parked 2026-08-24 on John's ruling, after the ah build
thread certified the run-prompt crash family dead on the gate-fixed build).
ONE call, per the corrected contract in issue #58: POST /ai-threads/run-prompt
with new:true. The earlier two-call flow (/ai-threads/new then run-prompt
with its sid) was WRONG and shipped fleet-wide: /new has never returned an
id, because a thread has no sid until its first submitted prompt, so the
button opened a tab and delivered nothing.
Still NEW threads only: resume and typing-into-existing stay parked.
Every failure returns ok:False with a readable `why`; the page then
error-toasts AND copies the phrase, so a user is never stranded."""
import urllib.request, urllib.parse
base = _hydrogen_control_url()
if not base:
return {"ok": False, "why": "no AI panel on this machine (hydrogen-control-url missing)"}
def _get(path, timeout):
with urllib.request.urlopen(base + path, timeout=timeout) as r:
return json.loads(r.read().decode("utf-8") or "{}")
# SAFETY GATE (issue #59): a shell older than 1.0.142 still carries the
# crash, so refuse honestly there instead of killing the user's editor.
# Numeric compare, never lexical.
try:
st = _get("/updater/status", 8)
d = st.get("data") if isinstance(st.get("data"), dict) else st
cur = (d.get("updater") or d).get("current_version") or d.get("current_version")
if cur and _ver_tuple(cur) < MIN_SHELL_VERSION:
return {"ok": False, "shellTooOld": True, "currentVersion": str(cur),
"why": f"your Hydrogen ({cur}) predates the Send to AI fix; update to "
f"1.0.142 or newer to enable it. The phrase was copied instead."}
except Exception:
pass # cannot read the updater: fall through rather than block a healthy shell
# Wiki #77 (Hydrogen 1.0.213 AI picker): NO `type` here, so the human picks
# the AI (Claude, Codex, Kimi) in the shell's picker; the app: tag keys the
# "remember for this app" checkbox. dock: stays for the dock's own bookkeeping.
body = {"new": True, "prompt": phrase, "submit": True, "background": True,
"name": "KiCad - say it", "tags": ["dock:kicad-bridge", "app:adom/kicad-bridge"]}
# background:true (Hydrogen 1.0.219, wiki #77): the answer comes back at once
# with {job, state:"accepted"}; the picker's 5 to 45 s block and the delivery
# happen detached, and the status route narrates them (state + human).
req = urllib.request.Request(base + "/ai-threads/run-prompt",
data=json.dumps(body).encode("utf-8"), method="POST",
headers={"Content-Type": "application/json"})
try:
# 120s: creating the tab alone can take 60s. NEVER retry blind on a
# timeout - a retry can open a second tab for one click.
with urllib.request.urlopen(req, timeout=120) as r:
res = json.loads(r.read().decode("utf-8") or "{}")
except Exception as e:
return {"ok": False, "why": f"the AI panel did not answer ({e.__class__.__name__}). "
f"Not retried, so no duplicate tab. The phrase was copied instead."}
d = res.get("data") if isinstance(res.get("data"), dict) else res
if isinstance(d, dict) and d.get("cancelled"):
# Wiki #77: the user dismissed Hydrogen's AI picker. Nothing opened,
# nothing was delivered; say so and never retry.
return {"ok": False, "cancelled": True,
"why": "you dismissed the AI picker; nothing was opened or sent. The phrase is still on your clipboard."}
# Hydrogen 1.0.187+ (newbie-panel.js 2.22.0 contract): run-prompt returns in
# under a second with {ok, job, state:"opening"}; ok means the job STARTED
# and delivered_to is never filled on this path. Progress is a poll of
# GET /ai-threads/run-prompt/status?job=<job>: opening, prefilled, submitted,
# working, needs_you, done; terminal failures submit_failed,
# submitted_unidentified, submit_unconfirmed, still_working. "working" or
# later counts as sent. Older builds return delivered_to and fall through.
job = d.get("job") if isinstance(d, dict) else None
if job:
import time as _t
# Hydrogen 1.0.219 job states (wiki #77): accepted, picking, picked, opening,
# prefilled | tab_found, submitted, working, needs_you | done; Codex and Kimi
# end at delivered. Failure terminals: cancelled, rejected, open_failed,
# open_unconfirmed, tab_not_found, inject_failed, submit_failed, failed,
# timeout. `terminal:true` says the job is over; `human` is the toast line.
sent_states = ("working", "needs_you", "done", "delivered")
fail_states = ("cancelled", "rejected", "open_failed", "open_unconfirmed", "tab_not_found",
"inject_failed", "submit_failed", "failed", "timeout",
"submitted_unidentified", "submit_unconfirmed", "still_working")
last = str(d.get("state") or "accepted"); human = str(d.get("human") or "")
picked = None
deadline = _t.monotonic() + 75.0 # the picker alone may hold 45 s
while _t.monotonic() < deadline:
try:
st = _get("/ai-threads/run-prompt/status?job=" + urllib.parse.quote(str(job)), 4)
sd = st.get("data") if isinstance(st.get("data"), dict) else st
last = str(sd.get("state") or last); human = str(sd.get("human") or human)
jb = sd.get("job") if isinstance(sd.get("job"), dict) else {}
picked = picked or jb.get("type") or sd.get("type")
terminal = bool(sd.get("terminal"))
except Exception:
terminal = False
if last == "cancelled":
return {"ok": False, "cancelled": True, "job": job, "state": last,
"why": human or "you dismissed the AI picker; nothing was opened or sent. The phrase is still on your clipboard."}
if last in sent_states:
return {"ok": True, "job": job, "state": last, "ai": picked,
"why": human or f"sent to {picked or 'your AI'}: the new thread has the phrase"}
if last in fail_states or (terminal and last not in sent_states):
why = ("the prompt is sitting in the new tab and Send was never clicked; press Enter there"
if last == "submit_failed" else (human or f"the AI panel reported {last}"))
return {"ok": False, "job": job, "state": last, "why": why + ". The phrase was also copied."}
_t.sleep(1.0)
return {"ok": True, "job": job, "state": last, "unconfirmed": True,
"why": (human or f"the new thread is still {last}") + " after 75 s; check the new AI tab. The phrase was also copied."}
# VERIFY BY DELIVERY, never by the thread's name (issue #59: a correctly
# named tab that delivered nothing, and an unnamed tab that delivered -
# measured live on AdomLapper 2026-08-25, where a successful send came back
# with "title": null).
#
# BUT delivered_to ALONE PRODUCES FALSE FAILURES. The API's own hint calls
# it "a diff, not a guess" that is "empty ... if verification has not caught
# up", and that is exactly what my first live send hit: an empty array on a
# prompt that had in fact been typed and submitted. Accept EITHER the
# transcript diff or the injector's own confirmation.
inject = d.get("inject") if isinstance(d.get("inject"), dict) else {}
delivered = (d.get("delivered_to")
or inject.get("delivered_in_conversation")
or d.get("delivered_in_conversation"))
if not delivered:
hint = d.get("_hints") or d.get("_hint") or res.get("_hints") or ""
if isinstance(hint, dict):
hint = " ".join(str(v) for v in hint.values())
# ok:true with no delivery signal YET is "opened but unconfirmed", which
# is not the same as failed. Say that precisely.
if d.get("ok") or res.get("ok"):
return {"ok": False, "unconfirmed": True,
"why": "the thread opened and the prompt was submitted, but delivery is "
"not confirmed yet. Check the new AI tab; the phrase was also "
"copied so you can paste it if it did not land."}
return {"ok": False, "why": ("the AI panel opened a thread but did not confirm "
"delivery. " + (str(hint)[:200] if hint else "")).strip()}
return {"ok": True, "deliveredTo": str(delivered)[:200],
"tid": d.get("tid"), "sid": d.get("sid")}
def ab(target, verb, args=None, timeout=140):
"""One bridge verb. Returns the parsed payload dict (never raises)."""
cmd = ["adom-bridge", "--ai-thread", AI_THREAD]
if target and target != "-":
cmd += ["--target", target]
cmd += [verb, json.dumps(args or {})]
try:
p = subprocess.run(cmd, capture_output=True, text=True, timeout=timeout)
d = json.loads(p.stdout or "{}")
# ab wraps some payloads in output-as-JSON-string; unwrap when it is one
out = d.get("output")
if isinstance(out, str) and out.strip().startswith("{"):
try:
inner = json.loads(out)
merged = dict(d)
merged.update(inner.get("data") if isinstance(inner.get("data"), dict) else inner)
return merged
except Exception:
pass
return d
except subprocess.TimeoutExpired:
return {"error": f"no answer within {timeout}s", "timedOut": True}
except Exception as e:
# never leak a raw Python exception (a TimeoutExpired repr used to land
# verbatim in the Bridge LED as "Command '['adom-bridge', ...]'")
return {"error": str(e).split("\n")[0][:120]}
def get_targets():
try:
p = subprocess.run(["adom-bridge", "targets"], capture_output=True, text=True, timeout=30)
d = json.loads(p.stdout or "{}")
return [t.get("name") for t in (d.get("targets") or []) if t.get("name")]
except Exception:
return []
# ------------------------------------------------------------------- SSE hub
class Hub:
def __init__(self):
self._clients = []
self._lock = threading.Lock()
def add(self):
q = queue.Queue(maxsize=64)
with self._lock:
self._clients.append(q)
return q
def drop(self, q):
with self._lock:
if q in self._clients:
self._clients.remove(q)
def count(self):
with self._lock:
return len(self._clients)
def broadcast(self, event, data):
msg = f"event: {event}\ndata: {json.dumps(data)}\n\n"
with self._lock:
for q in list(self._clients):
try:
q.put_nowait(msg)
except queue.Full:
pass
HUB = Hub()
CONSOLE = []
_console_lock = threading.Lock()
def log(line):
with _console_lock:
CONSOLE.append(f"{time.strftime('%H:%M:%S')} {line}")
del CONSOLE[:-300]
print(line, file=sys.stderr, flush=True)
# ------------------------------------------------------------------ LED model
DEV_MODE = bool(os.environ.get("KICAD_DASHBOARD_DEV"))
def _suite_led():
"""DEV-ONLY (John, 2026-08-18: "the daily suite should not be in here. no
normal adom user runs that. that's just for me as the main dev"). Shown only
with --dev / KICAD_DASHBOARD_DEV=1: the nightly verb-suite verdict plus
container-cron liveness (the cron died unnoticed for 4 days once)."""
try:
try:
hb = time.time() - os.path.getmtime("/tmp/cron-heartbeat")
if hb > 600:
return {"label": "Daily suite", "state": "err",
"detail": f"container cron DEAD ({int(hb // 60)} min since heartbeat)",
"_hint": "The container's cron stopped, so the 6:30am verb suite will not run. "
"Restart cron in the container and re-check."}
except OSError:
pass
rp = "/home/adom/.kicad-verbtest-last-report.json"
if not os.path.exists(rp):
rp = "/tmp/kicad-verbtest/report.json"
age_h = (time.time() - os.path.getmtime(rp)) / 3600
t = (json.load(open(rp)).get("totals")) or {}
det = f"{t.get('pass', '?')} pass / {t.get('fail', '?')} fail ({int(age_h)}h ago)"
if age_h > 48:
return {"label": "Daily suite", "state": "warn", "detail": f"stale: {det}",
"_hint": "No suite run in over 48h. Check the cron entry and /tmp/kicad-verbtest/cron.log."}
if t.get("fail"):
return {"label": "Daily suite", "state": "warn", "detail": det,
"_hint": "Failures are triaged on the wiki issues; the suite files a ticket only for NEW ones."}
return {"label": "Daily suite", "state": "ok", "detail": det,
"_hint": "Every verb passed on the last nightly run."}
except Exception:
return {"label": "Daily suite", "state": "off", "detail": "no suite report yet",
"_hint": "The nightly suite writes ~/.kicad-verbtest-last-report.json; "
"it runs at 6:30am when the box is idle."}
# ── The dock card's ROLLUP (wiki #43, John) ────────────────────────────────
# Without a top-level `state` the dock takes our WORST led, so a yellow
# "update available" turns the whole card yellow while the bridge answers
# perfectly. That is not what a user should read off a card. We own the
# rollup, so we own the judgement about which facets are load-bearing.
#
# I initially declined this, reasoning that a disagreement between our rollup
# and the dock's derivation would be a signal worth seeing. John's argument is
# better: the card is a USER-facing health light, not a debugging surface, and
# an advisory LED must not be able to raise an alarm.
_CRITICAL_LEDS = ("relay", "box", "bridge", "kicad") # can this thing work at all
_ADVISORY_LEDS = ("update", "focus", "presence", "plugin") # informational, never alarming
def _emit_short_states(leds: dict) -> None:
"""THE SERVER emits each pill's short state (#59). The page used to trim
`detail` to 30 chars, which is what produced the fleet's "in talled and
pinned" - a client cannot shorten prose it does not understand. Anything
without a hand-written short label falls back to the STATE word, never to
a truncated sentence."""
SHORT = {
"relay": lambda l: ("connected" if l.get("state") == "ok" else "no boxes"),
"box": lambda l: ("online" if l.get("state") == "ok" else "offline"),
"bridge": lambda l: ("answering" if l.get("state") == "ok" else "not answering"),
"kicad": lambda l: (str(l.get("detail") or "").strip()
if l.get("state") == "ok" else "not installed"),
"update": lambda l: ("current" if l.get("state") == "ok" else "update ready"),
"plugin": lambda l: (str(l.get("detail") or "").split(" - ")[0].strip()
if l.get("state") == "ok" else "not bound"),
"focus": lambda l: ("quiet" if l.get("state") == "ok" else "busy"),
"presence": lambda l: ("idle" if l.get("state") == "ok" else "active"),
"suite": lambda l: ("passing" if l.get("state") == "ok" else "check it"),
}
for key, led in (leds or {}).items():
if not isinstance(led, dict):
continue
try:
val = SHORT.get(key, lambda l: None)(led)
except Exception:
val = None
if not val:
val = {"ok": "ok", "err": "broken", "warn": "degraded",
"busy": "working", "off": "unknown"}.get(led.get("state"), "unknown")
led["short"] = str(val)[:24]
def rollup_state(leds: dict) -> dict:
_emit_short_states(leds)
"""Green when the bridge can actually do its job; yellow only when a
CRITICAL facet is degraded; red when one is broken."""
def st(k):
return (leds.get(k) or {}).get("state")
crit = [st(k) for k in _CRITICAL_LEDS if st(k)]
if "err" in crit:
worst, why = "err", [k for k in _CRITICAL_LEDS if st(k) == "err"]
elif "warn" in crit:
worst, why = "warn", [k for k in _CRITICAL_LEDS if st(k) == "warn"]
else:
worst, why = "ok", []
degraded = [k for k in _ADVISORY_LEDS if st(k) in ("warn", "err")]
return {
"state": worst,
"reason": (", ".join(why) if why else
("healthy" + (f" (advisory: {', '.join(degraded)})" if degraded else ""))),
"criticalLeds": list(_CRITICAL_LEDS),
"advisoryLeds": list(_ADVISORY_LEDS),
"_hint": ("Top-level rollup for the dock card LED. Advisory facets (an available "
"update, focus bounces, user presence, a plugin that only degrades "
"background navigation) never turn the card yellow, because the bridge "
"still works. Only relay/box/bridge/kicad can."),
}
def build_status(target):
"""The LED payload. Each facet: {label, state: ok|warn|err|off, detail, _hint}."""
targets = get_targets()
if not target or target == "-":
target = targets[0] if targets else ""
leds = {}
box_ok = bool(target and target in targets)
leds["relay"] = {"label": "Relay", "state": "ok" if targets else "err",
"detail": f"{len(targets)} box(es) connected" if targets else "no boxes on the relay",
"_hint": "Boxes connect through adom-bridge. `adom-bridge targets` lists them; "
"if empty, the desktop app is closed or offline."}
leds["box"] = {"label": "Box", "state": "ok" if box_ok else "err",
"detail": target if box_ok else f"'{target or 'none'}' not connected",
"_hint": "Pick a connected box with the pills above; the bridge runs there, not in this container."}
if not box_ok:
for k, lbl in (("bridge", "Bridge"), ("kicad", "KiCad"), ("update", "Updates"),
("plugin", "Plugin"), ("focus", "Focus contract"), ("presence", "User presence")):
leds[k] = {"label": lbl, "state": "off", "detail": "box offline",
"_hint": "Unknowable while the box is off the relay - dormant, not broken."}
if DEV_MODE:
leds["suite"] = _suite_led()
return {"target": target, "targets": targets, "leds": leds, "settings": {},
**rollup_state(leds),
"windows": [], "dashboardVersion": VERSION,
"generatedAt": time.strftime("%H:%M:%S"),
"_hint": "Every LED carries its own _hint naming the verb that fixes it. "
"POST /api/action {action,args} to act; subscribe to /events for live pushes."}
T = 25 # per-facet ceiling: a box that cannot answer in 25s is "not responding"
r = ab(target, "kicad_readiness", {}, timeout=T)
if r.get("error"):
# THE TRUTHFULNESS RULE (ralph iter-3): if the box did not answer, we do
# not KNOW anything downstream. Saying "KiCad not installed" or "up to
# date" here would be a confident lie built on a failed call.
leds["box"] = {"label": "Box", "state": "err",
"detail": f"{target} not responding",
"_hint": "The box is on the relay but its bridge did not answer in time. "
"It may be asleep, mid-upgrade, or the desktop app is wedged - "
"pick another box with the pills, or retry with Refresh."}
for k, lbl in (("bridge", "Bridge"), ("kicad", "KiCad"), ("update", "Updates"),
("plugin", "Plugin"), ("focus", "Focus contract"),
("presence", "User presence")):
leds[k] = {"label": lbl, "state": "off", "detail": "unknown",
"_hint": f"Unknowable while {target} is not answering ({r.get('error')})."}
if DEV_MODE:
leds["suite"] = _suite_led()
return {"target": target, "targets": targets, "leds": _apply_activity(leds),
**rollup_state(_apply_activity(leds)),
**rollup_state(_apply_activity(leds)),
"settings": {},
"windows": [], "dashboardVersion": VERSION,
"generatedAt": time.strftime("%H:%M:%S"),
"_hint": "The selected box did not answer; every dependent facet reads unknown "
"rather than guessing. Switch boxes or retry."}
installed = bool(r.get("installed", r.get("defaultVersion")))
cu = ab(target, "kicad_check_for_updates", {}, timeout=T)
ver = cu.get("currentVersion") or r.get("defaultVersion") or "?"
leds["bridge"] = {"label": "Bridge", "state": "ok" if not r.get("error") else "err",
"detail": "answering" if not r.get("error") else str(r.get("error"))[:70],
"_hint": "The kicad_* verb surface is served by the bridge on the box. If red: "
"bridge_kill {name:'kicad',confirm:true} respawns it from cache."}
leds["kicad"] = {"label": "KiCad", "state": "ok" if installed else "err",
"detail": f"KiCad {ver}" if installed else "not installed",
"_hint": "KiCad is installed and driveable." if installed
else "kicad_upgrade {} installs it silently per-user (no UAC prompt)."}
if cu.get("upgradeAvailable"):
leds["update"] = {"label": "Updates", "state": "warn",
"detail": f"{cu.get('currentVersion')} -> {cu.get('latestVersion')} available",
"_hint": "kicad_close then kicad_upgrade {} installs the official release silently. "
"Verify with a gerber export version stamp."}
else:
leds["update"] = {"label": "Updates", "state": "ok",
"detail": f"up to date (latest {cu.get('latestVersion') or '?'})",
"_hint": "The bridge checks upstream every 24h and nudges the AI when a release lands."}
bs = ab(target, "kicad_bridge_status", {}, timeout=T)
summary = bs.get("summary") or {}
alive = summary.get("alive") or bs.get("alive") or 0 # alive lives under summary
registered = summary.get("total") or len(bs.get("plugins") or [])
# ONE state read for the whole payload: the plugin LED used to make its own
# call, so it could say "KiCad closed" while the windows line right below it
# listed an open window (ralph iter-7).
st = ab(target, "kicad_state", {"includeImages": False}, timeout=T)
ks = st.get("kicadState") or st
kicad_open = bool(ks.get("windows"))
if alive:
leds["plugin"] = {"label": "Plugin", "state": "ok",
"detail": f"{alive} live instance(s)",
"_hint": "The in-process plugin is bound, so background part navigation "
"(symbol/footprint loading without touching the GUI) is available."}
elif not kicad_open:
leds["plugin"] = {"label": "Plugin", "state": "off", "detail": "KiCad closed",
"_hint": "The plugin lives inside KiCad, so it can only bind once KiCad "
"runs. Launch KiCad and it binds shortly after."}
else:
leds["plugin"] = {"label": "Plugin", "state": "warn",
"detail": f"KiCad open but plugin not answering ({registered} stale)",
"_hint": "KiCad is running yet no plugin instance answers. On KiCad 10.0.5 "
"the embedded Python binds late or not at all (tracked as issue #31): "
"kicad_install_plugin redeploys it, and closing/reopening KiCad "
"usually binds it. Everything else works without it - only "
"background part navigation is degraded."}
fd = ks.get("focusDebug") or {}
steals = [e for e in (ks.get("focusEvents") or [])[-20:] if "bounced" in str(e.get("action"))]
leds["focus"] = {"label": "Focus contract", "state": "ok" if not steals else "warn",
"detail": (f"quiet - {fd.get('parkedCount', 0)} parked, guards {fd.get('guardCount', 0)}"
if not steals else f"{len(steals)} self-raise(s) bounced recently"),
"_hint": "Bridge-owned KiCad windows always open in the background; self-raises are "
"bounced and logged. Your own KiCad launches are never touched."}
idle = ks.get("userIdleSeconds")
if idle is None:
leds["presence"] = {"label": "User presence", "state": "off", "detail": "unknown",
"_hint": "Presence needs a live bridge; it gates window automation."}
elif idle < 600:
leds["presence"] = {"label": "User presence", "state": "warn",
"detail": f"USER ACTIVE ({int(idle)}s idle)",
"_hint": "Window-spawning automation defers while a human is at the box. "
"Headless kicad-cli work (exports, DRC/ERC) is exempt."}
else:
leds["presence"] = {"label": "User presence", "state": "ok",
"detail": f"idle {int(idle // 60)} min",
"_hint": "Idle long enough that the daily suite and window automation may run."}
if DEV_MODE:
leds["suite"] = _suite_led()
settings = (ab(target, "kicad_get_settings", {}, timeout=T) or {}).get("settings") or {}
return {"target": target, "targets": targets, "leds": _apply_activity(leds),
**rollup_state(_apply_activity(leds)),
"settings": settings,
"windows": [w.get("title") for w in (ks.get("windows") or [])],
"bridgeVersion": (ks.get("bridgeVersion") or st.get("bridgeVersion")
or r.get("bridgeVersion") or cu.get("bridgeVersion")),
"dashboardVersion": VERSION, "generatedAt": time.strftime("%H:%M:%S"),
"_hint": "Every LED carries its own _hint naming the verb that fixes it. "
"POST /api/action {action,args} to act; GET /api/status to re-read; "
"subscribe to /events (SSE) for live pushes; POST /ui/toast to narrate on the app."}
# ---------------------------------------------------------------- activity
# John, 2026-08-18: "while you're opening kicad show an LED indicator that it's
# opening so the user knows there's background activity rather than just showing
# a dead LED". Every long action registers here, the matching LED switches to a
# pulsing `busy` state with a live label, and a state frame is pushed at once so
# the page reacts immediately instead of waiting for the 15s poll.
_activity: dict = {}
_activity_lock = threading.Lock()
def set_activity(led, detail, target=None):
with _activity_lock:
_activity[led] = {"detail": detail, "t": time.time()}
_push_state(target)
def clear_activity(led, target=None):
with _activity_lock:
_activity.pop(led, None)
_push_state(target)
def _apply_activity(leds):
with _activity_lock:
live = dict(_activity)
for led, info in live.items():
base = leds.get(led) or {"label": led.title(), "_hint": ""}
leds[led] = {**base, "state": "busy", "detail": info["detail"],
"_hint": "Working on it now - this LED is live, not stale. "
+ (base.get("_hint") or "")}
return leds
def _push_state(target=None):
"""Recompute and broadcast, so a busy LED appears the instant work starts."""
def _go():
try:
HUB.broadcast("state", cached_status(target or "-", force=True))
except Exception:
pass
threading.Thread(target=_go, daemon=True).start()
_cache = {"t": 0.0, "target": None, "data": None}
_cache_lock = threading.Lock()
def cached_status(target, force=False):
with _cache_lock:
if (not force and _cache["data"] and _cache["target"] == target
and time.time() - _cache["t"] < 8):
return _cache["data"]
data = build_status(target)
with _cache_lock:
_cache.update(t=time.time(), target=target, data=data)
HUB.broadcast("state", data)
return data
_targets_cache = {"t": 0.0, "rows": None}
def probe_targets():
"""Names first, detail second. This endpoint gates the page's first paint,
so it must never wait on a wedged box: a 10s per-box ceiling, results cached
for 30s, and a box that does not answer still appears (just without detail)."""
if _targets_cache["rows"] and time.time() - _targets_cache["t"] < 30:
return {"targets": _targets_cache["rows"], "cached": True,
"_hint": "Cached for 30s; the page paints from names alone."}
rows = []
for name in get_targets():
st = ab(name, "kicad_state", {"includeImages": False}, timeout=10)
ks = st.get("kicadState") or st
wins = ks.get("windows") or []
rows.append({"name": name, "kicadRunning": bool(wins), "windows": len(wins),
"kicadVersion": ks.get("kicadVersion"),
"responding": not bool(st.get("error"))})
_targets_cache.update(t=time.time(), rows=rows)
return {"targets": rows,
"_hint": "Pick where KiCad should run. The bridge only drives instances IT launched, "
"so a KiCad the user opened on any box is never parked, bounced, or badged."}
# -------------------------------------------------------------------- actions
def part_tour(target, args):
"""Clone a wiki part page, push its CAD files to the box, install symbol +
footprint + 3D model, then generate a single-component project."""
page = (args or {}).get("page") or "ray/xl555"
steps = []
tmp = tempfile.mkdtemp(prefix="kdash-part-")
try:
p = subprocess.run(["adom-wiki", "repo", "clone", page, "--dir", tmp],
capture_output=True, text=True, timeout=180)
if p.returncode != 0:
return {"success": False, "error": f"clone failed: {p.stderr[:200]}",
"_hint": "Give an owner/slug page reference that exists, e.g. ray/xl555."}
found = {".kicad_sym": None, ".kicad_mod": None, ".step": None}
for root, _dirs, names in os.walk(tmp):
for n in names:
for ext in found:
if n.lower().endswith(ext) and not found[ext]:
found[ext] = os.path.join(root, n)
if not found[".kicad_sym"] and not found[".kicad_mod"]:
return {"success": False, "error": "page ships no .kicad_sym or .kicad_mod",
"_hint": "This tour needs a part page carrying KiCad CAD files."}
steps.append("cloned " + page)
send = [f for f in found.values() if f]
sent = ab(target, "send_files", {"filePaths": send}, timeout=180)
dest = sent.get("destinationPaths") or []
if not dest:
return {"success": False, "steps": steps,
"error": f"send_files failed: {str(sent.get('error'))[:150]}"}
by_ext = {os.path.splitext(d)[1].lower(): d for d in dest}
steps.append(f"sent {len(dest)} file(s) to the box")
part = None
if by_ext.get(".kicad_sym"):
r = ab(target, "kicad_install_symbol", {"filePath": by_ext[".kicad_sym"]}, timeout=180)
part = r.get("symbolName") or r.get("partName")
steps.append("installed symbol" + (f" ({part})" if part else ""))
if by_ext.get(".kicad_mod"):
fargs = {"filePath": by_ext[".kicad_mod"]}
if by_ext.get(".step"):
fargs["model"] = {"filePath": by_ext[".step"]}
ab(target, "kicad_install_footprint", fargs, timeout=180)
steps.append("installed footprint" + (" + 3D model" if by_ext.get(".step") else ""))
if not part:
part = os.path.basename(page)
proj = ab(target, "kicad_make_part_project", {"partName": part}, timeout=180)
if proj.get("success") is False:
return {"success": False, "steps": steps,
"error": f"project generation failed: {str(proj.get('error'))[:150]}"}
steps.append("generated project " + str(proj.get("projectDir") or part))
return {"success": True, "steps": steps, "part": part, "project": proj.get("project"),
"_hint": "Open it with kicad_open_project, or show any of the eight surfaces "
"with the kicad_show_* verbs - all in the background."}
finally:
shutil.rmtree(tmp, ignore_errors=True)
# ---------------------------------------------------------------- surfaces
# The per-part surface ladder, ported from ab's web-control demo
# (adom/adom-bridge web-control-demo.html): ONE real wiki part, eight surfaces,
# each fetching the part's files from its wiki page, installing them into the
# user's real KiCad, opening JUST that surface in the background, and rendering
# it from kicad_state. The whole circle: wiki component -> the app the EE
# designs in.
PART = {"page": "adom/in-s42atr", "name": "IN-S42ATR", "lib": "Adom"}
SURFACES = [
("symbol", "Symbol", "kicad_show_symbol", "part", "symbol editor",
"installs the part from its wiki page, then shows JUST the symbol"),
("footprint", "Footprint", "kicad_show_footprint", "part", "footprint editor",
"just the land pattern: pads + courtyard"),
("3d-chip", "3D chip", "kicad_show_3d_chip", "part", "3d viewer",
"the part's STEP model on its footprint"),
("library", "Library", "kicad_show_library", "part", "symbol editor",
"the library holding the symbol, footprint and 3D model"),
("schematic", "Schematic", "kicad_show_schematic", "project", "schematic editor",
"the generated single-component schematic"),
("2d-board", "2D board", "kicad_show_2d_board", "project", "pcb editor",
"the part-dominant board, tight edge cuts"),
("3d-board", "3D board", "kicad_show_3d_board", "project", "3d viewer",
"the populated board in 3D"),
("project", "Project", "kicad_show_project", "project", "kicad 10",
"the Project Manager: schematic + board hang off it"),
]
_stage_lock = threading.Lock()
_staged = {"part": None, "project": None} # per-server-run cache, like ab's
def _kicad_hwnd(target):
"""Main KiCad window hwnd, for taskbar progress."""
try:
st = ab(target, "kicad_state", {"includeImages": False}, timeout=60)
for w in ((st.get("kicadState") or st).get("windows") or []):
if w.get("hwnd"):
return int(w["hwnd"])
except Exception:
pass
return None
def taskbar_progress(target, value=None, state="normal", hwnd=None):
"""John, 2026-08-18: "when you automate kicad and you know something is going
to take a while, use ab's progress bar indicator in the windows taskbar".
Paints on the KiCad button itself, so the wait is legible without the user
watching this page: indeterminate while staging, a real percentage as steps
land, and cleared (or red) when the job ends."""
h = hwnd or _kicad_hwnd(target)
if not h:
return None
prog = {"state": state}
if value is not None:
prog["value"] = int(value)
ab(target, "desktop_taskbar", {"hwnd": h, "progress": prog}, timeout=30)
return h
def _emit(surface, **kw):
"""Broadcast to the page AND record in /console: an AI driving this app reads
the console back, so surface progress has to land in both places."""
if kw.get("log"):
log(f"[{surface}] {kw['log']}")
HUB.broadcast("surface", {"surface": surface, **kw})
def stage_part(target, surface):
"""Fetch the part's CAD files from its wiki page and install them."""
with _stage_lock:
if _staged["part"]:
_emit(surface, log="part already staged this session (reusing)")
return _staged["part"]
_emit(surface, log=f"staging {PART['name']} from {PART['page']}", percent=10)
tmp = tempfile.mkdtemp(prefix="kdash-surface-")
try:
p = subprocess.run(["adom-wiki", "repo", "clone", PART["page"], "--dir", tmp],
capture_output=True, text=True, timeout=180)
if p.returncode != 0:
raise RuntimeError(f"wiki clone failed: {p.stderr[:150]}")
found = {}
for root, _d, names in os.walk(tmp):
for n in names:
for ext in (".kicad_sym", ".kicad_mod", ".step"):
if n.lower().endswith(ext) and ext not in found:
found[ext] = os.path.join(root, n)
if ".kicad_sym" not in found or ".kicad_mod" not in found:
raise RuntimeError("the wiki page ships no .kicad_sym / .kicad_mod")
_emit(surface, log=f"sending {len(found)} file(s) to the box", percent=25)
sent = ab(target, "send_files", {"filePaths": list(found.values())}, timeout=180)
dest = {os.path.splitext(d)[1].lower(): d for d in (sent.get("destinationPaths") or [])}
if not dest:
raise RuntimeError(f"send_files failed: {str(sent.get('error'))[:120]}")
_emit(surface, log="installing the symbol", percent=40)
r = ab(target, "kicad_install_symbol",
{"filePath": dest[".kicad_sym"], "quietInstall": True}, timeout=180)
if r.get("libraryName"):
PART["lib"] = r["libraryName"]
# Use the names the bridge ACTUALLY installed, not the wiki page's
# name. The in-file symbol name is frequently not the filename
# (tja1051t-3-118 installs as TJA1051T_3,118) - honouring the
# returned symbolNames is what fixed four parts on the web-control
# side (#42/#49), and this dashboard never picked that up.
if r.get("symbolNames"):
PART["symbol"] = r["symbolNames"][0]
if r.get("staleEditorOpen"):
_emit(surface, log="note: a Symbol Editor was already open - the bridge "
"will re-index it automatically when we show the part")
_emit(surface, log=f"installing the footprint (+3D model) into '{PART['lib']}'", percent=55)
fargs = {"filePath": dest[".kicad_mod"], "quietInstall": True}
if dest.get(".step"):
fargs["model"] = {"filePath": dest[".step"]}
fr = ab(target, "kicad_install_footprint", fargs, timeout=180)
if fr.get("footprintName"):
PART["footprint"] = fr["footprintName"]
if fr.get("staleEditorOpen"):
_emit(surface, log="note: a Footprint Editor was already open - showing the "
"part will restart KiCad to rebuild its cache (~1 min)")
_staged["part"] = {"lib": PART["lib"]}
return _staged["part"]
finally:
shutil.rmtree(tmp, ignore_errors=True)
def stage_project(target, surface):
"""The part, plus the generated single-component project."""
stage_part(target, surface)
with _stage_lock:
if _staged["project"]:
return _staged["project"]
_emit(surface, log="generating the part's demo project", percent=70)
r = ab(target, "kicad_make_part_project", {"partName": PART["name"]}, timeout=240)
if r.get("success") is False:
raise RuntimeError(f"kicad_make_part_project: {str(r.get('error'))[:120]}")
_staged["project"] = {"pro": r.get("project"), "sch": r.get("schematic"),
"pcb": r.get("board")}
return _staged["project"]
def _surface_args(key, staged_project):
lib = PART["lib"]
name = PART.get("symbol") or PART["name"]
fp_name = PART.get("footprint") or PART["name"]
pj = staged_project or {}
return {
"symbol": {"symbolName": name, "libraryName": lib},
"footprint": {"footprintName": fp_name, "library": lib},
"3d-chip": {"footprintName": fp_name, "library": lib},
# symbolName routes show_library through the VERIFIED load path (the
# #49 fix made it honoured; before that it was silently dropped). In
# bare browse mode the verb cannot verify a title, _render then grabbed
# ANY symbol editor, and this card proudly rendered
# AdomRP2040:ADOM_MECHANICAL_PIN as proof of the Adom library - the
# wrong-library-under-a-green-tick bug, on the one surface #49 missed.
"library": {"libraryName": lib, "symbolName": name},
"schematic": {"filePath": pj.get("sch")},
"2d-board": {"filePath": pj.get("pcb")},
"3d-board": {"filePath": pj.get("pcb")},
"project": {"filePath": pj.get("pro")},
}[key]
def ensure_ready(target, surface):
"""A click is a GOAL, not a request for an audit (the Adom automation
doctrine). If KiCad is not running, launch it and wait - never hand the user
an error they would fix by pressing a button we could have pressed."""
st = ab(target, "kicad_state", {"includeImages": False}, timeout=30)
if (st.get("kicadState") or st).get("windows"):
return True
_emit(surface, log="KiCad is closed - launching it in the background first", percent=15)
ab(target, "kicad_launch", {}, timeout=120)
for _ in range(20): # up to ~60s for the first window
time.sleep(3)
st = ab(target, "kicad_state", {"includeImages": False}, timeout=30)
if (st.get("kicadState") or st).get("windows"):
_emit(surface, log="KiCad is up", percent=25)
return True
return False
def _live_progress_poller(target, surface, stop_ev):
"""While a surface verb runs, relay the BRIDGE's live progress frames
(kicad_progress active[]) onto the card's SSE channel.
This is the piece John flagged missing (2026-08-21): the card used to show
only the dashboard's own coarse percent emits (5/40/80), while the bridge
was publishing real measured frames - stepLabel, etaSec, confidence,
overdue - that only web control rendered. One poll every 2.5s, read-only,
and the frames carry `live:true` so the UI can style them (dim non-measured,
amber overdue/blocked, marquee when percent is null - a guessed number
presented as a measurement is the failure mode the timing system kills)."""
while not stop_ev.wait(2.5):
try:
r = ab(target, "kicad_progress", {"all": True}, timeout=20)
frames = r.get("active") or []
if not frames:
continue
f = frames[0]
_emit(surface, live=True,
percent=f.get("percent"), stepLabel=f.get("stepLabel"),
etaSec=f.get("etaSec"), confidence=f.get("confidence"),
overdue=bool(f.get("overdue")), blockedBy=f.get("blockedBy"))
except Exception:
pass
def run_surface(target, key):
"""Stage what this surface needs, open it in the BACKGROUND, then render it
from kicad_state. Streams progress over SSE and onto the taskbar button."""
spec = next((s for s in SURFACES if s[0] == key), None)
if key == "state":
spec = ("state", "KiCad state", "kicad_state", None, None,
"captures every KiCad window as a screenshot")
if not spec:
_emit(key, status="error", log=f"unknown surface {key}")
return
_key, label, verb, need, pick, note = spec
hwnd = None
try:
_emit(key, status="running", log=note, percent=5)
set_activity("kicad", f"opening the {label.lower()} in the background...", target)
hwnd = taskbar_progress(target, None, "indeterminate")
if not ensure_ready(target, key):
raise RuntimeError("KiCad did not come up within 60s - try Launch KiCad, "
"or check the KiCad LED for an install problem")
project = None
if need == "part":
stage_part(target, key)
elif need == "project":
project = stage_project(target, key)
taskbar_progress(target, 80, "normal", hwnd)
_emit(key, log=f"opening the {label.lower()} in the background", percent=80)
args = _surface_args(key, project) if need else {"maxWidth": 900}
if need:
args["capture"] = True # the #45 shot contract: the verb
args["captureMaxWidth"] = 900 # hands us the frame, no 2nd trip
_stop = threading.Event()
threading.Thread(target=_live_progress_poller, args=(target, key, _stop),
daemon=True).start()
try:
r = ab(target, verb, args, timeout=300)
finally:
_stop.set()
# RECOVERY, branching on KEYS never message text (Colby's client shape
# from #47, which this dashboard never picked up - web control healed
# these invisibly while this card surfaced call-1 failures raw).
attempts = 0
while r.get("success") is False and r.get("retryable") and attempts < 2:
attempts += 1
wait_s = max(2.0, float(r.get("retryAfterMs") or 15000) / 1000.0)
_emit(key, log=f"KiCad restarted to rebuild its library cache - "
f"retrying in {int(wait_s)}s (measured)", percent=70)
time.sleep(wait_s)
_stop = threading.Event()
threading.Thread(target=_live_progress_poller, args=(target, key, _stop),
daemon=True).start()
try:
r = ab(target, verb, args, timeout=300)
finally:
_stop.set()
if r.get("success") is False:
if r.get("errorCode") == "library_cache_stale" and r.get("rescanRefused"):
raise RuntimeError("a KiCad window you opened yourself is in the way - "
"close KiCad, then click this card again")
raise RuntimeError(str(r.get("error") or r.get("errorCode"))[:160])
if r.get("autoRescanned"):
_emit(key, log="the library index predated this part, so the bridge "
"re-indexed - that is why this took about a minute")
# Prefer the verb's own shot (same window that satisfied the verb, so
# the evidence cannot disagree with the verdict); kicad_state is the
# fallback for surfaces that did not return one.
sh = r.get("shot") or {}
shot, title = sh.get("thumbnailDataUrl"), sh.get("title")
if shot and shot.startswith("data:"):
shot = shot.split(",", 1)[1] # _emit sends bare base64
ev = r.get("evidence") or {}
if ev.get("canvasRendered") is False:
_emit(key, log="note: the editor is on the part but its canvas has "
"not painted yet (KiCad still loading libraries)")
if not shot:
_emit(key, log="rendering it from kicad_state", percent=92)
shot, title = _render(target, pick)
taskbar_progress(target, 100, "none", hwnd)
if not shot:
# ralph iter-4: the job used to report success with no picture at
# all, which is the one thing this card exists to show.
_emit(key, status="error", percent=100,
log=f"{label} opened, but no window could be captured to render it")
return
_emit(key, status="done", percent=100, image=shot, title=title,
log=f"{label} shown in the background" + (f" - {title}" if title else ""))
except Exception as e:
if hwnd:
taskbar_progress(target, 100, "error", hwnd)
_emit(key, status="error", log=str(e)[:200])
finally:
clear_activity("kicad", target)
def _render(target, pick):
"""Pull this surface's window out of kicad_state as a base64 PNG.
ralph iter-5: the captures live in state.SHOTS (each {title, image, hwnd,
path}), NOT in state.windows - windows carries geometry only, so the render
always came back empty and the card had nothing to show. `includeImages` is
what produces shots at all."""
st = ab(target, "kicad_state", {"includeImages": True, "maxWidth": 900}, timeout=240)
ks = st.get("kicadState") or st
shots = ks.get("shots") or []
# EVIDENCE MUST BE THE THING THAT WAS ASKED FOR (John, 2026-08-19: "i had
# asked for a footprint and the evidence you provided was just of the project
# window"). The old fallback grabbed the last window on screen when the
# expected one was missing, which turns a failure into a screenshot of
# something else entirely - worse than no evidence, because it looks like
# proof. If the requested surface is not there, say so.
if pick:
chosen = next((w for w in shots if pick in (w.get("title") or "").lower()), None)
if not chosen:
seen = ", ".join((w.get("title") or "?")[:40] for w in shots) or "nothing"
raise RuntimeError(f"the {pick} window never appeared - on screen instead: {seen}")
else:
chosen = shots[-1] if shots else None
if not chosen:
return None, None
img = (chosen.get("image") or chosen.get("screenshot")
or chosen.get("imageBase64") or chosen.get("png"))
return img, chosen.get("title")
def _launch_watched(target):
"""Launch with a live LED: 'launching...' then 'waiting for the first
window...' then back to the real status once it is up."""
try:
set_activity("kicad", "launching KiCad in the background...", target)
r = ab(target, "kicad_launch", {}, timeout=180)
set_activity("kicad", "waiting for the first window...", target)
for _ in range(20):
time.sleep(3)
st = ab(target, "kicad_state", {"includeImages": False}, timeout=25)
if ((st.get("kicadState") or st).get("windows")):
break
set_activity("plugin", "KiCad is up - waiting for the plugin to bind...", target)
for _ in range(10):
time.sleep(6)
bs = ab(target, "kicad_bridge_status", {}, timeout=25)
if (bs.get("summary") or {}).get("alive"):
break
return r
finally:
clear_activity("kicad", target)
clear_activity("plugin", target)
def do_action(target, name, args):
args = args or {}
if name == "launch":
threading.Thread(target=_launch_watched, args=(target,), daemon=True).start()
return {"success": True, "started": "launch",
"_hint": "Launching in the background; the KiCad and Plugin LEDs show live "
"progress and the page updates over /events as it happens."}
if name == "close":
# Bridge 0.9.347: a plain close leaves an editor with unsaved work open and
# answers errorCode unsaved_changes; the page asks the user before it
# sends discard:true. force is never used from here (it kills silently).
set_activity("kicad", "closing the bridge's KiCad windows...", target)
try:
payload = {"discardChanges": True} if args.get("discard") else {}
return ab(target, "kicad_close", payload, timeout=180)
finally:
clear_activity("kicad", target)
if name == "demo":
return ab(target, "kicad_demo", {"all": True, "background": True,
"muted": bool(args.get("muted"))})
if name == "demo_progress":
return ab(target, "kicad_demo", {"progress": True}, timeout=60)
if name == "upgrade":
set_activity("update", "installing the latest KiCad (silent, no UAC)...", target)
try:
return ab(target, "kicad_upgrade", {}, timeout=600)
finally:
clear_activity("update", target)
if name == "install_plugin":
return ab(target, "kicad_install_plugin", {})
if name == "set_overlay_badges":
return ab(target, "kicad_set_settings",
{"overlayBadges": bool(args.get("overlayBadges", True))}, timeout=60)
if name == "set_background_guard":
# The A/B switch (John, 2026-09-21): off = the bridge never pushes or moves a KiCad window.
# Applies at once on the box; needs bridge 1.0.42 or newer (older bridges store the key and
# ignore it, so say so instead of toasting a success that changes nothing).
r = ab(target, "kicad_set_settings",
{"backgroundGuard": bool(args.get("backgroundGuard", True))}, timeout=60)
ver = str((r or {}).get("bridgeVersion") or "0")
try:
too_old = tuple(int(x) for x in ver.split(".")[:3]) < (1, 0, 42)
except ValueError:
too_old = False
if too_old:
return {"success": False, "error": f"the KiCad bridge on this box is {ver}; this switch needs 1.0.42 or newer"}
return r
if name == "set_activity_indicator":
# Fusion parity (John, 2026-08-22). Its code review caught fusion's
# dashboard shipping this control with NO action arm - the click
# toasted success while changing nothing. Wire the arm before the UI.
mode = str(args.get("mode") or "progress")
if mode not in ("progress", "flash", "both", "none"):
return {"success": False, "error": f"unknown mode {mode}"}
return ab(target, "kicad_set_settings", {"activityIndicator": mode}, timeout=60)
if name == "say":
return {"success": True, "phrases": say_phrases()}
if name == "try_phrase":
# RETIRED 2026-08-24: superseded by Send to AI (say_send: one call, new
# threads only). Body DELETED rather than dormant because it contained
# an /ai-threads/resume call, and resume stays parked by ruling.
return {"success": False, "retired": True,
"error": "try_phrase is retired. Use the dashboard's Send to AI "
"(POST /api/say-send), which opens a NEW thread and submits.",
"_hint": "Resuming or typing into EXISTING threads remains parked."}
if name == "restore_settings":
return ab(target, "kicad_set_settings", {"restoreDefaults": True}, timeout=60)
if name == "part_tour":
return part_tour(target, args)
if name == "surface":
key = (args.get("surface") or "").strip()
threading.Thread(target=run_surface, args=(target, key), daemon=True,
name=f"surface-{key}").start()
return {"success": True, "started": key,
"_hint": "Progress streams on /events as `surface` events (status, log, "
"percent, image) and on the KiCad taskbar button's progress bar."}
if name == "surfaces":
return {"success": True, "part": PART,
"surfaces": [{"key": k, "label": lbl, "note": note}
for k, lbl, _v, _n, _p, note in SURFACES]}
return {"error": f"unknown action {name}",
"_hint": "Actions: launch, close, demo, demo_progress, upgrade, install_plugin, "
"set_overlay_badges, set_background_guard, part_tour."}
# --------------------------------------------------------------------- server
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
server_version = "kicad-dashboard/" + VERSION
def log_message(self, *a):
pass
def _send(self, code, ctype, body, extra=None):
if isinstance(body, str):
body = body.encode("utf-8")
self.send_response(code)
self.send_header("Content-Type", ctype)
self.send_header("Content-Length", str(len(body)))
self.send_header("Access-Control-Allow-Origin", "*")
for k, v in (extra or {}).items():
self.send_header(k, v)
self.end_headers()
self.wfile.write(body)
def _json(self, obj, code=200):
self._send(code, "application/json", json.dumps(obj))
def _read_body(self):
try:
n = int(self.headers.get("Content-Length") or 0)
return json.loads(self.rfile.read(n) or b"{}") if n else {}
except Exception:
return {}
def do_GET(self):
path = urlparse(self.path).path.rstrip("/") or "/"
if path.endswith("/newbie-panel.js"):
# The ONE shared panel implementation (adom-ui-design 2.0.0). Served
# verbatim, never forked: fixes land in the pack, not here.
try:
return self._send(200, "application/javascript; charset=utf-8",
(UI_DIR / "newbie-panel.js").read_bytes(),
{"Cache-Control": "no-store"})
except Exception as e:
return self._json({"error": f"module missing: {e}"}, 500)
if path.endswith(("/favicon.svg", "/favicon.png", "/favicon.ico",
"/dock-kicad.svg", "/kicad-logo.png")):
# endswith: the adom.cloud proxy mounts the app under a sub-path.
try:
return self._send(200, "image/svg+xml", LOGO_PATH.read_bytes())
except Exception:
return self._json({"error": "logo missing"}, 404)
if path == "/health":
# pid published so a SUCCESSOR can retire us directly instead of
# inferring ownership. Older builds do not send it, which is why
# _pid_owning_port() exists and does not depend on this.
return self._json({"ok": True, "app": "kicad-dashboard", "version": VERSION,
"pid": os.getpid(), "sseClients": HUB.count()})
am = re.search(r"/assets/(.+)$", path)
if am:
# Settings screenshots + the shared Adom cog. Real captures only
# (settings-layout skill); anything else under here 404s. Matched
# anywhere in the path because the adom.cloud proxy mounts the
# app under a sub-path.
rel = am.group(1)
f = (UI_DIR / "assets" / rel).resolve()
if not str(f).startswith(str((UI_DIR / "assets").resolve())) or not f.is_file():
return self._json({"error": "no such asset"}, 404)
mime = {"svg": "image/svg+xml", "png": "image/png", "jpg": "image/jpeg",
"webp": "image/webp"}.get(f.suffix.lstrip(".").lower(),
"application/octet-stream")
return self._send(200, mime, f.read_bytes(),
{"Cache-Control": "no-cache"})
if path.endswith("/ai-threads/run-prompt/status"):
# Hydrogen 1.0.187+ (newbie-panel.js 2.22.0): POST run-prompt returns a
# job id in under a second and the module polls THIS route once a
# second for opening -> prefilled -> submitted -> working -> done.
# The query string (?job=<uuid>) must travel intact: a proxy that
# forwards only the path answers about no job at all, and every poll
# 404s. Same verbatim pass-through as the other two routes.
q = urlparse(self.path).query or ""
return _proxy_control(self, "/ai-threads/run-prompt/status" + (("?" + q) if q else ""),
timeout=8)
if path.endswith("/updater/status"):
# LOST IN AN EARLIER REBUILD, which meant the module's version
# gate silently no-opped: its gate() catches a failure and
# returns open, so a missing proxy reads as "allowed".
return _proxy_control(self, "/updater/status", timeout=8)
if path.endswith("/api/viewport"):
if "matrix" in (urlparse(self.path).query or ""):
return self._json({"reports": sorted(VIEWPORT_LOG,
key=lambda r: r.get("innerWidth") or 0)})
return self._json(LAST_VIEWPORT or {"note": "no page has reported a viewport yet"})
if path == "/console":
with _console_lock:
return self._json({"lines": list(CONSOLE)})
if path == "/events":
return self._sse()
if path == "/api/targets":
return self._json(probe_targets())
m = re.match(r"^/api/verb_times(?:/([^/]+))?$", path)
if m:
# The verb-times panel (fusion parity). Proxied per target so the
# numbers are THAT box's measured history, not this container's.
t = unquote(m.group(1) or "") or (get_targets() or ["-"])[0]
r = ab(t, "kicad_verb_times", {}, timeout=30)
return self._json({"target": t, "verbs": r.get("verbs") or {},
"error": r.get("error") or None})
m = re.match(r"^/(?:api/status|state)(?:/([^/]+))?(/force)?$", path)
if m:
target = unquote(m.group(1) or "-")
force = bool(m.group(2))
log(f"status target={target}{' force' if force else ''}")
return self._json(cached_status(target, force))
if not _wants_app_shell(self, path):
# NOT a catch-all any more. An unmatched asset, an unknown route in a
# namespace we own, or any non-browser client gets a real 404, so the
# launch-console 404 watch can actually fire on a virgin install.
return self._json({"error": "not found", "path": path}, 404)
try: # a real browser navigation renders the app (proxy sub-paths still work)
# no-store: the dock's Hard Refresh Open must never get a cached
# shell after a version bump (dockbar guide).
# Inject the phrase list at SERVE time. It comes from dockbar.json,
# which is static config, so making the panel wait on an /api round
# trip only created a boot race: on first paint the target is not
# resolved yet, the call came back empty, and the page rendered NO
# panel at all - the same "panel with no prompts" shape the fleet
# audit kept finding elsewhere. Static data belongs in the document.
html = (UI_DIR / "index.html").read_text(encoding="utf-8")
html = html.replace("</head>",
"<script>window.__SAY_PHRASES__=" +
json.dumps(say_phrases()) + ";</script></head>", 1)
return self._send(200, "text/html; charset=utf-8", html,
{"Cache-Control": "no-store, must-revalidate"})
except Exception as e:
return self._json({"error": f"ui missing: {e}"}, 500)
def do_OPTIONS(self):
# CORS preflight on the routes we ACTUALLY serve (dockbar guide): the dock
# card and the webview both call cross-origin. This used to answer 204 for
# every path, which made the module's checkSend "send route wired" row
# green for routes that do not exist - it probes with OPTIONS, so a blanket
# preflight answered the probe instead of the route.
path = urlparse(self.path).path.rstrip("/") or "/"
if not _route_exists(path):
return self._json({"error": "not found", "path": path}, 404)
self.send_response(204)
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Access-Control-Allow-Methods", "GET, POST, OPTIONS")
self.send_header("Access-Control-Allow-Headers", "Content-Type")
self.send_header("Access-Control-Max-Age", "86400")
self.end_headers()
def do_POST(self):
path = urlparse(self.path).path.rstrip("/") or "/"
body = self._read_body()
if path == "/api/action":
name = body.get("action") or ""
target = body.get("target") or (get_targets() or ["-"])[0]
log(f"action {name} target={target}")
return self._json(do_action(target, name, body.get("args")))
if path == "/ui/toast":
msg, kind = body.get("message") or "", body.get("kind") or "info"
log(f"toast [{kind}] {msg}")
HUB.broadcast("toast", {"message": msg, "kind": kind})
return self._json({"ok": True, "delivered": HUB.count(),
"_hint": "delivered = live SSE viewers that showed it. "
"0 means nobody has the page open - open it first."})
if path.endswith("/api/viewport"):
# The page reports the width of the surface it actually rendered in.
# Assumption is how the 760px breakpoint bug survived: John's dock
# pane is 744px and nobody's test matrix included it.
global LAST_VIEWPORT
LAST_VIEWPORT = dict(body or {})
LAST_VIEWPORT["at"] = time.strftime("%H:%M:%S")
VIEWPORT_LOG.append(dict(LAST_VIEWPORT))
del VIEWPORT_LOG[:-16]
log(f"viewport {LAST_VIEWPORT.get('innerWidth')}px cols={LAST_VIEWPORT.get('panelCols')}")
return self._json({"ok": True})
if path.endswith("/ai-threads/run-prompt"):
# Wiki #77: the shared newbie-panel module posts here. Strip any `type`
# (absent = ask the user via Hydrogen's AI picker) and carry the app:
# tag so a remembered pick applies to KiCad. The response goes back
# verbatim: cancelled:true, delivered:false and the picker's 45 s
# block are the module's to render.
b = dict(body or {}); b.pop("type", None)
b.setdefault("background", True) # Hydrogen 1.0.219: immediate {job, state}; the module polls status
tags = list(b.get("tags") or [])
if "app:adom/kicad-bridge" not in tags:
tags.append("app:adom/kicad-bridge")
b["tags"] = tags
return _proxy_control(self, "/ai-threads/run-prompt", body=b, timeout=125)
if path.endswith("/api/say-send"):
phrase = str((body or {}).get("phrase") or "").strip()
if not phrase:
return self._json({"ok": False, "why": "phrase required"}, 400)
log(f"say-send: {phrase[:60]}")
return self._json(say_send(phrase))
if path == "/ui/click":
t = body.get("target") or ""
log(f"ui click {t}")
HUB.broadcast("click", {"target": t})
return self._json({"ok": True, "delivered": HUB.count(),
"_hint": "Targets: setup, launch, close, demo, part-tour, settings, refresh. "
"The PAGE performs the click, so what the AI drives is what the user gets."})
return self._json({"error": "not found",
"_hint": "POST /api/action {action,args}, /ui/toast {message,kind}, /ui/click {target}"}, 404)
def _sse(self):
"""Headers immediately, then one flush PER EVENT. The 2KB comment padding
plus no-transform / X-Accel-Buffering defeat proxy buffering (the adom.cloud
proxy otherwise holds every byte until the stream ends)."""
self.send_response(200)
self.send_header("Content-Type", "text/event-stream")
self.send_header("Cache-Control", "no-cache, no-transform")
self.send_header("X-Accel-Buffering", "no")
self.send_header("Access-Control-Allow-Origin", "*")
self.send_header("Connection", "keep-alive")
self.end_headers()
q = HUB.add()
log(f"sse client connected ({HUB.count()} live)")
try:
self.wfile.write((": " + "p" * 2048 + "\n\n").encode())
self.wfile.write(f"event: hello\ndata: {json.dumps({'version': VERSION})}\n\n".encode())
self.wfile.flush()
while True:
try:
msg = q.get(timeout=15)
except queue.Empty:
msg = ": ping\n\n"
self.wfile.write(msg.encode())
self.wfile.flush()
except Exception:
pass
finally:
HUB.drop(q)
log(f"sse client disconnected ({HUB.count()} live)")
# ------------------------------------------------------------------------ CLI
def claim_dock_webview(port):
"""Take ownership of the dock's pre-opened webview tab (dock contract,
2026-08-23, blessed as THE model: the webview belongs to the CLI).
The dock sets ADOM_DOCK_WEBVIEW to the tab name it opened for this app.
open-or-refresh with the same --name is idempotent - re-running it just
navigates the existing tab, which is the ownership handle: we renavigate
after restarts and repaint on state changes. Unset env (manual serve) or
a failed call are both silent no-ops; the dock's own interstitial
navigation remains the fallback either way."""
name = os.environ.get("ADOM_DOCK_WEBVIEW") or ""
if not name:
return
try:
subprocess.run(["adom-cli", "hydrogen", "webview", "open-or-refresh",
"--name", name, "--url", direct_url(port)],
capture_output=True, timeout=20)
log(f"claimed dock webview tab '{name}' -> {direct_url(port)}")
except Exception:
pass
def direct_url(port):
"""What `serve --print-url` prints: the DIRECT port, always.
Dock-bar thread request (2026-08-23): the consumer of --print-url is the
Hydrogen dock, whose workspace shares the port namespace (WSL2 mirrored
networking locally; container-side probe in cloud), so the /proxy/<port>/
hop is pure indirection - and its prefix is what caused the dock's
health-URL join bugs. 127.0.0.1 rather than the NAME localhost, because
Windows resolves localhost IPv6-first while WSL binds IPv4 only: measured
~820ms per request vs 13ms, most of the dock's 2000ms probe budget."""
return f"http://127.0.0.1:{port}/"
def proxy_url(port):
tpl = os.environ.get("VSCODE_PROXY_URI") or ""
if "{{port}}" in tpl:
return tpl.replace("{{port}}", str(port))
return f"http://127.0.0.1:{port}/"
def local_call(port, path, payload=None):
import urllib.request
url = f"http://127.0.0.1:{port}{path}"
data = json.dumps(payload).encode() if payload is not None else None
req = urllib.request.Request(url, data=data, method="POST" if data else "GET",
headers={"Content-Type": "application/json"})
try:
with urllib.request.urlopen(req, timeout=20) as r:
return json.loads(r.read().decode())
except Exception as e:
return {"error": str(e),
"_hint": f"Is the dashboard serving? `kicad-dashboard serve --port {port}` (health: /health)."}
def main():
argv = sys.argv[1:] or ["serve"]
cmd = argv[0]
def flag(name, default=None):
return argv[argv.index(name) + 1] if name in argv and argv.index(name) + 1 < len(argv) else default
port = int(flag("--port", DEFAULT_PORT))
target = flag("--target", "")
if "--dev" in argv: # maintainer view: adds the nightly-suite / cron LED
globals()["DEV_MODE"] = True
if cmd == "serve":
# IDEMPOTENT SERVE (wiki #43, John). Hydrogen runs the serve step on
# EVERY launch - `skip_if` guards the install step, not this one. So the
# second click of a session used to run serve against our own live
# instance and die on EADDRINUSE with a raw traceback in the launch
# console. An already-running dashboard should make relaunch INSTANT,
# not fatal, so: if a healthy instance of OURS already owns the port,
# print its URL and exit 0.
def _running_version(p_):
"""Version of OUR instance on the port, or None if none/not ours."""
import urllib.request # imported per-use in this module
try:
with urllib.request.urlopen(
f"http://127.0.0.1:{p_}/health", timeout=1.5) as r:
d_ = json.loads(r.read().decode("utf-8") or "{}")
# Confirm it is OURS - another service on the port is not a
# reason to claim success.
if bool(d_.get("ok")) and d_.get("app") == "kicad-dashboard":
return str(d_.get("version") or "?")
except Exception:
pass
return None
def _healthy_instance(p_):
return _running_version(p_) is not None
def _pid_owning_port(p_):
"""The pid LISTENING on p_, via /proc/net/tcp inode -> /proc/*/fd.
PORT OWNERSHIP, not argv (dockbar guide: "argv is a claim, not a
fact"). Rehearsed against a real predecessor 2026-08-25: selecting
victims by argv alone killed an UNRELATED instance serving a
different port, because every instance of this tool has identical
argv. The port is the thing being contested, so the port is what
decides who is retired."""
want = []
for fam, path in ((4, "/proc/net/tcp"), (6, "/proc/net/tcp6")):
try:
for line in open(path).read().splitlines()[1:]:
f = line.split()
if len(f) < 4 or f[3] != "0A": # 0A = LISTEN
continue
if int(f[1].split(":")[1], 16) == int(p_):
want.append(f[9]) # inode
except Exception:
pass
if not want:
return None
me = os.getpid()
for pid in os.listdir("/proc"):
if not pid.isdigit() or int(pid) == me:
continue
try:
for fd in os.listdir(f"/proc/{pid}/fd"):
try:
link = os.readlink(f"/proc/{pid}/fd/{fd}")
except Exception:
continue
if link.startswith("socket:[") and link[8:-1] in want:
return int(pid)
except Exception:
continue
return None
def _retire_stale(p_, their):
"""A healthy instance of a DIFFERENT version is not a reason to skip
serving; it is the thing to replace (dock thread, 2026-08-23).
Retire the process that OWNS THE PORT, and use argv only as a
SAFETY FILTER on that one pid - never as the selector. Selecting by
argv killed a bystander on another port in rehearsal."""
log(f"kicad-dashboard {their} is serving on {p_} but this is {VERSION}; "
f"retiring the stale instance")
victim = _pid_owning_port(p_)
if victim is None:
log(f"could not identify the process owning port {p_}; not killing anything")
return False
try:
argv_ = open(f"/proc/{victim}/cmdline", "rb").read().split(b"\0")
except Exception:
argv_ = []
ours = (len(argv_) >= 3
and os.path.basename(argv_[0]).startswith(b"python")
and argv_[1].endswith(b"kicad_dashboard.py")
and argv_[2] == b"serve")
if not ours:
log(f"port {p_} is held by pid {victim} which is not one of ours; refusing to kill")
return False
import signal
try:
os.kill(victim, signal.SIGTERM)
except Exception:
pass
for _ in range(40): # up to ~4s for the port to free
if _running_version(p_) is None:
return True
time.sleep(0.1)
try:
os.kill(victim, signal.SIGKILL)
except Exception:
pass
time.sleep(0.5)
return _running_version(p_) is None
their = _running_version(port)
if their is not None and their != VERSION:
if not _retire_stale(port, their):
print(f"kicad-dashboard: a stale {their} instance owns port {port} and "
f"would not exit; kill it and relaunch.", file=sys.stderr, flush=True)
raise SystemExit(2)
their = None
if their is not None:
if "--print-url" in argv:
print(direct_url(port), flush=True)
else:
log(f"kicad-dashboard {VERSION} already serving on {port}; nothing to do")
return
# BIND FIRST. The old order printed the URL and only then tried to bind,
# so a port clash advertised a URL nothing was listening on (wiki #43).
try:
srv = ThreadingHTTPServer(("0.0.0.0", port), Handler)
except OSError as e:
# Lost a race, or something else owns the port. Re-probe once: if it
# is us after all, this launch is still a success.
if _healthy_instance(port):
if "--print-url" in argv:
print(direct_url(port), flush=True)
return
# A readable one-liner beats a traceback in the launch console.
print(f"kicad-dashboard: cannot serve on port {port}: {e}. "
f"Something else owns it - pick another with --port.",
file=sys.stderr, flush=True)
raise SystemExit(2)
srv.daemon_threads = True
if "--print-url" in argv:
# The Hydrogen dock contract (adom/hydrogen-dock-sdk): print the URL
# and SELF-BACKGROUND, so the workspace exec returns promptly.
# Hydrogen deliberately does not detach for us - it runs this step
# synchronously, and a serve that stays in the foreground holds the
# exec for its whole timeout, after which the process tree is KILLED
# and the readiness probe finds nothing listening. That is the
# "Waiting to answer" hang John measured on the Fusion button.
#
# The socket is already bound above, so by the time the parent
# prints, the URL is genuinely reachable.
url = direct_url(port)
if hasattr(os, "fork"):
if os.fork() > 0:
print(url, flush=True)
os._exit(0) # _exit: never run atexit / close the fd
os.setsid()
# Detach stdio or the exec's pipe stays open and it keeps waiting
# on a command that has, as far as it can tell, not finished.
_null = os.open(os.devnull, os.O_RDWR)
for _fd in (0, 1, 2):
try:
os.dup2(_null, _fd)
except Exception:
pass
claim_dock_webview(port) # socket already bound above
else:
# No fork (Windows): print and serve in the foreground. Honest
# fallback - the dock runs this in a Linux container.
print(url, flush=True)
claim_dock_webview(port)
log(f"kicad-dashboard {VERSION} serving on {port} (target={target or 'auto'})")
try:
srv.serve_forever()
except KeyboardInterrupt:
pass
return
if cmd == "status":
print(json.dumps(build_status(target or "-"), indent=2))
return
if cmd == "health":
print(json.dumps(local_call(port, "/health")))
return
if cmd == "ui":
sub = argv[1] if len(argv) > 1 else ""
if sub == "toast":
print(json.dumps(local_call(port, "/ui/toast",
{"message": argv[2] if len(argv) > 2 else "",
"kind": flag("--kind", "info")})))
elif sub == "click":
print(json.dumps(local_call(port, "/ui/click",
{"target": argv[2] if len(argv) > 2 else ""})))
elif sub == "state":
print(json.dumps(local_call(port, "/state"), indent=2))
elif sub == "console":
print(json.dumps(local_call(port, "/console"), indent=2))
else:
print(json.dumps({"error": "ui <toast|click|state|console>"}))
return
if cmd == "open":
url = proxy_url(port)
box = target or (get_targets() or [""])[0]
if "--webview" in argv:
print(json.dumps({"url": url, "surface": "webview",
"_hint": "Open this URL in a Hydrogen webview panel (the `webview` skill); "
"panel placement stays the AI's decision."}))
return
r = ab(box, "pup_open_window", {"sessionId": "kicad-dashboard", "url": url}, timeout=120)
ab(box, "pup_alert_window", {"sessionId": "kicad-dashboard"}, timeout=60)
print(json.dumps({"url": url, "surface": "pup", "target": box, "opened": r.get("success"),
"_hint": "Opened in a BACKGROUND pup window and flashed its taskbar button. "
"Never raise it yourself - let the user click it."}))
return
print(json.dumps({"error": f"unknown command {cmd}",
"_hint": "serve | status | ui <toast|click|state|console> | open | health"}))
if __name__ == "__main__":
main()