KiCad - the KiCad Bridge

@@ -0,0 +1,282 @@+"""Outbound calls from this bridge BACK to Adom Desktop (AD) core.++(Ported verbatim from adom/fusion-bridge ad_client.py, 2026-08-17 - keep in sync.)++Why this exists: as of **AD 1.9.84** a bridge's loopback callback+(`POST $ADOM_DIRECT_API_URL/command`) reaches the FULL AD verb set - not just+`desktop_*`. That means this bridge can, on its own:++  * `notify_user` the user directly (it used to be "Unknown desktop command", so+    the old code returned a `notifyUser` PAYLOAD and asked the driving AI to relay+    it - a workaround, now dropped);+  * enumerate peer ADs on the relay (`targets`) and route ANY verb to one via a+    cross-AD `target` - so a bridge running on a VM can reach the user on their+    LAPTOP ("notify them where they actually are").++Everything here is BEST-EFFORT and never raises: an older AD (pre-1.9.84), a+missing env var, or a closed port just yields None/False and the caller falls+back to its prior behavior. We CAPABILITY-PROBE (`GET /commands`) rather than+hardcode, per the SDK, so this keeps working as the verb set evolves.++Auth model (from disc #78): `X-Adom-Bridge-Token` = ATTRIBUTION only (AD binds+127.0.0.1 and trusts-by-install), so calls run ungated; a stale token 403s, so we+re-read it from the env on every call.+"""++import json+import os+import urllib.request+import urllib.error++_TIMEOUT = 8+# Cache the capability set for the process lifetime (the AD verb surface doesn't+# change under a running AD). None = not-yet-probed / unreachable.+_caps_cache = None+_self_name_cache = None+++_discovered_base = None  # None = not tried, "" = tried+failed, str = discovered base URL+++def _raw_get(url, timeout=4):+    """Low-level GET that does NOT go through _base_url (used by discovery so it can't recurse).+    Returns parsed JSON, or None."""+    try:+        req = urllib.request.Request(url, headers={"Accept": "application/json"}, method="GET")+        with urllib.request.urlopen(req, timeout=timeout) as resp:+            return json.loads(resp.read().decode("utf-8", "replace"))+    except Exception:+        return None+++def _looks_like_ad_api(data) -> bool:+    """Does a /commands response look like AD's verb catalog (so we don't bind a random service)?"""+    if isinstance(data, dict):+        if isinstance(data.get("commands"), list):+            return True+        # AD's /commands also carries a _hint mentioning POST /command+        if "command" in (data.get("_hint") or "").lower():+            return True+    return isinstance(data, list)+++def _discover_direct_api():+    """Find AD's loopback direct-API base by port-probing, for the case where AD did NOT inject+    ADOM_DIRECT_API_URL into our spawn env (seen live on an AD 1.9.103 Hyper-V VM: the direct API+    WAS up on 127.0.0.1 but the env var was absent, so available() short-circuited to False and the+    bridge fell back to slow CLI relays - John caught it 2026-07-14). We enumerate the loopback TCP+    ports adom-desktop.exe is LISTENING on and probe each `/commands`; the one that returns AD's verb+    catalog is the direct API. Cached for the process lifetime. Windows-only + best-effort; never raises."""+    global _discovered_base+    if _discovered_base:                # cache SUCCESS only; a transient miss must not stick forever+        return _discovered_base+    try:+        import subprocess as _sp, re as _re+        cnw = getattr(_sp, "CREATE_NO_WINDOW", 0)+        tl = _sp.run(["tasklist", "/FI", "IMAGENAME eq adom-desktop.exe", "/FO", "CSV", "/NH"],+                     capture_output=True, text=True, timeout=8, creationflags=cnw).stdout+        pids = set(_re.findall(r'"adom-desktop\.exe","(\d+)"', tl))+        if not pids:+            return None+        ns = _sp.run(["netstat", "-ano", "-p", "TCP"], capture_output=True, text=True,+                     timeout=8, creationflags=cnw).stdout+        seen = []+        for line in ns.splitlines():+            m = _re.search(r'\s(?:127\.0\.0\.1|\[::1\]):(\d+)\s+\S+\s+LISTENING\s+(\d+)', line)+            if m and m.group(2) in pids:+                p = int(m.group(1))+                if p not in seen:+                    seen.append(p)+        for p in seen:+            base = f"http://127.0.0.1:{p}"+            if _looks_like_ad_api(_raw_get(base + "/commands")):+                _discovered_base = base+                return base+    except Exception:+        pass+    return None+++def _base_url():+    """The loopback direct-API base: the URL AD injects into our spawn env, else one we DISCOVER by+    probing adom-desktop's loopback ports (AD sometimes fails to inject the env var even though the+    API is up - see _discover_direct_api). None only if neither yields a reachable API."""+    env = (os.environ.get("ADOM_DIRECT_API_URL") or "").rstrip("/")+    if env:+        return env+    return _discover_direct_api()+++def _token():+    # Re-read every call: after an AD restart the old token 403s.+    return os.environ.get("ADOM_BRIDGE_TOKEN") or ""+++def _http(method, path, body=None, timeout=_TIMEOUT):+    """Best-effort JSON request to the direct API. Returns parsed dict/list or None."""+    base = _base_url()+    if not base:+        return None+    url = base + path+    data = None+    headers = {"Accept": "application/json"}+    tok = _token()+    if tok:+        headers["X-Adom-Bridge-Token"] = tok+    if body is not None:+        data = json.dumps(body).encode("utf-8")+        headers["Content-Type"] = "application/json"+    req = urllib.request.Request(url, data=data, headers=headers, method=method)+    try:+        with urllib.request.urlopen(req, timeout=timeout) as resp:+            raw = resp.read().decode("utf-8", "replace")+    except Exception:+        return None+    try:+        return json.loads(raw)+    except Exception:+        return None+++def available() -> bool:+    """True if the direct API is reachable at all (env var present + port open)."""+    if not _base_url():+        return False+    return capabilities() is not None+++def _names_from_list(items):+    """Collect verb names from a list of strings and/or {name|command|verb} dicts."""+    out = set()+    for it in items or []:+        if isinstance(it, str):+            out.add(it)+        elif isinstance(it, dict):+            n = it.get("name") or it.get("command") or it.get("verb")+            if n:+                out.add(n)+    return out+++def _parse_command_names(data):+    """Extract the full verb-name set from a /commands response, tolerant of BOTH the legacy+    `{"commands":[...]}` shape AND the current AD (1.9.x) shape, where verbs are split across:+      topLevel: ["notify_user", "targets", ...]            (flat name list)+      cliRequired: ["pull_file", "shell_execute", ...]     (flat name list)+      bridges: [{name, verbs:[...]}, ...]                  (per-bridge verb lists)+      desktop: {commands:["screenshot_window", ...]}       (SHORT names; real verb = "desktop_"+name)+    Without this, capabilities() returned None on every current AD, so available() was always False+    and the bridge silently fell back to slow CLI relays (root-caused live, John 2026-07-14)."""+    names = set()+    if isinstance(data, list):+        return _names_from_list(data)+    if not isinstance(data, dict):+        return names+    if isinstance(data.get("commands"), list):          # legacy shape+        names |= _names_from_list(data["commands"])+    for k in ("topLevel", "cliRequired"):               # current: flat name lists+        if isinstance(data.get(k), list):+            names |= _names_from_list(data[k])+    for b in (data.get("bridges") or []):               # current: per-bridge verb lists+        if isinstance(b, dict):+            names |= _names_from_list(b.get("verbs"))+    dk = data.get("desktop")                            # current: desktop short names -> desktop_*+    if isinstance(dk, dict):+        for c in (dk.get("commands") or []):+            if isinstance(c, str):+                names.add(c)+                names.add("desktop_" + c)+    return names+++def capabilities():+    """Set of AD command names from `GET /commands` (cached). None if unreachable/unparseable."""+    global _caps_cache+    if _caps_cache is not None:+        return _caps_cache+    data = _http("GET", "/commands")+    if data is None:+        return None+    names = _parse_command_names(data)+    if not names:+        return None+    _caps_cache = names+    return names+++def has(command: str):+    """True/False if `command` is in the probed capability set; None if unknown+    (probe failed). Callers should try when the answer isn't an explicit False."""+    caps = capabilities()+    if caps is None:+        return None+    return command in caps+++def call(command: str, args=None, target=None, timeout=_TIMEOUT):+    """Dispatch ANY AD verb back through the direct API. Returns the parsed JSON+    response dict, or None on any failure (never raises).++    target: a peer AD clientName (from `peers()`), or "all" to fan out. Omit to+    run on THIS AD (the one hosting the bridge)."""+    payload = {"command": command}+    if args:+        payload["args"] = args+    if target:+        payload["target"] = target+    return _http("POST", "/command", body=payload, timeout=timeout)+++def self_name():+    """This AD's own clientName (to exclude it from peer routing). Best-effort."""+    global _self_name_cache+    if _self_name_cache is not None:+        return _self_name_cache or None+    st = _http("GET", "/status") or {}+    name = st.get("clientName") or st.get("name") or st.get("hostname")+    _self_name_cache = name or ""+    return name or None+++def peers():+    """clientNames of the OTHER ADs on the relay (excludes self). [] if none/unknown."""+    res = call("targets") or {}+    # tolerate shape drift: the peer list is under `clients` on current AD, `targets` on older.+    tlist = None+    if isinstance(res, dict):+        tlist = res.get("targets") or res.get("clients")+    if not isinstance(tlist, list):+        return []+    me = self_name()+    out = []+    for t in tlist:+        n = t if isinstance(t, str) else ((t or {}).get("name") or (t or {}).get("hostname") or (t or {}).get("clientName"))+        if n and n != me:+            out.append(n)+    return out+++def notify(title, body, level="info", buttons=None, target=None, reach_user=False):+    """Fire an AD `notify_user` toast. Returns the response dict, or None if the+    direct API is unreachable or AD is too old to have notify_user (caller falls+    back to returning a notifyUser payload for the driving AI to relay).++    reach_user: when this bridge may be running UNATTENDED (e.g. on a VM) while the+    user is on another machine, set True to also reach peer ADs. Since AD can't yet+    resolve which peer is "attended", we fan out to "all" when peers exist so the+    toast lands wherever the user actually is. `target` (explicit) overrides this.+    """+    if has("notify_user") is False:+        return None+    args = {"title": title, "body": body, "level": level}+    if buttons:+        args["buttons"] = buttons+    tgt = target+    if tgt is None and reach_user:+        # Fan out to EVERY connected desktop so the toast lands wherever the user actually is. Do NOT+        # gate on peers() (John 2026-07-14): the `targets` response shape drifted to {clients:[...]}+        # so peers() wrongly returns [] and the toast would hit only THIS (unattended VM). The relay+        # honors target="all" directly - verified reaching the user's laptop.+        tgt = "all"+    res = call("notify_user", args, target=tgt)+    return res

