Fusion - the Fusion 360 Bridge

@@ -715,6 +715,87 @@ def _popen_fusion_background(exe_path: str):     )  ++# ── The background envelope (John, 2026-08-14: "ALWAYS open in the background") ──────────────+# The launch guard above only protected fusion_start. But the focus thefts users actually feel+# happen DURING operations on a running Fusion: a cloud open activates the document window, a+# picker/recovery dialog pops, an import raises a chooser - all while the user is typing in+# another app. The envelope generalizes the guard so ANY window-raising verb runs inside it:+#   - foreground-lock raised for the duration (steals become taskbar flashes)+#   - a watcher demotes any Autodesk-family window that still grabs focus+#   - refcounted, so overlapping verbs nest and only the LAST one restores the user's settings+# Two demotion modes: minimize=True (launch: splash/seat windows vanish) and minimize=False+# (running ops: the main window is pushed to the BOTTOM of the z-order WITHOUT minimizing, so+# PrintWindow captures keep rendering it correctly).+_ENVELOPE_LOCK = threading.Lock()+_ENVELOPE_COUNT = 0+_ENVELOPE_SAVED = {}+++def begin_background_envelope(seconds: float, minimize: bool = False):+    """Guard the next `seconds` against Fusion-family focus theft. Fire-and-forget."""+    global _ENVELOPE_COUNT+    try:+        with _ENVELOPE_LOCK:+            _ENVELOPE_COUNT += 1+            if _ENVELOPE_COUNT == 1:+                _ENVELOPE_SAVED["fg"] = _get_fg_lock_timeout()+                _ENVELOPE_SAVED["anim"] = _get_min_animate()+                _set_fg_lock_timeout(300000)+                _set_min_animate(0)+        threading.Thread(target=_envelope_watch, args=(seconds, minimize), daemon=True).start()+    except Exception:+        pass+++def _envelope_watch(seconds: float, minimize: bool):+    try:+        user32 = ctypes.windll.user32+        HWND = ctypes.wintypes.HWND+        user32.GetForegroundWindow.restype = HWND+        user32.GetForegroundWindow.argtypes = []+        user32.GetWindowThreadProcessId.argtypes = [HWND, ctypes.POINTER(ctypes.wintypes.DWORD)]+        user32.GetWindowThreadProcessId.restype = ctypes.wintypes.DWORD+        user32.ShowWindowAsync.argtypes = [HWND, ctypes.c_int]+        user32.ShowWindowAsync.restype = ctypes.wintypes.BOOL+        user32.SetWindowPos.argtypes = [HWND, HWND, ctypes.c_int, ctypes.c_int,+                                        ctypes.c_int, ctypes.c_int, ctypes.c_uint]+        user32.SetWindowPos.restype = ctypes.wintypes.BOOL+        HWND_BOTTOM = HWND(1)+        SWP_FLAGS = 0x0001 | 0x0002 | 0x0010  # NOSIZE | NOMOVE | NOACTIVATE+        end = time.time() + seconds+        pids, last_refresh = set(), 0.0+        while time.time() < end:+            now = time.time()+            if now - last_refresh > 2.0:+                pids = _fusion_family_pids()+                last_refresh = now+            try:+                fg = user32.GetForegroundWindow()+                if fg and pids:+                    wpid = ctypes.wintypes.DWORD(0)+                    user32.GetWindowThreadProcessId(fg, ctypes.byref(wpid))+                    if wpid.value in pids:+                        if minimize:+                            user32.ShowWindowAsync(fg, _SW_SHOWMINNOACTIVE)+                        else:+                            user32.SetWindowPos(fg, HWND_BOTTOM, 0, 0, 0, 0, SWP_FLAGS)+            except Exception:+                pass+            time.sleep(0.1)+    except Exception:+        pass  # non-Windows / ctypes unavailable: envelope is a no-op, refcount still drains+    finally:+        global _ENVELOPE_COUNT+        with _ENVELOPE_LOCK:+            _ENVELOPE_COUNT -= 1+            if _ENVELOPE_COUNT == 0:+                if _ENVELOPE_SAVED.get("fg") is not None:+                    _set_fg_lock_timeout(_ENVELOPE_SAVED["fg"])+                if _ENVELOPE_SAVED.get("anim") is not None:+                    _set_min_animate(_ENVELOPE_SAVED["anim"])++ def _focus_guard(seconds: float = 150.0, restore_fg_lock=None, restore_min_anim=None):     """For `seconds`, demote any Autodesk-FAMILY window that steals the foreground. @@ -822,10 +903,13 @@ def ensure_fusion_running(fusion_info: dict, wait_addin: bool = False) -> dict |         # streams + CREATE_NO_WINDOW: the bridge is a detached, console-less child of         # AD, so inherited handles are invalid.         _popen_fusion_background(exe_path)-        # Reactive backstop: minimize any family window that still slips to foreground,-        # and restore the user's original foreground-lock + animation when done.-        _start_focus_guard(seconds=150.0, restore_fg_lock=_orig_fg_lock,-                           restore_min_anim=_orig_min_anim)+        # Reactive backstop via the refcounted envelope (minimize mode: splash/seat+        # windows vanish rather than being demoted). The envelope restores the user's+        # foreground-lock + animation when the LAST active guard drains, so it composes+        # with per-verb envelopes armed by the dispatcher.+        _set_fg_lock_timeout(_orig_fg_lock)   # envelope snapshots + re-raises itself+        _set_min_animate(_orig_min_anim)+        begin_background_envelope(150.0, minimize=True)     except Exception as e:         return {"success": False, "error": f"Failed to launch Fusion 360: {e}"} 

