KiCad - the KiCad Bridge

@@ -0,0 +1,105 @@+"""Caller provenance — WHO asked (AD 1.9.180 + 1.9.183).++A user runs ~20 AI threads against one desktop. AD now stamps three headers on+every relayed request so a bridge can attribute work to the thread that caused+it. This module keeps that identity for the life of a request and re-forwards it+on any AD callback the bridge makes while carrying out that request, so the+Activity Log shows `chip-fetcher tab 3 (via kicad)` instead of a nameless bridge.++Trust rule (AD's, and it is not optional): these values are SELF-ASSERTED, like+`reason`. Use them for logging, UX, window labels and arbitration hints — NEVER+for authorization. `if thread == "admin-tab": allow()` is a hole, not a check.+"""++import threading++# The identity of the request THIS thread is currently serving. A threading+# HTTPServer handles each request on its own thread, so thread-local is the+# right scope — no request can see another's caller.+_current = threading.local()++_HEADERS = ("X-Adom-Caller-Thread", "X-Adom-Caller-Container", "X-Adom-Caller-Reason")++# What we name ourselves when we forward. Kept generic ("kicad") to match the+# verb prefix AD already knows us by.+DELEGATE = "kicad"+++def set_from_headers(headers) -> dict:+    """Capture the caller headers off an incoming request. `headers` is anything+    with .get() (http.server's message object). Returns the captured dict."""+    ident = {h: (headers.get(h, "") or "") for h in _HEADERS}+    _current.ident = ident+    return ident+++def clear() -> None:+    _current.ident = None+++def thread_name() -> str:+    """The caller's thread name, or '' if none was supplied (loopback callers —+    HD, another bridge, the local CLI — are exempt and send nothing)."""+    ident = getattr(_current, "ident", None) or {}+    return ident.get("X-Adom-Caller-Thread", "")+++def log_suffix() -> str:+    """A `caller="..."` fragment for per-verb log lines, or '' when anonymous."""+    t = thread_name()+    return f' caller="{t}"' if t else ""+++def forward_headers() -> dict:+    """Headers to put on an AD callback made WHILE serving an AI's verb. Echoes+    the three we were handed and ADDS X-Adom-Caller-Delegate naming us, so AD can+    show both facts: who asked AND who ran it. Empty when there's no caller to+    forward (e.g. a self-initiated poll) — see self_caller() for that case."""+    ident = getattr(_current, "ident", None) or {}+    if not ident.get("X-Adom-Caller-Thread"):+        return {}+    fwd = {h: ident.get(h, "") for h in _HEADERS}+    fwd["X-Adom-Caller-Delegate"] = DELEGATE+    return fwd+++def self_caller() -> dict:+    """The `caller` args block to send on a callback the bridge makes on its OWN+    behalf (health poll, timer, crash cleanup). An explicit caller in args always+    wins over the forwarded headers, so this never leaks a stale thread name."""+    return {"aiThread": "kicad bridge (self)", "containerName": "local"}+++def ad_callback(command: str, args: dict = None, on_behalf: bool = True, timeout: int = 30):+    """Call an AD verb back over the loopback direct API, forwarding the caller.++    `on_behalf=True` (default): this call is part of serving an AI's verb, so it+    forwards the X-Adom-Caller-* headers + adds our Delegate — AD shows+    "chip-fetcher tab 3 (via kicad)". `on_behalf=False`: a self-initiated call+    (poll/timer/cleanup) — sends our own `caller` in args so no stale thread leaks.++    Reads ADOM_DIRECT_API_URL + ADOM_BRIDGE_TOKEN from the spawn env (AD injects+    both). Returns the parsed JSON, or {"success": False, ...} if AD is+    unreachable or the env isn't populated. NEVER raises — a labeling/notify+    callback failing must not fail the KiCad verb the user actually asked for.+    """+    import os, json as _json, urllib.request+    base = os.environ.get("ADOM_DIRECT_API_URL", "").rstrip("/")+    if not base:+        return {"success": False, "error": "ADOM_DIRECT_API_URL not in env (inbound-only spawn?)"}+    body = {"command": command, "args": dict(args or {})}+    headers = {"Content-Type": "application/json"}+    tok = os.environ.get("ADOM_BRIDGE_TOKEN")+    if tok:+        headers["X-Adom-Bridge-Token"] = tok   # attribution badge, not an auth gate+    if on_behalf:+        headers.update(forward_headers())+    else:+        body["args"].setdefault("caller", self_caller())+    try:+        req = urllib.request.Request(f"{base}/command", data=_json.dumps(body).encode(),+                                     headers=headers, method="POST")+        with urllib.request.urlopen(req, timeout=timeout) as r:+            return _json.loads(r.read().decode() or "{}")+    except Exception as e:+        return {"success": False, "error": f"AD callback failed: {e}"}