KiCad - the KiCad Bridge

@@ -476,6 +476,30 @@ def _job_update(**kw):         _JOB.update(kw)  +def _foreground_once(window_title: str) -> bool:+    """Bring the window this beat opened to the front ONCE, then hands off.+    The bridge's background contract marks its KiCad windows no-activate, so+    lift that first; AttachThreadInput makes SetForegroundWindow honest from a+    background process. Never re-raised later: if the user alt-tabs away+    mid-beat, that is their desktop."""+    if not window_title:+        return False+    try:+        from handlers.kicad_ui import _find_all_kicad_windows+        from handlers import win_focus+        for w in _find_all_kicad_windows():+            if window_title.lower() in (w.get("title") or "").lower():+                h = w["hwnd"]+                try:+                    win_focus.allow_activation(h)+                except Exception:  # pylint: disable=broad-except+                    pass+                return bool(win_focus._set_foreground_robust(h))+    except Exception:  # pylint: disable=broad-except+        pass+    return False++ def _beat_shot(kicad_info: dict, step_label: str, window_title: str) -> dict | None:     """The per-step EVIDENCE frame for the web-control page (#25, John 2026-08-16:     the page's own guess-snaps race the tour and shoot the wrong window — only@@ -536,6 +560,7 @@ _NARR: dict = {"proc": None, "muted": False}   def _play_narration(step: str) -> None:+    _NARR["step"] = step     if _NARR.get("muted"):         return     f = _AUDIO_DIR / f"{step}.mp3"@@ -620,17 +645,46 @@ _PANEL_STOPS = ["Preparing the project", "The schematic", "The 2D board", "The b                 "The symbol", "The footprint", "The part in 3D"]  -def _panel_show(stage: int, state: str = "playing") -> bool:+_PANEL_SEEN: set = set()          # `at` stamps already honored via the instant callback+_PANEL_SHOWN_ONCE = {"v": False}+++def _callback_url() -> str | None:+    """Loopback URL ab POSTs panel presses to (ab 2.1.17). None when the bridge's+    HTTP server has not bound yet (staged single-step runs from a test harness)."""+    try:+        import server as _srv+        port = (_srv.BOUND or {}).get("port")+        host = (_srv.BOUND or {}).get("host") or "127.0.0.1"+        if port:+            return f"http://{host}:{int(port)}/demo/event"+    except Exception:  # pylint: disable=broad-except+        pass+    return None+++def _panel_show(stage: int, state: str | None = "playing") -> bool:+    """show at stage 0 carries everything (title, stops, reason, callback URL);+    later calls send ONLY {id, stage}: title/stops/state/muted persist on ab's+    side, and resending state would un-pause or un-mute a user who pressed+    those (DEMO_KIT: "a stage-advance show never un-pauses or un-mutes").+    Pass state explicitly only when the SCRIPT changes it (our own pause)."""     if _PANEL["ok"] is False:         return False     try:         from handlers import ad_client-        r = ad_client.call("desktop_demo_panel", {-            "action": "show", "id": _PANEL_ID, "title": "KiCad bridge tour",-            "stops": _PANEL_STOPS, "stage": int(stage), "state": state,-            "position": "bottom-right",-            "reason": "Transport for the KiCad tour the user started: play, pause, next, back, stop"},-            timeout=8) or {}+        args = {"action": "show", "id": _PANEL_ID, "stage": int(stage)}+        if not _PANEL_SHOWN_ONCE["v"]:+            args.update({"title": "KiCad bridge tour", "stops": _PANEL_STOPS,+                         "state": state or "playing", "position": "bottom-right",+                         "muted": bool(_NARR.get("muted")),+                         "reason": "Transport for the KiCad tour the user started: play, pause, next, back, mute, stop"})+            cb = _callback_url()+            if cb:+                args["eventCallbackUrl"] = cb+        elif state in ("playing", "paused") and _PANEL.get("scriptState") != state:+            args["state"] = state+        r = ad_client.call("desktop_demo_panel", args, timeout=8) or {}         err = str(r.get("error") or "")         if "unknown verb" in err.lower() or "no installed bridge" in err.lower():             _PANEL["ok"] = False@@ -643,16 +697,69 @@ def _panel_show(stage: int, state: str = "playing") -> bool:             except Exception:  # pylint: disable=broad-except                 out = {}         data = (out or {}).get("data") if isinstance(out, dict) else None-        shown = bool((data or out or {}).get("shown")) or (r.get("success") is True and not err)-        _PANEL["ok"] = True if shown else (_PANEL["ok"] or None)+        d = data or out or {}+        shown = bool(d.get("shown")) or (r.get("success") is True and not err)+        if shown:+            _PANEL["ok"] = True+            if not _PANEL_SHOWN_ONCE["v"]:+                _job_update(panel={"shown": True, "callbackUrl": args.get("eventCallbackUrl"),+                                   "reply": {k: d.get(k) for k in ("shown", "id", "callbackAccepted", "error") if k in d}})+            _PANEL_SHOWN_ONCE["v"] = True+            if state in ("playing", "paused"):+                _PANEL["scriptState"] = state+            if "callbackAccepted" in d or "eventCallbackUrl" in d:+                _PANEL["callback"] = bool(d.get("callbackAccepted", True))+        elif _PANEL["ok"] is None:+            _PANEL["ok"] = None         return shown     except Exception:  # pylint: disable=broad-except         return False  +def _apply_mute(muted: bool, source: str) -> None:+    """The user's mute is the law: stop the sound within the beat, keep the+    show and captions running; on unmute, replay the current beat's line."""+    was = bool(_NARR.get("muted"))+    _NARR["muted"] = bool(muted)+    if muted and not was:+        _stop_narration()+    elif was and not muted:+        step = _NARR.get("step")+        if step:+            _play_narration(step)+    _job_update(muted=bool(muted), muteSource=source)+++def on_panel_event(ev: dict) -> dict:+    """ab's instant callback (POST /demo/event) for one press. Applies mute at+    once, stops narration at once on pause/stop, and queues the transport+    command for the beat boundary (a beat in flight cannot be interrupted)."""+    if not isinstance(ev, dict) or str(ev.get("id") or _PANEL_ID) != _PANEL_ID:+        return {"ignored": True}+    at = ev.get("at")+    if at is not None:+        _PANEL_SEEN.add(str(at))+    b = str(ev.get("button") or "").lower()+    if b in ("mute", "unmute"):+        _apply_mute(b == "mute", "panel")+        return {"applied": b}+    if "muted" in ev and bool(ev["muted"]) != bool(_NARR.get("muted")):+        _apply_mute(bool(ev["muted"]), "panel")+    mapping = {"back": "prev", "play": "play", "pause": "pause", "next": "next", "stop": "exit"}+    cmd = mapping.get(b)+    if not cmd:+        return {"ignored": b}+    if cmd in ("pause", "exit"):+        _stop_narration()           # the user's hand: silence within the beat+    set_demo_control(cmd)+    _job_update(lastPress={"button": b, "at": at, "via": "callback"})+    return {"queued": cmd}++ def _panel_poll() -> str | None:-    """Drain the panel's button presses; newest press wins. Maps the panel's-    vocabulary (back/play/pause/next/stop) onto the demo's (prev/play/pause/next/exit)."""+    """Backstop drain of the panel's press queue (the instant callback is the+    fast path; presses it already delivered are skipped by their `at` stamp).+    Maps back/play/pause/next/stop onto prev/play/pause/next/exit."""     if not _PANEL["ok"]:         return None     try:@@ -667,14 +774,22 @@ def _panel_poll() -> str | None:                 out = {}         data = (out or {}).get("data") if isinstance(out, dict) else None         d = data or out or {}-        events = d.get("events") or []+        if d.get("state") == "gone":+            # hid on idle or replaced: re-show once at the current stage+            _PANEL_SHOWN_ONCE["v"] = False+            _panel_show(int(_JOB.get("beatIndex") or 0), None)+            return None+        if "muted" in d and bool(d["muted"]) != bool(_NARR.get("muted")):+            _apply_mute(bool(d["muted"]), "panel-poll")+        events = [e for e in (d.get("events") or []) if str((e or {}).get("at")) not in _PANEL_SEEN]         mapping = {"back": "prev", "play": "play", "pause": "pause", "next": "next", "stop": "exit"}         for ev in reversed(events):             b = str((ev or {}).get("button") or "").lower()+            if b in ("mute", "unmute"):+                _apply_mute(b == "mute", "panel-poll")+                continue             if b in mapping:                 return mapping[b]-        # No press since last poll: honor the panel's own state (the user may-        # have toggled pause through a press we already consumed).         if d.get("state") == "paused":             return "pause-held"     except Exception:  # pylint: disable=broad-except@@ -751,13 +866,12 @@ def _run_demo_background(kicad_info: dict) -> None:     # inside the thread for the full job + margin. The guardian is KiCad-aware     # (adopts the user's own window switches, never fights them), so a long arm     # is harmless. (Gap found in the 2026-08-15 background audit.)-    try:-        from handlers import win_focus-        _ufg = win_focus.capture_foreground()-        if _ufg and not win_focus.is_kicad_window(_ufg):-            win_focus.restore_foreground_persistent(_ufg, seconds=total_est + 90)-    except Exception:  # pylint: disable=broad-except-        pass+    # John, 2026-09-03: the tour is the one place the bridge DOES foreground+    # KiCad: "you must foreground otherwise the user doesn't know what window+    # you're demoing", but exactly ONCE per beat so the user keeps full control+    # of their desktop between beats. So: no persistent focus guardian here+    # (it would fight that single raise), and no push-to-back of the demo's+    # windows. Every other verb keeps the background contract.     done_weight = 0.0     try:         _panel_show(0, "playing")@@ -839,6 +953,8 @@ def _run_demo_background(kicad_info: dict) -> None:                       file=_sys.stderr)             r["step"] = s_             results.append(r)+            if r.get("ok"):+                r["foregrounded"] = _foreground_once(r.get("window") or r.get("windowTitle") or "")             shot = _beat_shot(kicad_info, f"Opening {s_.replace('_', ' ')}",                               r.get("window") or r.get("windowTitle") or "")             if shot: