KiCad - the KiCad Bridge

@@ -97,12 +97,6 @@ def say_phrases():  HYDROGEN_CONTROL = "http://127.0.0.1:47084" -# SHIP-NIGHT PARK, server half (dock thread, 2026-08-23, copying nb's model):-# the UI already hides the Try it button, but a parked feature must refuse at-# the ENDPOINT too, or a direct POST (curl, a script, another AI) still-# reaches run-prompt and can wake or crash ah. One line to unpark, with the-# page's TRY_IT_PARKED flag.-TRY_IT_PARKED = True   def _hydrogen_control_url():@@ -194,226 +188,6 @@ def say_send(phrase):     return {"ok": True, "deliveredTo": str(delivered), "sid": d.get("sid")}  -def stage_phrase(phrase):-    """'Try it': stage the phrase (unsubmitted) into this app's own AI thread-    through the Hydrogen control API. Reachable from WSL2 under mirrored-    networking; from a cloud container it is not, and the page then falls back-    to copying. Never raises; returns {staged, reason, sid|tid}.--    Lookup recipe measured against the live API (dock thread, 2026-08-23):-    CENSUS FIRST (GET /ai-threads -> groups[].threads[] with tags, sid|null,-    tid), HISTORY SECOND (GET /ai-threads/history?tag=... -> items[].id; it-    EXCLUDES open threads by design, so history-first returns empty exactly-    when the user has the thread open). Deliver with sid when present, else-    tid. A closed thread is resumed (POST /ai-threads/resume {ai,id}) before-    run-prompt {sid:id}."""-    import urllib.request-    tag = "app:adom/kicad-bridge"--    def _get(path):-        with urllib.request.urlopen(f"{HYDROGEN_CONTROL}{path}", timeout=2) as r:-            return json.loads(r.read().decode("utf-8") or "{}")--    def _post(path, body):-        req = urllib.request.Request(f"{HYDROGEN_CONTROL}{path}",-                                     data=json.dumps(body).encode("utf-8"), method="POST",-                                     headers={"Content-Type": "application/json"})-        with urllib.request.urlopen(req, timeout=4) as r:-            return json.loads(r.read().decode("utf-8") or "{}")--    try:-        census = _get("/ai-threads")-    except Exception as e:-        return {"staged": False, "reason": f"hydrogen control api unreachable ({e.__class__.__name__})"}-    # 1. an OPEN thread tagged for this app-    for g in census.get("groups") or []:-        for t in g.get("threads") or []:-            if tag in (t.get("tags") or []):-                target = {"sid": t["sid"]} if t.get("sid") else {"tid": t.get("tid")}-                try:-                    _post("/ai-threads/run-prompt", {**target, "prompt": phrase, "submit": False})-                    return {"staged": True, "where": "open thread", **target}-                except Exception as e:-                    return {"staged": False, "reason": f"run-prompt failed ({e.__class__.__name__})", **target}-    # 2. a CLOSED thread from history: resume it, then stage-    try:-        hist = _get(f"/ai-threads/history?tag={tag}")-    except Exception as e:-        return {"staged": False, "reason": f"history lookup failed ({e.__class__.__name__})"}-    items = hist.get("items") or []-    if items:-        it = items[0]-        try:-            _post("/ai-threads/resume", {"ai": it.get("ai") or "claude", "id": it["id"]})-            _post("/ai-threads/run-prompt", {"sid": it["id"], "prompt": phrase, "submit": False})-            return {"staged": True, "where": "resumed thread", "sid": it["id"]}-        except Exception as e:-            return {"staged": False, "reason": f"resume/run-prompt failed ({e.__class__.__name__})",-                    "sid": it.get("id")}-    # 3. neither-    return {"staged": False, "reason": f"no AI thread tagged {tag} yet (open or closed)"}---# ---------------------------------------------------------------- bridge calls-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 rollup_state(leds: dict) -> dict:-    """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()@@ -1089,20 +863,16 @@ def do_action(target, name, args):     if name == "say":         return {"success": True, "phrases": say_phrases()}     if name == "try_phrase":-        if TRY_IT_PARKED:-            return {"success": False, "staged": False, "parked": True,-                    "error": "try_phrase is parked for ship night: the ai-threads "-                             "run-prompt path has an unverified crash fix, and this "-                             "endpoint must not wake ah's AI backend. Copy the phrase "-                             "into your AI instead."}-        phrase = str(args.get("phrase") or "").strip()-        if not phrase:-            return {"success": False, "error": "phrase required"}-        r = stage_phrase(phrase)-        return {"success": True, "phrase": phrase, **r}-    if name == "demo_control":-        cmd = str(args.get("cmd") or "").strip().lower()-        return ab(target, "kicad_demo", {"control": cmd}, timeout=30)+        # RETIRED 2026-08-24. This was the staged-prompt path, parked on ship+        # night and then superseded by Send to AI (say_send, one call, new+        # threads only). Its implementation is DELETED rather than dormant:+        # it contained an /ai-threads/resume call, and resume stays parked by+        # ruling, so leaving it alive-but-flagged put a forbidden call one+        # flag flip from running. Nothing calls this; answer honestly.+        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":