app
Altium - the Altium Bridge
Public Made by Adomby adom
Adom Bridge for Altium Designer: a resident DelphiScript agent gives the AI eval, the 1,100-command process launcher, all 8 EDA surfaces with screenshots, compile/ERC, DRC, OutJob outputs, native SchLib/PcbLib authoring with STEP 3D bodies, and dialog handling by PID.
main
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584
#!/usr/bin/env python3
"""altium-dashboard - the Altium 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:
altium-dashboard serve [--port 8931] [--target BOX] [--print-url] [--dev]
altium-dashboard status [--target BOX]
altium-dashboard ui toast "msg" [--kind ok|err|info]
altium-dashboard ui click <setup|launch|close|demo|part-tour|settings|refresh>
altium-dashboard ui state | ui console
altium-dashboard open [--pup | --webview] [--target BOX]
altium-dashboard health
--dev (or ALTIUM_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 = "0.1.1"
AI_THREAD = "altium-dashboard"
DEFAULT_PORT = 8932
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 Altium 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 Altium 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-altium.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.
_SAY_FALLBACK = [
"open the symbol in altium",
"send this footprint to altium",
"show me the board in altium 3d",
"open my project in altium",
"install this part into altium",
"export gerbers from altium",
]
def say_phrases():
for cand in (Path(__file__).resolve().parent.parent / "dockbar.json",
Path.home() / ".local" / "share" / "adom" / "altium-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/altium/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-altium.svg", "/altium-logo.png", "/updater/status",
"/api/viewport", "/ai-threads/run-prompt", "/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/altium"), 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
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
body = {"new": True, "prompt": phrase, "submit": True,
"name": "Altium - say it", "tags": ["dock:altium-bridge"]}
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
# 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 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("ALTIUM_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 / ALTIUM_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/.altium-verbtest-last-report.json"
if not os.path.exists(rp):
rp = "/tmp/altium-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/altium-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 ~/.altium-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", "altium", "license") # can this thing work at all
_ADVISORY_LEDS = ("update", "running", "agent") # 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"),
"altium": 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"),
"license": lambda l: (str(l.get("detail") or "").strip()[:18]
if l.get("state") == "ok" else "no licence"),
"running": lambda l: ("running" if l.get("state") == "ok" else "closed"),
"agent": lambda l: ("resident" if l.get("state") == "ok" else "dormant"),
"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, Altium closed, the resident agent dormant) never turn the card "
"yellow, because the bridge still works. Only relay/box/bridge/altium/license 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"), ("altium", "Altium"), ("license", "Licence"),
("update", "Updates"), ("running", "Running"), ("agent", "Agent")):
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, "altium_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 "Altium 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"), ("altium", "Altium"), ("license", "Licence"),
("update", "Updates"), ("running", "Running"), ("agent", "Agent")):
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"))
inst = r.get("install") or {}
ver = r.get("adVersion") or inst.get("fileVersion") or "?"
leds["bridge"] = {"label": "Bridge", "state": "ok",
"detail": f"answering (bridge {r.get('bridgeVersion') or '?'})",
"_hint": "The altium_* verb surface is served by the bridge on the box. If red: "
"bridge_kill {name:'altium',confirm:true} respawns it from cache."}
leds["altium"] = {"label": "Altium", "state": "ok" if installed else "err",
"detail": f"Altium Designer {ver}" if installed else "not installed",
"_hint": ("Installed at " + str(inst.get("base_dir") or "") + " and driveable.") if installed
else "No X2.EXE under Program Files\\Altium. The altium-install skill drives "
"the installer + trial sign-in; the installer elevates, so the last "
"clicks are the user's."}
running = bool(r.get("running"))
leds["running"] = {"label": "Running", "state": "ok" if running else "off",
"detail": "Altium is open" if running else "Altium is closed",
"_hint": "altium_launch {} starts it in the background (cold start about 60s). "
"Every show_* verb launches it on demand, so closed is not a problem."}
lic = r.get("licenceState") or ""
if not installed:
leds["license"] = {"label": "Licence", "state": "off", "detail": "n/a",
"_hint": "Unknowable until Altium is installed."}
elif not running:
leds["license"] = {"label": "Licence", "state": "ok" if r.get("licensed") else "off",
"detail": lic or "unknown until launched",
"_hint": "The licence is read from the running instance's title bar; "
"launch Altium to refresh it."}
elif r.get("licensed"):
leds["license"] = {"label": "Licence", "state": "ok", "detail": lic or "licensed",
"_hint": "Scripting is allowed. altium_license {action:'show'} opens the "
"License Management page if you need to switch."}
else:
leds["license"] = {"label": "Licence", "state": "err", "detail": lic or "No License",
"_hint": "Altium refuses scripting without a licence. altium_license "
"{action:'start_trial'} or {action:'use_license'} drives the "
"License Management page; altium_signin_status checks the account."}
agent = bool(r.get("agentAlive"))
leds["agent"] = {"label": "Agent", "state": "ok" if agent else "off",
"detail": "resident DelphiScript loop answering" if agent
else ("dormant (auto-starts on the first verb)" if running else "needs Altium"),
"_hint": "The resident agent is the REPL lane inside Altium (eval, open/show, "
"library authoring). altium_agent_start {} boots it (about 8s); "
"altium_agent_recover {} clears a halted engine."}
cu = ab(target, "bridge_check_updates", {"name": "altium"}, timeout=T)
row = {}
for b in (cu.get("bridges") or []):
if isinstance(b, dict) and b.get("name") == "altium":
row = b
if not row and cu.get("name") == "altium":
row = cu
if cu.get("error"):
leds["update"] = {"label": "Updates", "state": "off", "detail": "unknown",
"_hint": f"bridge_check_updates did not answer ({str(cu.get('error'))[:80]})."}
elif row.get("updateAvailable"):
leds["update"] = {"label": "Updates", "state": "warn",
"detail": f"bridge {row.get('current')} -> {row.get('latest')} on the wiki",
"_hint": "refresh_bridges {name:'altium'} pulls it; bridge_kill respawns on the new code."}
else:
leds["update"] = {"label": "Updates", "state": "ok",
"detail": f"bridge {r.get('bridgeVersion') or row.get('current') or '?'} is current",
"_hint": "Adom Bridge re-syncs every bridge from its wiki page every 4h."}
if DEV_MODE:
leds["suite"] = _suite_led()
wins = []
if running:
wi = ab(target, "altium_window_info", {}, timeout=T)
wins = [w.get("title") for w in (wi.get("windows") or []) if w.get("title")]
return {"target": target, "targets": targets, "leds": _apply_activity(leds),
**rollup_state(_apply_activity(leds)),
"settings": {},
"windows": wins,
"bridgeVersion": r.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 altium 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():
r = ab(name, "altium_readiness", {}, timeout=10)
rows.append({"name": name, "altiumRunning": bool(r.get("running")), "windows": 0,
"altiumVersion": r.get("adVersion"),
"responding": not bool(r.get("error"))})
_targets_cache.update(t=time.time(), rows=rows)
return {"targets": rows,
"_hint": "Pick where Altium should run. The bridge only drives instances IT launched, "
"so a Altium the user opened on any box is never parked, bounced, or badged."}
# -------------------------------------------------------------------- actions
def part_tour(target, args):
"""Not a dashboard button on Altium (yet): the full wiki-part import is a
creative job (variants, pin-1 checks, 3D binds), so it belongs to an AI
thread with the say-strip phrase, not a one-click arm that would hide the
decisions. The strip's 'add this wiki part to my altium library' does it."""
page = (args or {}).get("page") or "adom/lan7800"
return {"success": False, "retired": False,
"error": "Part import is driven by your AI on Altium, not by this button.",
"_hint": f"Send to AI: 'add {page} to my altium library' - the thread fetches the page, "
"authors the symbol/footprint/3D body with altium_create_* + altium_add_body, "
"and proves it with frames."}
# ---------------------------------------------------------------- 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 Altium, opening JUST that surface in the background, and rendering
# it from altium_state. The whole circle: wiki component -> the app the EE
# designs in.
PART = {"page": "adom/altium-bridge", "name": "Altium examples", "lib": "Altium examples"}
SURFACES = [
("symbol", "Symbol", "altium_show_symbol", None, "schlib",
"the SpiritLevel example SchLib, one symbol selected in the SCH Library panel"),
("footprint", "Footprint", "altium_show_footprint", None, "pcblib",
"the example PcbLib, one footprint selected in the PCB Library panel"),
("3d-chip", "3D chip", "altium_show_3d_chip", None, "pcblib",
"that footprint in the library editor's 3D mode"),
("library", "Library", "altium_show_library", None, "schlib",
"the library manager panel listing every component"),
("schematic", "Schematic", "altium_show_schematic", None, "schdoc",
"the Bluetooth Sentinel example schematic"),
("2d-board", "2D board", "altium_show_2d_board", None, "pcbdoc",
"the Bluetooth Sentinel board in 2D"),
("3d-board", "3D board", "altium_show_3d_board", None, "pcbdoc",
"the same board in 3D"),
("project", "Project", "altium_show_project", None, "prjpcb",
"the example project open in the Projects panel"),
]
def _altium_hwnd(target):
"""Main Altium window hwnd, for taskbar progress."""
try:
wi = ab(target, "altium_window_info", {}, timeout=30)
for w in (wi.get("windows") or []):
if w.get("kind") == "main" and w.get("hwnd"):
return int(w["hwnd"])
for w in (wi.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 altium and you know something is going
to take a while, use ab's progress bar indicator in the windows taskbar".
Paints on the Altium 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 _altium_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 ensure_ready(target, surface):
"""A click is a GOAL, not a request for an audit: if Altium is closed,
launch it in the background and wait - never hand the user an error they
would fix by pressing a button we could have pressed."""
r = ab(target, "altium_readiness", {}, timeout=30)
if r.get("running"):
return True
_emit(surface, log="Altium is closed - launching it in the background first", percent=15)
ab(target, "altium_launch", {}, timeout=180)
for _ in range(30): # cold start is about 60s, measured
time.sleep(3)
r = ab(target, "altium_readiness", {}, timeout=30)
if r.get("running"):
_emit(surface, log="Altium 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
(altium_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, "altium_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):
"""Open this surface in the BACKGROUND on the bridge's shipped example
files, then show the frame the verb itself returned. Streams progress over
SSE and onto the Altium taskbar button."""
spec = next((s for s in SURFACES if s[0] == key), None)
if key == "state":
spec = ("state", "Altium state", "altium_state", None, None,
"captures every Altium 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("running", f"opening the {label.lower()} in the background...", target)
hwnd = taskbar_progress(target, None, "indeterminate")
if not ensure_ready(target, key):
raise RuntimeError("Altium did not come up within 90s - try Launch Altium, "
"or check the Altium and Licence LEDs")
taskbar_progress(target, 60, "normal", hwnd)
_emit(key, log=f"opening the {label.lower()} in the background", percent=60)
args = {"timeout": 120} if key != "state" else {"screenshots": True}
_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:
code = r.get("errorCode")
if code == "blocked_by_dialog":
raise RuntimeError("a dialog is up on Altium's main window - "
"altium_check_dialogs shows it, altium_dismiss_dialogs clears it")
if code == "not_licensed":
raise RuntimeError("Altium is running without a licence - see the Licence LED")
raise RuntimeError(str(r.get("error") or code)[:160])
shot, title = None, None
sh = r.get("shot") or {}
if isinstance(sh, dict) and sh.get("image"):
shot, title = sh.get("image"), sh.get("title") or r.get("title")
if not shot and key == "state":
shots = r.get("shots") or []
if shots:
shot, title = shots[0].get("image"), shots[0].get("title")
if shot and str(shot).startswith("data:"):
shot = shot.split(",", 1)[1]
if not shot:
_emit(key, log="rendering it from altium_state", percent=92)
shot, title = _render(target, pick)
taskbar_progress(target, 100, "none", hwnd)
if not shot:
_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("running", target)
def _render(target, pick):
"""Pull this surface's window out of altium_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, "altium_state", {"screenshots": True}, timeout=240)
shots = st.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 licence and
the resident agent...' then back to the real status once it is up."""
try:
set_activity("running", "launching Altium in the background (about 60s)...", target)
r = ab(target, "altium_launch", {}, timeout=240)
for _ in range(30):
time.sleep(3)
rd = ab(target, "altium_readiness", {}, timeout=25)
if rd.get("running"):
break
set_activity("agent", "Altium is up - booting the resident agent...", target)
ab(target, "altium_agent_start", {"waitSec": 40}, timeout=90)
return r
finally:
clear_activity("running", target)
clear_activity("agent", target)
# ------------------------------------------------------------------- demo
# altium_demo is one beat per call ({stage}); the dashboard runs the loop so
# the page gets a real progress bar and one screenshot per finished beat.
_demo = {"running": False, "percent": 0, "stepLabel": "", "beatIndex": 0, "beatCount": 6,
"done": False, "error": None, "shots": [], "paused": False, "stop": False}
_demo_lock = threading.Lock()
def _demo_run(target):
stage = 0
with _demo_lock:
_demo.update(running=True, percent=0, stepLabel="starting", beatIndex=0,
done=False, error=None, shots=[], paused=False, stop=False)
try:
hwnd = taskbar_progress(target, None, "indeterminate")
while True:
while _demo["paused"] and not _demo["stop"]:
time.sleep(1)
if _demo["stop"]:
break
r = ab(target, "altium_demo", {"stage": stage}, timeout=240)
if r.get("done"):
break
if r.get("success") is False:
raise RuntimeError(str(r.get("error") or r.get("errorCode") or "beat failed")[:160])
label = r.get("label") or f"beat {stage + 1}"
shots = r.get("screenshots") or []
with _demo_lock:
_demo["beatIndex"] = stage + 1
_demo["stepLabel"] = label
_demo["percent"] = int(100 * (stage + 1) / max(1, _demo["beatCount"]))
for sh in shots:
if sh.get("image"):
_demo["shots"].append({"image": sh["image"], "title": label, "step": label})
taskbar_progress(target, _demo["percent"], "normal", hwnd)
stage += 1
taskbar_progress(target, 100, "none", hwnd)
with _demo_lock:
_demo.update(done=True, percent=100, stepLabel="complete" if not _demo["stop"] else "stopped",
running=False)
except Exception as e:
with _demo_lock:
_demo.update(done=True, error=str(e)[:200], running=False)
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 Running and Agent LEDs show live "
"progress and the page updates over /events as it happens."}
if name == "close":
set_activity("running", "closing Altium...", target)
try:
return ab(target, "altium_close", {}, timeout=180)
finally:
clear_activity("running", target)
if name in ("setup", "install_plugin", "agent_start"):
set_activity("agent", "booting the resident DelphiScript agent...", target)
try:
return ab(target, "altium_agent_start", {"waitSec": 40}, timeout=90)
finally:
clear_activity("agent", target)
if name == "demo":
if _demo.get("running"):
return {"success": False, "error": "a demo is already running",
"_hint": "Poll demo_progress, or demo_control {cmd:'stop'} first."}
threading.Thread(target=_demo_run, args=(target,), daemon=True).start()
return {"success": True, "started": "demo",
"_hint": "Six beats in the background; poll demo_progress every few seconds."}
if name == "demo_progress":
with _demo_lock:
snap = {k: _demo[k] for k in ("percent", "stepLabel", "beatIndex", "beatCount", "done", "error")}
snap["stepShot"] = _demo["shots"].pop(0) if _demo["shots"] else None
if _demo.get("paused"):
snap["stepLabel"] = "Paused"
return snap
if name == "demo_control":
cmd = str(args.get("cmd") or "")
if cmd == "pause":
_demo["paused"] = True
elif cmd == "play":
_demo["paused"] = False
elif cmd == "stop":
_demo["stop"] = True
_demo["paused"] = False
else:
return {"error": f"unknown demo command {cmd}"}
return {"success": True, "cmd": cmd}
if name == "upgrade":
set_activity("update", "pulling the latest altium bridge from the wiki...", target)
try:
return ab(target, "refresh_bridges", {"name": "altium"}, timeout=300)
finally:
clear_activity("update", target)
if name in ("set_overlay_badges", "set_activity_indicator", "restore_settings"):
return {"success": False, "error": "the altium bridge has no dashboard-side settings yet",
"_hint": "Taskbar badge + progress are always on (handlers/taskbar.py); nothing to toggle."}
if name == "say":
return {"success": True, "phrases": say_phrases()}
if name == "try_phrase":
return {"success": False, "retired": True,
"error": "try_phrase is retired. Use Send to AI (POST /api/say-send).",
"_hint": "Resuming or typing into EXISTING threads remains parked."}
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 Altium 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, setup, demo, demo_progress, demo_control, upgrade, surface, surfaces."}
# --------------------------------------------------------------------- server
class Handler(BaseHTTPRequestHandler):
protocol_version = "HTTP/1.1"
server_version = "altium-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")):
# The TAB icon. Hydrogen's webview reads the page's favicon after load and writes it
# into the tab (hydrogen-tab-icons), so THIS is what the user sees in the tab strip.
# It must be the monochrome #e6edf3 LINE MARK — this route used to alias to
# dock-altium.svg, which is a 60KB SVG wrapping an embedded PNG raster: Hydrogen's tab
# renderer shows nothing for it, which is exactly the iconless tab John reported
# (2026-09-01). ui/favicon.svg is the same line drawing as the header mark.
fav = UI_DIR / "favicon.svg"
try:
return self._send(200, "image/svg+xml",
(fav if fav.is_file() else LOGO_PATH).read_bytes())
except Exception:
return self._json({"error": "favicon missing"}, 404)
if path.endswith(("/dock-altium.svg", "/altium-logo.png")):
# the dock CARD art (larger, may be rich) — distinct from the tab favicon on purpose
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": "altium-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("/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, "altium_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"):
return _proxy_control(self, "/ai-threads/run-prompt", body=body or {}, 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? `altium-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") == "altium-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"altium-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"altium_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"altium-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"altium-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"altium-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"altium-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": "altium-dashboard", "url": url}, timeout=120)
ab(box, "pup_alert_window", {"sessionId": "altium-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()