@@ -1,98 +1,151 @@-"""win_taskbar.py — taskbar overlay badges via ITaskbarList3::SetOverlayIcon.+"""win_taskbar.py — taskbar identity + Adom overlay badges for KiCad windows.  John's design (2026-08-17, after seeing pup's overlay badges): stamp the Adom-favicon as an OVERLAY on every bridge-created KiCad window's taskbar button, so+favicon as an OVERLAY on every bridge-driven KiCad window's taskbar button, so the user sees at a glance "adom is controlling my kicad". Gated by the `overlayBadges` setting (handlers/settings.py, default on). -Mechanics mirror pup's: Windows' ITaskbarList3 COM interface takes a per-window-HICON overlay drawn in the button's corner. The COM call is made from the-parking sweep thread (CoInitialize'd once per thread).+THE LESSON (paid for by pup first, then fusion, now us — src/server.js in pup,+handlers/taskbar_overlay.py in fusion): an in-bridge ITaskbarList3::SetOverlayIcon+is IN-PROCESS-ONLY. The COM call returns S_OK cross-process and the shell silently+ignores it (verified live here 2026-08-17: badges never rendered on kicad.exe's+buttons). The working path is AD-core's `desktop_taskbar` verb: hand it the badge+PNG as base64 (resources/adom-overlay-32.png — fusion's exact asset, mark anchored+bottom-right of a transparent 32px canvas so Windows renders it at the right size)+and AD paints the overlay natively, cross-process.++    set:   desktop_taskbar {hwnd, overlay: {icon: <b64 png>, tooltip}}+    clear: desktop_taskbar {hwnd, overlay: {badge: "none"}}++SECOND LESSON (same day): eeschema.exe/pcbnew.exe taskbar buttons showed blank+white-document icons even though both exes carry icon resources — the shell's+icon cache for those exe paths was poisoned (buttons first appeared during the+pre-0.9.151 parking era, before wx assigned window icons, and Windows cached the+blank). Rebuilding the shell icon cache means killing Explorer; instead we+BYPASS the poisoned cache with pup's identity machinery: stamp each eeschema/+pcbnew window with a per-exe AUMID (`desktop_set_window_identity`) whose+iconPath is a real .ico extracted from the exe itself — a fresh identity gets a+fresh icon lookup. kicad.exe's cache entry is healthy; its windows stay native. """ import os+import subprocess import sys+import threading from pathlib import Path  IS_WINDOWS = sys.platform == "win32" -if IS_WINDOWS:-    import ctypes-    import ctypes.wintypes-    from ctypes import POINTER, byref, c_void_p+_BADGE_PATH = str(Path(__file__).resolve().parent.parent / "resources" / "adom-overlay-32.png")+_ICON_DIR = os.path.join(os.environ.get("USERPROFILE") or str(Path.home()), ".adom", "kicad-icons") -_ICON_PATH = str(Path(__file__).resolve().parent.parent / "resources" / "adom-overlay.ico")+# exe basename (lower) -> (appId, displayName). Only the exes whose shell icon+# cache is known-poisoned; kicad.exe renders fine natively.+_IDENTITY_EXES = {+    "eeschema.exe": ("Adom.KiCad.Eeschema", "KiCad Schematic Editor"),+    "pcbnew.exe": ("Adom.KiCad.Pcbnew", "KiCad PCB Editor"),+} -_taskbar = None          # ITaskbarList3 pointer (per process, sweep thread)-_overlay_icon = None     # HICON, loaded once-_com_ready = False _badged: set = set()+_identified: set = set()+_icons_extracted = False+_B64 = None+_LOCK = threading.Lock()  -def _guid(s: str):-    class GUID(ctypes.Structure):-        _fields_ = [("Data1", ctypes.c_uint32), ("Data2", ctypes.c_uint16),-                    ("Data3", ctypes.c_uint16), ("Data4", ctypes.c_ubyte * 8)]-    import uuid-    u = uuid.UUID(s)-    g = GUID()-    g.Data1, g.Data2, g.Data3 = u.time_low, u.time_mid, u.time_hi_version-    rest = u.bytes[8:]-    for i in range(8):-        g.Data4[i] = rest[i]-    return g+def _badge_b64():+    global _B64+    if _B64 is not None:+        return _B64+    try:+        import base64+        _B64 = base64.b64encode(open(_BADGE_PATH, "rb").read()).decode()+    except Exception:+        _B64 = ""+    return _B64  -def _ensure_taskbar() -> bool:-    """CoCreate ITaskbarList3 on the CALLING thread (call from one thread only)."""-    global _taskbar, _overlay_icon, _com_ready-    if not IS_WINDOWS:-        return False-    if _taskbar is not None:-        return True+def _ad(verb: str, args: dict) -> bool:     try:-        ole32 = ctypes.windll.ole32-        if not _com_ready:-            ole32.CoInitialize(None)-            _com_ready = True-        CLSID_TaskbarList = _guid("56FDF344-FD6D-11d0-958A-006097C9A090")-        IID_ITaskbarList3 = _guid("EA1AFB91-9E28-4B86-90E9-9E9F8A5EECAF")-        p = c_void_p()-        CLSCTX_INPROC_SERVER = 1-        hr = ole32.CoCreateInstance(byref(CLSID_TaskbarList), None,-                                    CLSCTX_INPROC_SERVER,-                                    byref(IID_ITaskbarList3), byref(p))-        if hr != 0 or not p.value:-            return False-        # vtable: [0]QI [1]AddRef [2]Release [3]HrInit ... [18]SetOverlayIcon-        vtbl = ctypes.cast(p, POINTER(POINTER(c_void_p))).contents-        HrInit = ctypes.WINFUNCTYPE(ctypes.c_long, c_void_p)(vtbl[3])-        if HrInit(p) != 0:-            return False-        _taskbar = p-        # load the overlay icon once (16px is what the badge renders at)-        if os.path.isfile(_ICON_PATH):-            LR_LOADFROMFILE = 0x0010-            IMAGE_ICON = 1-            _overlay_icon = ctypes.windll.user32.LoadImageW(-                None, _ICON_PATH, IMAGE_ICON, 16, 16, LR_LOADFROMFILE)-        return True+        from handlers import ad_client+        r = ad_client.call(verb, args, timeout=8)+        raw = str(r)+        return bool(r) and ("Taskbar updated" in raw+                            or (isinstance(r, dict) and r.get("success") is not False+                                and not r.get("error")))     except Exception:         return False  -def _set_overlay(hwnd, hicon, description: str) -> bool:+def _hwnd_exe_base(hwnd) -> str:     try:-        vtbl = ctypes.cast(_taskbar, POINTER(POINTER(c_void_p))).contents-        SetOverlayIcon = ctypes.WINFUNCTYPE(-            ctypes.c_long, c_void_p, ctypes.wintypes.HWND,-            c_void_p, ctypes.wintypes.LPCWSTR)(vtbl[18])-        return SetOverlayIcon(_taskbar, hwnd, hicon, description) == 0+        from handlers.win_focus import _hwnd_exe_path+        return os.path.basename(_hwnd_exe_path(hwnd) or "").lower()     except Exception:-        return False+        return ""+++def _ensure_exe_icons() -> None:+    """Extract eeschema/pcbnew .ico files from the installed exes (once). Runs a+    tiny PowerShell because a stdlib-only HICON→.ico writer isn't worth owning."""+    global _icons_extracted+    if _icons_extracted or not IS_WINDOWS:+        return+    if all(os.path.isfile(os.path.join(_ICON_DIR, e.replace(".exe", ".ico")))+           for e in _IDENTITY_EXES):+        _icons_extracted = True+        return+    try:+        from kicad_detect import detect_kicad+        info = detect_kicad() or {}+        bin_dir = os.path.dirname(info.get("kicad_cli_exe") or info.get("kicadCli") or "")+        if not bin_dir:+            return+        ps = r'''+Add-Type -AssemblyName System.Drawing+$sig = '[DllImport("shell32.dll", CharSet=CharSet.Unicode)] public static extern uint ExtractIconExW(string f, int i, IntPtr[] big, IntPtr[] small, uint n);'+Add-Type -MemberDefinition $sig -Name IX -Namespace W+New-Item -ItemType Directory -Force -Path "%OUT%" | Out-Null+foreach ($exe in @("eeschema","pcbnew")) {+  $big = New-Object IntPtr[] 1; $small = New-Object IntPtr[] 1+  [W.IX]::ExtractIconExW("%BIN%\$exe.exe", 0, $big, $small, 1) | Out-Null+  if ($big[0] -ne [IntPtr]::Zero) {+    $icon = [System.Drawing.Icon]::FromHandle($big[0])+    $fs = [System.IO.File]::Create((Join-Path "%OUT%" ($exe + ".ico")))+    $icon.Save($fs); $fs.Close()+  }+}+'''.replace("%OUT%", _ICON_DIR).replace("%BIN%", bin_dir)+        cnw = getattr(subprocess, "CREATE_NO_WINDOW", 0)+        subprocess.run(["powershell", "-NoProfile", "-Command", ps],+                       capture_output=True, timeout=30, creationflags=cnw)+        _icons_extracted = True+    except Exception:+        pass+++def _ensure_identity(hwnd) -> None:+    """Bypass the poisoned per-exe shell icon cache: stamp eeschema/pcbnew+    windows with a per-exe AUMID + real iconPath (idempotent per hwnd)."""+    if hwnd in _identified:+        return+    exe = _hwnd_exe_base(hwnd)+    ident = _IDENTITY_EXES.get(exe)+    if not ident:+        _identified.add(hwnd)  # kicad.exe & friends: nothing to fix+        return+    _ensure_exe_icons()+    app_id, display = ident+    ico = os.path.join(_ICON_DIR, exe.replace(".exe", ".ico"))+    args = {"hwnd": int(hwnd), "appId": app_id, "displayName": display}+    if os.path.isfile(ico):+        args["iconPath"] = ico+    if _ad("desktop_set_window_identity", args):+        _identified.add(hwnd)   def apply_adom_badge(hwnd) -> bool:-    """Stamp the Adom overlay on a window's taskbar button (idempotent)."""+    """Repair the button's base icon (identity) then stamp the Adom overlay+    (idempotent per hwnd). Best-effort: never raises, never blocks a verb."""     try:         from handlers.settings import get_settings         if not get_settings().get("overlayBadges", True):@@ -101,18 +154,22 @@ def apply_adom_badge(hwnd) -> bool:         pass     if hwnd in _badged:         return True-    if not _ensure_taskbar() or not _overlay_icon:-        return False-    ok = _set_overlay(hwnd, _overlay_icon, "Adom is driving this KiCad window")-    if ok:-        _badged.add(hwnd)-    return ok+    with _LOCK:+        _ensure_identity(hwnd)+        icon = _badge_b64()+        if not icon:+            return False+        ok = _ad("desktop_taskbar", {"hwnd": int(hwnd), "overlay": {+            "icon": icon, "tooltip": "Adom is driving this KiCad window"}})+        if ok:+            _badged.add(hwnd)+        return ok   def clear_adom_badge(hwnd) -> None:     try:-        if hwnd in _badged and _ensure_taskbar():-            _set_overlay(hwnd, None, "")+        if hwnd in _badged:+            _ad("desktop_taskbar", {"hwnd": int(hwnd), "overlay": {"badge": "none"}})         _badged.discard(hwnd)     except Exception:         pass@@ -152,10 +209,9 @@ def rebadge_all(enable: bool) -> int:  def badge_existing_on_boot() -> None:     """Server boot: badge KiCad windows that survived a bridge restart, off the-    boot path (COM + enumeration can stall; never block serving)."""+    boot path (AD calls + enumeration can stall; never block serving)."""     if not IS_WINDOWS:         return-    import threading     import time      def _later():