@@ -67,9 +67,26 @@ def expect(d, spec):         got = {v["name"] if isinstance(v, dict) else v for v in (d.get("verbs") or [])}         planned = set(PLAN.keys())         if got and (got - planned): errs.append("verbs described but UNTRIAGED in test plan: %s" % sorted(got - planned)[:6])+    if spec.get("pred") == "must_refuse" and not (d.get("success") is False or d.get("error")):+        errs.append("expected a refusal, got success")     if d.get("_testTimeout"): errs.append("timed out")     return errs ++def call_polled(target, verb, args, doc, poll_sec):+    """Fire a long open, then poll get_app_state until the doc is active (the bridge's own+    _timeoutHint prescribes exactly this instead of trusting the relay timeout)."""+    args = {k: v for k, v in args.items() if not k.startswith("_")}+    d, dt = call(target, verb, args, timeout=70)+    ok = (d.get("success") is True or d.get("status") == "ok")+    t0 = time.time()+    while time.time() - t0 < poll_sec:+        st, _ = call(target, "fusion_get_app_state", {}, timeout=45)+        if st.get("activeDocument") == doc:+            return {"success": True, "activeDocument": doc}, dt + (time.time() - t0)+        time.sleep(8)+    return (d if not ok else {"success": False, "error": "document never became active"}), dt + (time.time() - t0)+ # tier, args, expectation, [cleanup verb, cleanup args] P = {} def t(verb, tier, args=None, exp=None, note=""):@@ -88,7 +105,7 @@ t("fusion_aps_status", "daily", {}, {"ok": True}) t("fusion_aps_get_browser", "daily", {}, {"ok": True}) t("fusion_aps_search", "daily", {"query": "BQ25792", "limit": 5}, {"ok": True, "pred": "count_gt_0"}) t("fusion_aps_recent", "daily", {"limit": 5}, {"ok": True, "pred": "count_gt_0"})-t("fusion_aps_browse", "daily", {}, {"ok": True, "fields": ["items"]})+t("fusion_aps_browse", "daily", {}, {"ok": True, "fields": ["hubs"]}) t("fusion_aps_file_info", "daily", {"query": "BQ25792"}, {"ok": True, "fields": ["versionCount"]}) t("fusion_aps_versions", "weekly", {"query": "BQ25792"}, {"ok": True, "fields": ["versionCount"]}) t("fusion_aps_download", "weekly", {"query": "BQ25792", "saveDir": SCRATCH}, {"ok": True})@@ -113,10 +130,14 @@ t("fusion_mcp_enable", "manual", note="drives the prefs UI, takes foreground") t("fusion_prefs_open", "manual", note="opens a dialog on the user's screen") t("fusion_prefs_close", "manual") t("fusion_set_auto_update", "daily", {}, {"ok": True}, "no args = read-only")+t("fusion_dismiss_blocking_dialogs", "daily", {}, {"ok": True}, "no-op when nothing is blocking")+t("fusion_detect_layers", "weekly", {}, {"ok": True}, "board view active")+t("fusion_fetch_optimized_glb", "weekly", {}, {"ok": True}, "runs the stackup GLB fetch")+t("fusion_load_design_rules", "manual", note="mutates the board's DRC rules")  # ── stage 3: document flows on the canonical board ────────────────────────── OPEN = {"fileName": "BQ25792", "projectName": "Main", "folderPath": "Molecules/XRP/Power"}-t("fusion_open_cloud_file", "daily", OPEN, {"ok": True}, "the canonical demo board")+t("fusion_open_cloud_file", "daily", dict(OPEN, **{"_pollDoc": "BQ25792", "_pollSec": 180}), {"ok": True}, "fire-and-poll: cloud opens outlive the relay timeout") t("fusion_show_schematic", "daily", {}, {"ok": True}) t("fusion_show_2d_board", "daily", {}, {"ok": True}) t("fusion_show_3d_board", "daily", {}, {"ok": True})@@ -144,7 +165,7 @@ t("fusion_electron_select", "manual", note="selection state is user-visible") t("fusion_open_by_urn", "weekly", note="URN fetched from aps_search at runtime", args={"_dynamic": "urn_from_search"}) t("fusion_aps_open", "weekly", {"query": "BQ25792", "wait": False}, {"ok": True}) t("fusion_open_design", "manual", note="local-file open; superseded by cloud flows")-t("fusion_close_window", "daily", {}, {"ok": True}, "closes the test board tab, restores tab state")+t("fusion_close_window#noargs", "negative", {}, {"pred": "must_refuse"}, "dialog closer; hwnd required - refusal IS the pass") t("fusion_close_document", "manual"); t("fusion_close_all_documents", "manual", note="would close the user's own tabs")  # ── stage 4: exports (local writes to scratch) ──────────────────────────────@@ -205,7 +226,10 @@ def main():             results.append({"verb": name, "status": "SKIP", "why": "dynamic fixture not staged this run"})             continue         verb = name.split("#")[0]-        d, dt = call(a.target, verb, spec["args"])+        if "_pollDoc" in spec["args"]:+            d, dt = call_polled(a.target, verb, spec["args"], spec["args"]["_pollDoc"], spec["args"]["_pollSec"])+        else:+            d, dt = call(a.target, verb, spec["args"])         errs = expect(d, spec["exp"])         if name.startswith("fusion_search_cloud_files") or name.startswith("fusion_walk_cloud_tree"):             # must refuse: success:true here is the FAILURE@@ -215,6 +239,19 @@ def main():                         "why": "; ".join(errs), "sec": round(dt, 1)})         print("  %-42s %s %s" % (name, "PASS" if not errs else "FAIL", ("- " + "; ".join(errs)) if errs else "")) +    # Focus invariance (fusion-background-etiquette): the whole run must leave the user's+    # foreground alone. z=0 belonging to a Fusion-family window at the end = a steal happened.+    fg, _ = call(a.target, "desktop_list_windows", {}, timeout=45)+    try:+        wins = fg.get("windows") or []+        top = min(wins, key=lambda w: w.get("z", 999)) if wins else {}+        stolen = "Autodesk Fusion" in str(top.get("title", "")) or str(top.get("title", "")).strip() == "Fusion360"+        results.append({"verb": "focus_invariance", "status": "FAIL" if stolen else "PASS",+                        "why": ("foreground is %r after the run" % top.get("title")) if stolen else ""})+        print("  %-42s %s" % ("focus_invariance", "FAIL" if stolen else "PASS"))+    except Exception as e:+        results.append({"verb": "focus_invariance", "status": "SKIP", "why": str(e)})+     passed = sum(1 for r in results if r["status"] == "PASS")     failed = [r for r in results if r["status"] == "FAIL"]     stamp = datetime.datetime.now().strftime("%Y-%m-%d_%H%M")