app
Fusion - the Fusion 360 Bridge
Public Made by Adomby adom
Drive Autodesk Fusion 360 from the cloud via Adom Bridge: component libraries, IPC package generation, board layout, exports (STEP/Gerbers/BOM/CPL), fast APS cloud search, and parametric modeling.
← Commit history
server.pyadded+1412@@ -0,0 +1,1412 @@+#!/usr/bin/env python3+"""Adom Fusion 360 Bridge Server — localhost HTTP server for Fusion 360 integration.++Receives commands from the Adom Desktop (Tauri app) and controls+Fusion 360 via Win32 API, os.startfile, and the AdomBridge add-in.++Usage:+ python server.py+ python server.py --port 8773+"""++import json+import os+import sys+import traceback+import urllib.request+import urllib.error+from http.server import ThreadingHTTPServer, BaseHTTPRequestHandler+from pathlib import Path++# Ensure the plugin root is on the path+sys.path.insert(0, str(Path(__file__).parent))++from fusion_detect import detect_fusion, ensure_fusion_running, wait_for_addin, _is_fusion_running+from handlers.open_design import handle_open_design+from handlers.close_fusion import handle_close_fusion+from handlers.dismiss_recovery import dismiss_recovery_dialog+from handlers.fusion_ui import (+ screenshot_fusion_window,+ screenshot_hwnd,+ click_fusion,+ send_key_to_fusion,+ close_window,+ get_fusion_window_info,+)+from handlers.dialog_classify import classify_blocking_dialogs+from install_addin import install as install_addin+import aps++DEFAULT_PORT = 8773+ADDIN_PORT = 8774++# Read once at load so /status reports the same version we ship + bump.+try:+ BRIDGE_VERSION = (Path(__file__).parent / "BRIDGE_VERSION").read_text(encoding="utf-8").strip()+except Exception:+ BRIDGE_VERSION = "unknown"++# Populated at startup+fusion_info = None+++def _handle_launch(fusion_info: dict, args: dict) -> dict:+ """Launch Fusion 360 and optionally wait for the AdomBridge add-in."""+ if not fusion_info.get("installed"):+ return {+ "success": False,+ "error": "Fusion 360 is not installed. "+ "Download it from https://www.autodesk.com/products/fusion-360 "+ "(free for personal/hobby use).",+ "errorCode": "fusion_not_installed",+ "_hint": "Fusion 360 cannot be installed programmatically. Report this to the user and ask them to install it from https://www.autodesk.com/products/fusion-360, then retry.",+ }++ if _is_fusion_running():+ addin_ok = wait_for_addin(timeout=5)+ return {+ "success": True,+ "output": "Fusion 360 is already running.",+ "addinConnected": addin_ok,+ "alreadyRunning": True,+ "resolvedPath": fusion_info.get("exe_path", ""),+ }++ # Relocate any crash recovery files BEFORE launching Fusion.+ # This prevents the "Recovered Documents" dialog from appearing at all.+ # Files are moved to ~/.adom/recovery/fusion/<timestamp>/ (not deleted).+ from handlers.dismiss_recovery import relocate_recovery_files+ reloc = {"moved": 0, "dest": None}+ try:+ reloc = relocate_recovery_files()+ except Exception:+ pass++ # Launch Fusion and wait for it + add-in+ err = ensure_fusion_running(fusion_info, wait_addin=True)+ if err:+ return err++ addin_ok = wait_for_addin(timeout=5)++ # Auto-dismiss the "Recovered Documents" dialog if it appeared on launch.+ # Even though we relocated files above, Fusion may still show recovery+ # dialogs if files were locked or if Fusion cached them.+ recovery_dismissed = dismiss_recovery_dialog()++ # Probe the add-in to check if a modal dialog is blocking.+ # If Fusion has a "Recovered Documents" or other modal up, the add-in's+ # main thread will be blocked and commands will timeout. We detect this+ # PROACTIVELY so the AI knows immediately, rather than waiting for a+ # 30-second command timeout.+ dialog_blocking = False+ if addin_ok:+ dialog_blocking = _check_main_thread_blocked()++ output = "Fusion 360 launched successfully."+ if reloc["moved"] > 0:+ output += (+ f" Relocated {reloc['moved']} recovery file(s) to {reloc['dest']}"+ " (preserved for manual recovery if needed)."+ )+ if recovery_dismissed:+ output += " Dismissed 'Recovered Documents' dialog."+ if dialog_blocking:+ output = (+ "Fusion 360 launched but a MODAL DIALOG is blocking the UI. "+ "This is likely the 'Recovered Documents' dialog from a previous crash. "+ "Take a screenshot with desktop_screenshot_window to see what's showing, "+ "then dismiss it manually or use fusion_dismiss_recovery."+ )++ return {+ "success": True,+ "output": output,+ "addinConnected": addin_ok,+ "dialogBlocking": dialog_blocking,+ "recoveryDialogDismissed": recovery_dismissed,+ "recoveryFilesRelocated": reloc["moved"],+ "recoveryFilesPath": reloc["dest"],+ "resolvedPath": fusion_info.get("exe_path", ""),+ }+++def _handle_dismiss_recovery(fusion_info: dict, args: dict) -> dict:+ """Dismiss the 'Recovered Documents' dialog if it's showing.++ Recovery files are relocated to ~/.adom/recovery/fusion/<timestamp>/+ instead of being deleted, preserving the user's safety net.+ """+ from handlers.dismiss_recovery import relocate_recovery_files++ # Relocate files first (dismiss_recovery_dialog also does this,+ # but we capture the result here for reporting)+ reloc = {"moved": 0, "dest": None}+ try:+ reloc = relocate_recovery_files()+ except Exception:+ pass++ dismissed = dismiss_recovery_dialog()++ parts = []+ if dismissed:+ parts.append("Dismissed recovery dialog(s).")+ else:+ parts.append("No recovery dialog found.")+ if reloc["moved"] > 0:+ parts.append(+ f"Relocated {reloc['moved']} recovery file(s) to {reloc['dest']}. "+ "The user can restore them from there if needed."+ )++ return {+ "success": True,+ "output": " ".join(parts),+ "dismissed": dismissed,+ "recoveryFilesRelocated": reloc["moved"],+ "recoveryFilesPath": reloc["dest"],+ }+++def _post_open_screenshot(result: dict, settle_time: float = 2.0) -> dict:+ """Auto-screenshot Fusion after an open/switch command.++ Waits for Fusion to settle, then captures the full window (including+ any Qt dialogs or CEF overlays the API can't see). Attaches the+ screenshot path to the result so the AI can inspect it immediately+ without making a separate call.++ The screenshot is embedded into the result's 'data' field because the+ Tauri relay only forwards 'output', 'data', 'success', and 'error'.++ Also checks for blocking dialogs and warns in the response.+ """+ import re+ import time+ time.sleep(settle_time)++ # Check if main thread is blocked (dialog present)+ blocked = _check_main_thread_blocked()++ # Capture main Fusion window+ screenshot = screenshot_fusion_window()+ screenshot_info = {"screenshots": []}++ if screenshot.get("success"):+ screenshot_info["screenshots"].append({+ "type": "main_window",+ "savedTo": screenshot["savedTo"],+ "sizeKB": screenshot["sizeKB"],+ })+ else:+ screenshot_info["error"] = screenshot.get("error", "Screenshot failed")++ # Also capture ALL dialog windows — these are separate Qt windows+ # that the main window screenshot won't show+ window_info = get_fusion_window_info()+ if window_info.get("success"):+ dialogs = window_info.get("dialogs", [])+ for dialog in dialogs:+ try:+ title = dialog.get("title", "")+ hwnd = dialog.get("hwnd")+ # Skip tiny/hidden windows and banners (only screenshot real dialogs)+ rect = dialog.get("rect", {})+ w = rect.get("width", 0)+ h = rect.get("height", 0)+ if w < 50 or h < 50:+ continue+ # Screenshot this dialog — sanitize label for filesystem+ label = re.sub(r'[<>:"/\\|?*]', '', title).replace(" ", "_")[:30]+ dialog_ss = screenshot_hwnd(hwnd, label=f"dialog-{label}")+ if dialog_ss.get("success"):+ screenshot_info["screenshots"].append({+ "type": "dialog",+ "title": title,+ "hwnd": hwnd,+ "savedTo": dialog_ss["savedTo"],+ "sizeKB": dialog_ss["sizeKB"],+ })+ except Exception as e:+ screenshot_info.setdefault("errors", []).append(+ f"Failed to screenshot dialog '{title}': {e}"+ )++ screenshot_info["message"] = (+ f"Auto-captured {len(screenshot_info['screenshots'])} Fusion window(s) after open. "+ "READ EACH screenshot file to check for blocking dialogs "+ "('What to design?', 'PCB out of date', 'Save changes?'). "+ "Dismiss with fusion_send_key {\"key\": \"escape\"} if needed."+ )++ if blocked:+ screenshot_info["dialogBlocking"] = True+ screenshot_info["message"] = (+ "WARNING: A modal dialog is blocking Fusion. "+ "READ the screenshot to identify and dismiss it. " ++ screenshot_info["message"]+ )++ # Inject screenshot into the result's top-level 'data' dict.+ # The Tauri relay constructs its output as: json!({"message": output, "data": data})+ # so anything we put in result["data"] gets forwarded to the CLI.+ result.setdefault("data", {})+ if not isinstance(result["data"], dict):+ result["data"] = {}+ result["data"]["postOpenScreenshot"] = screenshot_info++ return result+++def _handle_open_cloud_file(fusion_info: dict, args: dict) -> dict:+ """Open a cloud file with post-open dialog detection + auto-screenshot.++ Proxies to the add-in's open_cloud_file command, then:+ 1. Waits for Fusion to settle+ 2. Checks if a modal dialog blocked the main thread+ 3. Auto-screenshots the Fusion window so the AI can see what happened+ """+ # Send the open command — cloud files can take 30+ seconds to download/open+ result = _proxy_to_addin("open_cloud_file", args, timeout=45)++ if not result.get("success"):+ # Failure is often a modal blocking the add-in's main thread (multi-design+ # "Select Electronics Design File" picker, update nag, recovery, etc.) — which+ # otherwise surfaces as a misleading "add-in not responding". Classify the+ # actual blocking dialog so the caller gets the real cause + resolution.+ dialogs = classify_blocking_dialogs()+ if dialogs:+ result["dialogBlocking"] = True+ result["blockingDialogs"] = dialogs+ result["hint"] = (+ "A modal dialog is blocking Fusion — see blockingDialogs for the "+ "classified cause + resolution. The multi-link 'Select Electronics "+ "Design File' picker is a CEF dialog that must be resolved on the "+ "desktop (its list isn't keyboard/Win32 navigable); recovery prompts "+ "use fusion_dismiss_recovery; an update nag must be cleared on the desktop."+ )+ # Screenshot even on failure — shows what went wrong+ return _post_open_screenshot(result, settle_time=1.0)++ return _post_open_screenshot(result)+++def _handle_open_with_screenshot(fusion_info: dict, args: dict, command: str) -> dict:+ """Proxy an open/switch command to the add-in, then auto-screenshot.++ Used for open_schematic, open_board, show_3d_board, show_2d_board —+ any command that changes the Fusion UI state and might trigger a dialog.+ """+ result = _proxy_to_addin(command, args, timeout=30)+ if result.get("success"):+ return _post_open_screenshot(result)+ return result+++def _handle_screenshot_fusion(fusion_info: dict, args: dict) -> dict:+ """Screenshot Fusion main window or a specific dialog by HWND."""+ hwnd = args.get("hwnd") or args.get("dialogHwnd")+ if hwnd:+ result = screenshot_hwnd(int(hwnd), label="dialog")+ else:+ result = screenshot_fusion_window()+ if result.get("success"):+ result["output"] = json.dumps({"savedTo": result["savedTo"], "sizeKB": result["sizeKB"]})+ return result+++def _handle_click_fusion(fusion_info: dict, args: dict) -> dict:+ """Click at coordinates within Fusion window or a specific dialog HWND."""+ x = args.get("x", 0.5)+ y = args.get("y", 0.5)+ relative = args.get("relative", True)+ hwnd = args.get("hwnd") or args.get("dialogHwnd")+ result = click_fusion(x, y, relative, hwnd=hwnd)+ if result.get("success"):+ result["output"] = json.dumps(result.get("clickedAt", {}))+ return result+++def _handle_send_key(fusion_info: dict, args: dict) -> dict:+ """Send a key to Fusion or a specific dialog via SendInput."""+ key = args.get("key", "")+ if not key:+ return {"success": False, "error": "No key specified. Use 'enter', 'escape', 'tab', etc."}+ hwnd = args.get("hwnd") or args.get("dialogHwnd")+ result = send_key_to_fusion(key, hwnd=hwnd)+ if result.get("success"):+ result["output"] = f"Sent key '{key}' to Fusion."+ return result+++def _handle_close_window(fusion_info: dict, args: dict) -> dict:+ """Close a dialog window by sending WM_CLOSE (equivalent to clicking X).++ Unlike Escape, this works on dialogs that don't have Cancel/Escape handling+ (e.g. the Recovered Documents dialog). Does NOT force-kill — the dialog can+ still intercept WM_CLOSE and prompt for confirmation.+ """+ hwnd = args.get("hwnd")+ if not hwnd:+ return {"success": False, "error": "Missing required arg: hwnd"}+ result = close_window(int(hwnd))+ if result.get("success"):+ result["output"] = f"Sent WM_CLOSE to hwnd {hwnd}."+ return result+++def _handle_window_info(fusion_info: dict, args: dict) -> dict:+ """Get Fusion window info including any dialog windows."""+ result = get_fusion_window_info()+ if result.get("success"):+ dialogs = result.get("dialogs", [])+ dialog_info = [f" hwnd={d['hwnd']}: {d['title']}" for d in dialogs]+ output_parts = [+ f"Main: hwnd={result['hwnd']}, title='{result['title']}'",+ ]+ if dialogs:+ output_parts.append(f"Dialogs ({len(dialogs)}):")+ output_parts.extend(dialog_info)+ else:+ output_parts.append("No dialog windows detected.")+ if dialogs:+ hint = (f"{len(dialogs)} dialog(s) detected. Screenshot each via "+ "desktop_screenshot_window {{\"hwnd\":<hwnd>}}, then dismiss with "+ "fusion_dismiss_blocking_dialogs or fusion_send_key/fusion_close_window.")+ else:+ hint = "No dialogs — Fusion is clear. Safe to run fusion_* commands."+ result["output"] = json.dumps({+ "hwnd": result["hwnd"],+ "title": result["title"],+ "rect": result["rect"],+ "dialogs": dialogs,+ "message": "\n".join(output_parts),+ "_hint": hint,+ })+ return result+++def _handle_screenshot_all_fusion(fusion_info: dict, args: dict) -> dict:+ """Screenshot Fusion main window AND all dialog windows.++ Returns paths to all captured screenshots so the AI can see+ CEF overlay dialogs and separate Qt dialog windows.+ """+ screenshots = []++ # 1. Capture main Fusion window from screen (captures CEF overlays)+ main_result = screenshot_fusion_window()+ if main_result.get("success"):+ screenshots.append({+ "type": "main_window",+ "path": main_result["savedTo"],+ "sizeKB": main_result["sizeKB"],+ })++ # 2. List all Qt dialog windows (AI should use desktop_screenshot_window for each)+ info = get_fusion_window_info()+ if info.get("success") and info.get("dialogs"):+ for dialog in info["dialogs"]:+ screenshots.append({+ "type": "dialog",+ "title": dialog["title"],+ "hwnd": dialog["hwnd"],+ "rect": dialog["rect"],+ "hint": f"Use desktop_screenshot_window with hwnd={dialog['hwnd']} to capture this dialog.",+ })++ return {+ "success": True,+ "screenshots": screenshots,+ "dialogCount": len(info.get("dialogs", [])) if info.get("success") else 0,+ "_hint": f"Captured {len(screenshots)} screenshot(s). Screenshots saved on Windows — "+ "use pull_file to get them to Docker, or Read tool to view each path directly. "+ "If dialogs are present, dismiss with fusion_dismiss_blocking_dialogs.",+ }+++def _handle_relocate_recovery(fusion_info: dict, args: dict) -> dict:+ """Relocate Fusion crash recovery files to ~/.adom/recovery/fusion/.++ Call this proactively before launching Fusion to prevent recovery+ dialogs from appearing. Files are preserved (not deleted) so the+ user can manually restore them if needed.+ """+ from handlers.dismiss_recovery import relocate_recovery_files+ try:+ reloc = relocate_recovery_files()+ except Exception as exc:+ return {+ "success": False,+ "error": f"Failed to relocate recovery files: {exc}",+ }++ if reloc["moved"] > 0:+ return {+ "success": True,+ "output": (+ f"Relocated {reloc['moved']} recovery file(s) to {reloc['dest']}. "+ "These files are preserved — the user can restore them from that "+ "directory if they need to recover unsaved work."+ ),+ "moved": reloc["moved"],+ "dest": reloc["dest"],+ "sources": reloc["sources"],+ }+ return {+ "success": True,+ "output": "No crash recovery files found.",+ "moved": 0,+ }+++COMMAND_HANDLERS = {+ "open_design": handle_open_design,+ "close": handle_close_fusion,+ "launch": _handle_launch,+ "start": _handle_launch, # alias — CLI's fusion_start delegates here on Docker+ "dismiss_recovery": _handle_dismiss_recovery,+ "relocate_recovery": _handle_relocate_recovery,+ "open_cloud_file": _handle_open_cloud_file,+ "screenshot_fusion": _handle_screenshot_fusion,+ "click_fusion": _handle_click_fusion,+ "send_key": _handle_send_key,+ "close_window": _handle_close_window,+ "window_info": _handle_window_info,+ "screenshot_all": _handle_screenshot_all_fusion,+ "addin_status": lambda fi, args: _handle_addin_status(),+ # APS cloud search (pure HTTPS, works with Fusion closed). Lives in+ # COMMAND_HANDLERS so it returns BEFORE any Fusion-running gate.+ "aps_status": lambda fi, args: aps.handle_status(args),+ "aps_set_client_id": lambda fi, args: aps.handle_set_client_id(args),+ "aps_signin": lambda fi, args: aps.handle_signin(args),+ "aps_search": lambda fi, args: aps.handle_search(args),+ "aps_browse": lambda fi, args: aps.handle_browse(args),+ "aps_recent": lambda fi, args: aps.handle_recent(args),+ "aps_file_info": lambda fi, args: aps.handle_file_info(args),+ "aps_versions": lambda fi, args: aps.handle_versions(args),+ "aps_download": lambda fi, args: aps.handle_download(args),+ "aps_create_folder": lambda fi, args: aps.handle_create_folder(args),+ "aps_upload": lambda fi, args: aps.handle_upload(args),+ "aps_open": lambda fi, args: _handle_aps_open(fi, args),+ "aps_get": lambda fi, args: aps.handle_get(args),+}+++# Surfaced inline so the AI doesn't draw the WRONG conclusion (it has, repeatedly):+# a "Read Only" / expired / trial / personal-use Fusion CAN open + view + browse files+# (Basic Access ~365 days) — it only blocks save/export/modify. NEVER blame a failed/slow+# OPEN on the subscription; the cause is the open path (URN resolution, the Electronics+# design picker, a slow assembly download, a stuck main thread).+_READONLY_OPEN_NOTE = ("NOTE: a 'Read Only'/expired/trial Fusion still opens+views files — if "+ "it never opens, debug the open path (URN/picker/slow assembly), NOT the "+ "license. Only save/export are blocked in read-only.")+++def _handle_aps_open(fusion_info: dict, args: dict) -> dict:+ """Search the cloud (APS) for a file, then OPEN the best match in Fusion.++ APS finds it instantly by name across the whole team hub; the add-in's+ open_cloud_file (projectName + fileName + fileExtension) opens it.+ """+ query = (args.get("query") or args.get("fileName") or "").strip()+ if not query:+ return {"success": False, "error": "Missing query/fileName."}+ match = aps.find_one(query)+ if not match:+ return {"success": False, "error": f"No cloud file matching '{query}'.",+ "_hint": "Run fusion_aps_search to see candidates, or refine the query. "+ + _READONLY_OPEN_NOTE}+ # Fusion cloud displayNames are NOT filenames — do NOT splitext (it mangled+ # "...(1.6mm gasket)" into a bogus extension). Pass the full name.+ name = match.get("name") or ""+ open_args = {"projectName": match.get("projectName"), "fileName": name}+ if not fusion_info.get("installed") or not _is_fusion_running():+ return {"success": True, "output": f"Found '{name}' in project {match.get('projectName')}.",+ "data": {"match": match, "openArgs": open_args},+ "_hint": "Fusion isn't running — call fusion_start, then fusion_aps_open again to open it."}+ # Open by the EXACT file URN (works for any nesting). The FIRST cloud-open can+ # take 60-90s (download) — longer than AD's relay timeout — so FIRE it in the+ # background and return immediately. Caller polls fusion_get_app_state. Unless+ # {wait:true} is passed (then block and return the open result).+ urn = match.get("id")++ def _do_open():+ r = _proxy_to_addin("open_by_urn", {"urn": urn}, timeout=180)+ if not r.get("success"):+ _proxy_to_addin("open_cloud_file", open_args, timeout=120) # by-name fallback+ return r++ if args.get("wait"):+ result = _do_open()+ result.setdefault("data", {})+ if isinstance(result.get("data"), dict):+ result["data"]["match"] = match+ return _post_open_screenshot(result) if result.get("success") else result++ import threading as _threading+ _threading.Thread(target=_do_open, daemon=True).start()+ return {+ "success": True,+ "output": f"Found '{name}' in project {match.get('projectName')} — opening in Fusion.",+ "data": {"match": match, "opening": True},+ "statusVerb": "fusion_get_app_state", # AD 1.9.9 convention — poll this for completion+ "_hint": "The cloud file is opening in Fusion in the background (first open can take "+ "~60-90s; large ASSEMBLIES download all referenced parts and take longer). "+ "Poll fusion_get_app_state until activeDocument is the file. Pass "+ "{\"wait\": true} to block instead. " + _READONLY_OPEN_NOTE,+ }++# Commands that are proxied to the Fusion add-in (port 8774).+# Keys are the CLI-facing names (after stripping "fusion_" prefix).+ADDIN_COMMANDS = {+ "get_app_state",+ "document_info",+ "activate_document",+ "import_step", # add-in calls this "import_file" — mapped below+ "export_step", "export_stl", "export_3mf", "export_f3d", "export_fbx", "export_usdz",+ "export_dxf", "export_dwg", "export_iges", "export_obj", "export_sat", "export_skp",+ "get_design_info",+ "get_parameters", "set_parameter",+ "take_screenshot",+ # Electronics commands (EAGLE via Electron.run)+ "electron_run", "execute_text_command",+ "open_electronics", "list_text_commands",+ # Electronics source export (.fsch, .fbrd, .flbr via Document.CopyToDesktop)+ "export_source",+ # EAGLE-format export (.sch, .brd — extracted from .fsch/.fbrd ZIP container)+ "export_eagle_source",+ # Library file commands (EXPORT SCRIPT — open_lbr is orchestrated at bridge level)+ "export_lbr",+ # Document management+ "close_document",+ "close_all_documents",+ # Electronics file opening (open_schematic, open_board, show_3d_board, show_2d_board+ # are orchestrated at bridge level for auto-screenshot — not in this set)+ # Board data query+ "board_info",+ # Open any cloud file directly by its APS/Fusion URN (any folder depth).+ # fusion_aps_open also proxies this internally (fire-and-poll); registering it+ # here makes the direct fusion_open_by_urn verb work too instead of being+ # rejected as "Unknown command".+ "open_by_urn",+ # In-app parametric modeling — run an adsk.fusion script in the live session+ # (free path to programmatic CAD; the APS Fusion Automation API is the paid+ # cloud alternative).+ "run_modeling_script",+ # Cloud document management+ "save_to_cloud",+ "list_cloud_projects",+ "list_cloud_files",+ "delete_cloud_file",+ "create_cloud_folder",+ # open_cloud_file is orchestrated at bridge level (not direct proxy)+ # to detect blocking dialogs after open+ "check_recovery",+ "search_cloud_files",+ "walk_cloud_tree",+ "export_cloud_file",+ # Manufacturing exports+ "export_bom",+ "export_cpl",+ "export_gerbers",+ "set_design_rules",+ "export_board_image",+ "detect_layers",+}++# Map CLI command names → add-in command names (where they differ)+ADDIN_COMMAND_MAP = {+ "import_step": "import_file",+}++# Per-command timeout overrides for _proxy_to_addin. Heavy 3D exports on+# panelized boards (100+ placements) can take 120s+. The add-in-side+# timeout in http_server.py should be the source of truth; these are+# matched to that so urllib doesn't cut off before the add-in does.+ADDIN_COMMAND_TIMEOUTS = {+ "export_step": 300,+ "export_iges": 300,+ "export_sat": 300,+ "export_stl": 300,+ "export_3mf": 300,+ "export_usdz": 300,+ "export_obj": 300,+ "export_f3d": 300,+ "export_fbx": 300,+ "export_skp": 300,+ "export_dxf": 120,+ "export_dwg": 120,+ "export_gerbers": 180,+ "export_bom": 60,+ "export_cpl": 60,+ "export_board_image": 60,+ "close_all_documents": 60,+ # Cloud tree walker / search — can hit hundreds of folders on large projects.+ # Must match or exceed http_server.py PER_COMMAND_TIMEOUT so urllib doesn't+ # cut off before the add-in does.+ "walk_cloud_tree": 600,+ "search_cloud_files": 180,+ # Opening a cloud design downloads + loads it; large assemblies and+ # electronics/PCB designs (which spin up the Electronics editor) can take+ # minutes. fusion_aps_open's fire-and-poll path is the preferred way in for+ # these — but when open_by_urn is proxied synchronously, give it room.+ "open_by_urn": 240,+ # Modeling scripts can build many features; give them room without being unbounded.+ "run_modeling_script": 180,+}++# ── Busy gate: prevent command stacking during long-running add-in work ──+# When walk_cloud_tree or search_cloud_files is running, the Fusion main thread+# is blocked for 30-300+ seconds. Any add-in command sent during that time would+# pile up behind _main_thread_lock in http_server.py, eating HTTP threads and+# potentially crashing the host. The gate rejects those commands immediately at+# the bridge level with progress info, BEFORE they reach the add-in.+import threading as _threading+import time as _time++_long_command_lock = _threading.Lock()+_long_command = None # None or {"command": str, "startedAt": float}++LONG_RUNNING_COMMANDS = {"walk_cloud_tree", "search_cloud_files"}+++def _set_long_command(command: str):+ with _long_command_lock:+ global _long_command+ _long_command = {"command": command, "startedAt": _time.time()}+++def _clear_long_command():+ with _long_command_lock:+ global _long_command+ _long_command = None+++def _get_long_command() -> dict | None:+ with _long_command_lock:+ if _long_command is None:+ return None+ return dict(_long_command)+++def _get_busy_progress() -> dict | None:+ """Poll the add-in /status endpoint for walkProgress during a long command.++ Uses /status (not /health) because it's lighter and proven reliable under+ GIL contention during heavy walks. 3s timeout matches _check_addin_status.+ """+ status = _check_addin_status(timeout=3.0)+ if status and status.get("walkProgress"):+ return status["walkProgress"]+ return None+++def _check_main_thread_blocked() -> bool:+ """Quick check: is Fusion's main thread blocked by a modal dialog?++ Sends a fast command (get_app_state) with a short timeout. If the add-in's+ HTTP server responds but the command times out, a modal dialog is blocking.+ Returns True if blocked, False if responsive.+ """+ try:+ body = json.dumps({"command": "get_app_state", "args": {}}).encode("utf-8")+ req = urllib.request.Request(+ f"http://127.0.0.1:{ADDIN_PORT}/command",+ data=body,+ headers={"Content-Type": "application/json"},+ method="POST",+ )+ with urllib.request.urlopen(req, timeout=3) as resp:+ result = json.loads(resp.read())+ # If we got a response, main thread is fine+ if result.get("success"):+ return False+ # Add-in responded but with an error — check if it's a timeout+ if "timed out" in result.get("error", "").lower():+ return True+ return False+ except Exception as e:+ # HTTP timeout = main thread blocked (HTTP server is up but command didn't complete)+ if "timed out" in str(e).lower() or "timeout" in str(e).lower():+ return True+ # Connection refused = add-in not running (different problem)+ return False+++def _probe_addin(timeout: float = 0.5) -> dict | None:+ """Check if the Fusion add-in HTTP server is running.++ Uses a short timeout (default 0.5s) to avoid blocking the /health endpoint.+ The add-in runs on localhost so if it's up, it responds in <50ms.+ """+ try:+ req = urllib.request.Request(f"http://127.0.0.1:{ADDIN_PORT}/health", method="GET")+ with urllib.request.urlopen(req, timeout=timeout) as resp:+ return json.loads(resp.read())+ except Exception:+ return None+++def _check_addin_status(timeout: float = 3.0) -> dict | None:+ """GET /status from the add-in. Returns dict or None.++ This is the cross-bridge busy probe — reads add-in busy state without+ acquiring the main thread lock. Safe to call from any bridge/container.+ Usually ~50ms, but during heavy walks GIL contention can push to 1-2s.+ """+ try:+ req = urllib.request.Request(f"http://127.0.0.1:{ADDIN_PORT}/status", method="GET")+ with urllib.request.urlopen(req, timeout=timeout) as resp:+ return json.loads(resp.read())+ except Exception:+ return None+++def _handle_addin_status() -> dict:+ """Bridge-level handler for addin_status — wraps /status in standard format."""+ status = _check_addin_status(timeout=3.0) or {"busy": False}+ return {"success": True, "output": json.dumps(status), **status}+++def _proxy_to_addin(command: str, args: dict, timeout: int = 30) -> dict:+ """Proxy a command to the Fusion add-in HTTP server.++ On timeout, probes /health to distinguish:+ - Add-in alive but main thread blocked (modal dialog) → distinct error+ - Add-in HTTP server crashed → connection error+ """+ body = json.dumps({"command": command, "args": args}).encode("utf-8")+ req = urllib.request.Request(+ f"http://127.0.0.1:{ADDIN_PORT}/command",+ data=body,+ headers={"Content-Type": "application/json"},+ method="POST",+ )+ try:+ with urllib.request.urlopen(req, timeout=timeout) as resp:+ result = json.loads(resp.read())++ # Check if the add-in itself returned a timeout (main thread didn't respond)+ if (not result.get("success")+ and "timed out" in result.get("error", "").lower()):+ return _diagnose_addin_timeout(command, result)++ return result++ except urllib.error.URLError as e:+ if "Connection refused" in str(e) or "No connection" in str(e):+ return {+ "success": False,+ "error": "Fusion 360 AdomBridge add-in not running. "+ "Install it with: python plugins/fusion360/install_addin.py, "+ "then restart Fusion 360.",+ }+ # Could be a socket timeout — the HTTP request itself took too long+ if "timed out" in str(e).lower() or "timeout" in str(e).lower():+ return _diagnose_addin_timeout(command, {"error": str(e)})+ return {"success": False, "error": f"Add-in request failed: {e}"}+ except Exception as e:+ if "timed out" in str(e).lower() or "timeout" in str(e).lower():+ return _diagnose_addin_timeout(command, {"error": str(e)})+ return {"success": False, "error": f"Add-in request failed: {e}"}+++def _diagnose_addin_timeout(command: str, original_result: dict) -> dict:+ """After a command timeout, probe /health to determine the cause.++ Returns a distinct error if the add-in is alive but its main thread+ is blocked (e.g. by a modal dialog in Fusion). Always captures+ auto-screenshots of Fusion windows on timeout so the AI can see+ what dialog is blocking.+ """+ # Auto-screenshot on timeout — the most common cause is a blocking dialog+ # that's invisible to the API. Capture Fusion windows so the AI can+ # identify and dismiss the dialog.+ timeout_screenshots = {}+ try:+ timeout_screenshots = _post_open_screenshot(command)+ except Exception:+ pass # Best effort — don't let screenshot failure mask the real error++ # Identify WHICH modal is blocking (titles enumerate over Win32 even while the+ # add-in's main thread is stuck). Turns the opaque "add-in not responding" into+ # an actionable cause + resolution — and stops the needless restart loop.+ blocking_dialogs = classify_blocking_dialogs()++ health = _probe_addin(timeout=2.0)+ if health and health.get("status") == "ok":+ main_thread = health.get("main_thread", "unknown")+ pending = health.get("pending_commands", 0)+ if main_thread == "blocked" or pending > 0:+ # Before assuming a modal dialog, check /status — if the add-in+ # is busy with a known command, it's working (not dialog-blocked).+ status = _check_addin_status(timeout=3.0)+ if status and status.get("busy"):+ busy_cmd = status.get("busyCommand", "unknown")+ elapsed = status.get("elapsedSeconds", 0)+ walk = status.get("walkProgress")+ resp = {+ "success": False,+ "error": f"Fusion main thread busy — {busy_cmd} running for {elapsed}s.",+ "errorCode": "main_thread_busy",+ "busyCommand": busy_cmd,+ "elapsedSeconds": elapsed,+ "_hint": (+ "Add-in is busy with a long-running command. Do NOT retry add-in commands. "+ "Do NOT press Escape — the add-in is working, not stuck on a dialog. "+ "Commands that still work: fusion_window_info, fusion_screenshot_fusion, "+ "fusion_click_fusion, fusion_send_key, fusion_close_window."+ ),+ }+ if walk:+ resp["progress"] = walk+ return resp++ # If we recognized the blocking modal, name it and give the precise fix+ # instead of the generic "read the screenshots" guidance.+ if blocking_dialogs:+ titles = ", ".join(f"'{d['title']}' ({d['category']})" for d in blocking_dialogs)+ resolutions = " ".join(dict.fromkeys(d["resolution"] for d in blocking_dialogs))+ message = (f"The AdomBridge add-in is alive but its main thread is blocked by a "+ f"modal dialog: {titles}. This is NOT an add-in crash — do not restart "+ f"Fusion. {resolutions}")+ else:+ message = (f"The AdomBridge add-in is alive but its main thread is not "+ f"responding (status: {main_thread}, pending: {pending}). "+ f"Fusion 360 likely has a modal dialog open (Document Recovery, "+ f"error, or update prompt) that is blocking execution. "+ f"READ the screenshots to identify the dialog, then dismiss "+ f"with fusion_send_key {{\"key\": \"escape\"}} or "+ f"fusion_send_key {{\"key\": \"tab\"}} + {{\"key\": \"enter\"}}.")+ return {+ "success": False,+ "error": "addin_main_thread_blocked",+ "message": message,+ "blockingDialogs": blocking_dialogs,+ "data": {+ "command": command,+ "health": health,+ "blockingDialogs": blocking_dialogs,+ "postOpenScreenshot": timeout_screenshots,+ },+ }+ # Health says responsive but command still timed out — unusual+ return {+ "success": False,+ "error": "addin_command_timeout",+ "message": f"Command '{command}' timed out but the add-in reports main thread "+ f"is {main_thread}. The command may be long-running or stuck.",+ "data": {+ "command": command,+ "health": health,+ "postOpenScreenshot": timeout_screenshots,+ },+ }++ # Health probe failed — add-in HTTP server is down+ return {+ "success": False,+ "error": "AdomBridge add-in not responding. It may have crashed. "+ "Restart Fusion 360 or re-enable the add-in: "+ "UTILITIES > ADD-INS > AdomBridge > Run",+ }+++def _orchestrate_open_lbr(args: dict) -> dict:+ """Open an EAGLE .lbr library file in Fusion 360 Electronics.++ Uses Document.newDesignFromLocal (via the add-in's execute_text_command)+ to open the .lbr file, which auto-switches to the Electronics Library+ editor. Then optionally navigates to a symbol and verifies via export.++ This is orchestrated at bridge level to avoid add-in module caching+ issues and to allow multi-step operations with waits.+ """+ import tempfile+ import time++ file_path = args.get("filePath", "")+ symbol_name = args.get("symbolName", "")+ verify = args.get("verify", False)++ if not file_path:+ return {"success": False, "error": "No filePath specified"}++ file_path = file_path.replace("\\", "/")+ results = []++ # Step 1: Open the .lbr via Document.newDesignFromLocal+ open_result = _proxy_to_addin("execute_text_command", {+ "command": f"Document.newDesignFromLocal {file_path}",+ }, timeout=30)++ if not open_result.get("success"):+ return {+ "success": False,+ "error": f"Document.newDesignFromLocal failed: {open_result.get('error', 'unknown')}",+ "data": {"filePath": file_path},+ }+ results.append("Document.newDesignFromLocal: ok")++ # Step 2: Navigate to symbol if requested+ if symbol_name:+ time.sleep(3) # Let Fusion finish opening and switching workspace++ sym_ref = symbol_name if symbol_name.endswith(".sym") else f"{symbol_name}.sym"+ edit_result = _proxy_to_addin("electron_run", {+ "command": f"EDIT {sym_ref}",+ }, timeout=10)+ results.append(f"EDIT {sym_ref}: success={edit_result.get('success')}")++ # Zoom to fit+ _proxy_to_addin("electron_run", {"command": "WINDOW FIT"}, timeout=5)+ results.append("WINDOW FIT: ok")++ # Step 3: Verify via EXPORT SCRIPT+ verification = None+ if verify:+ import uuid+ time.sleep(2) # Let Fusion settle before export+ # Use a unique path to avoid "overwrite?" dialogs blocking the UI thread+ export_path = str(Path(tempfile.gettempdir()) / f"_adom_verify_{uuid.uuid4().hex[:8]}.scr")+ export_result = _proxy_to_addin("export_lbr", {"outputPath": export_path}, timeout=30)+ if export_result.get("success"):+ preview = export_result.get("data", {}).get("preview", "")+ has_symbol = False+ if symbol_name:+ has_symbol = (+ f"'{symbol_name.upper()}.sym'" in preview+ or f"'{symbol_name}.sym'" in preview+ )+ verification = {+ "exported": True,+ "fileSize": export_result.get("data", {}).get("fileSize", 0),+ "hasSymbol": has_symbol,+ "preview": preview[:500],+ }+ else:+ verification = {"exported": False, "error": export_result.get("error", "unknown")}+ results.append(f"Verification: exported={verification.get('exported', False)}")++ import os+ response = {+ "success": True,+ "output": f"Opened library: {os.path.basename(file_path)}",+ "data": {"results": results, "filePath": file_path},+ }+ if symbol_name:+ response["data"]["symbolName"] = symbol_name+ if verification:+ response["data"]["verification"] = verification++ # Auto-screenshot after opening — catches blocking dialogs, same as other open commands+ return _post_open_screenshot(response)+++def _orchestrate_save_lbr(args: dict) -> dict:+ """Save the currently open Electronics library as a .flbr file.++ Uses Document.CopyToDesktop to export the library in Fusion's native+ .flbr format. The file can be re-opened later or uploaded to the cloud.+ """+ import os++ output_path = args.get("outputPath", "")+ if not output_path:+ return {"success": False, "error": "No outputPath specified"}++ output_path = output_path.replace("\\", "/")++ # Ensure path ends with .flbr+ if not output_path.lower().endswith(".flbr"):+ output_path += ".flbr"++ # Ensure parent directory exists+ parent = os.path.dirname(output_path)+ if parent and not os.path.exists(parent):+ try:+ os.makedirs(parent, exist_ok=True)+ except Exception as e:+ return {"success": False, "error": f"Cannot create directory {parent}: {e}"}++ # Execute Document.CopyToDesktop via the add-in+ result = _proxy_to_addin("execute_text_command", {+ "command": f"Document.CopyToDesktop {output_path}",+ }, timeout=30)++ if not result.get("success"):+ return {+ "success": False,+ "error": f"Document.CopyToDesktop failed: {result.get('error', 'unknown')}",+ "data": {"outputPath": output_path},+ }++ # Verify the file was created+ if os.path.exists(output_path):+ file_size = os.path.getsize(output_path)+ result = {+ "success": True,+ "output": f"Saved library to {os.path.basename(output_path)} ({file_size} bytes)",+ "data": {"outputPath": output_path, "fileSize": file_size},+ }+ else:+ result = {+ "success": True,+ "output": f"Document.CopyToDesktop executed (file may still be writing)",+ "data": {"outputPath": output_path},+ }++ # Auto-screenshot after save — catches blocking dialogs+ return _post_open_screenshot(result, settle_time=1.0)+++def dispatch_command(command: str, args: dict) -> dict:+ """Dispatch a command to the appropriate handler."""+ # Direct handlers (don't need the add-in)+ handler = COMMAND_HANDLERS.get(command)+ if handler is not None:+ return handler(fusion_info, args)++ # Check if Fusion is installed and running before any add-in-dependent command.+ # We do NOT auto-launch — that causes 60s+ hangs when Fusion isn't running.+ # The user should launch Fusion themselves; we just report the status.+ # Commands that need Fusion running (add-in commands + orchestrated commands)+ _OPEN_WITH_SCREENSHOT = {"open_schematic", "open_board", "show_3d_board", "show_2d_board", "import_electronics"}+ if command in ADDIN_COMMANDS or command in ("open_lbr", "save_lbr") or command in _OPEN_WITH_SCREENSHOT:+ if not fusion_info.get("installed"):+ return {+ "success": False,+ "error": "Fusion 360 is not installed on this machine.",+ "errorCode": "fusion_not_installed",+ "_hint": "Fusion 360 cannot be installed programmatically. Report this error to the user and ask them to install Fusion from https://www.autodesk.com/products/fusion-360 (free for personal/hobby use), then retry.",+ }+ if not _is_fusion_running():+ return {+ "success": False,+ "error": "Fusion 360 is installed but not running.",+ "errorCode": "fusion_not_running",+ "_hint": "Call fusion_start to start Fusion 360, then wait for the add-in to become ready (fusion_start blocks until ready). Then retry this command.",+ }+ # Fusion is running — check if add-in is responsive (short timeout).+ # But first, if the add-in is busy with a long command, skip the+ # responsiveness check and fall through to the busy gate below.+ addin_busy = _check_addin_status(timeout=3.0)+ if addin_busy and addin_busy.get("busy"):+ pass # Fall through to busy gate+ elif not wait_for_addin(timeout=5):+ return {+ "success": False,+ "error": "Fusion 360 is running but AdomBridge add-in not responding.",+ "errorCode": "fusion_addin_not_responding",+ "_hint": "The add-in may be disabled or crashed. Try fusion_start which will restart Fusion cleanly. If that fails, the user must manually enable it via UTILITIES > ADD-INS > AdomBridge > Run on Startup + Run.",+ }++ # ── Busy gate: reject add-in commands immediately if a long command is running ──+ # Bridge-level commands (COMMAND_HANDLERS) already returned above — they use+ # Win32 APIs and don't touch the add-in, so they always work during a walk.+ # But add-in commands (and orchestrated commands that proxy to the add-in)+ # would pile up behind _main_thread_lock for 300s and crash the host.+ busy = _get_long_command()+ if busy and command not in LONG_RUNNING_COMMANDS:+ # This command would block behind the long-running one — reject immediately+ progress = _get_busy_progress()+ elapsed = round(_time.time() - busy["startedAt"], 1)+ progress_pct = None+ if progress and progress.get("foldersVisited") and progress.get("queueSize") is not None:+ total = progress["foldersVisited"] + progress["queueSize"]+ if total > 0:+ progress_pct = round(100 * progress["foldersVisited"] / total)+ return {+ "success": False,+ "error": (+ f"Fusion main thread busy — {busy['command']} has been running for {elapsed}s. "+ f"Your command '{command}' cannot execute until it finishes."+ ),+ "errorCode": "main_thread_busy",+ "busyCommand": busy["command"],+ "elapsedSeconds": elapsed,+ "progress": progress,+ "_hint": (+ f"A cloud search ({busy['command']}) is in progress"+ + (f" (~{progress_pct}% done)" if progress_pct is not None else "")+ + f", running for {elapsed}s. "+ "Do NOT retry add-in commands (get_app_state, document_info, etc.) — they will "+ "all be rejected until the search finishes. Commands that still work right now: "+ "fusion_window_info, fusion_screenshot_fusion, fusion_click_fusion, "+ "fusion_send_key, fusion_close_window. Wait for the search to complete, then retry."+ ),+ }++ # Cross-bridge case: this bridge's _long_command is None but the add-in+ # might be busy from another bridge/session. Quick non-blocking check.+ if not busy and command not in LONG_RUNNING_COMMANDS:+ addin_status = _check_addin_status(timeout=3.0)+ if addin_status and addin_status.get("busy"):+ elapsed = addin_status.get("elapsedSeconds", 0)+ walk = addin_status.get("walkProgress")+ busy_cmd = addin_status.get("busyCommand", "unknown")+ resp = {+ "success": False,+ "error": (+ f"Fusion main thread busy — {busy_cmd} running for {elapsed}s "+ f"(from another bridge/session). Your command '{command}' cannot "+ f"execute until it finishes."+ ),+ "errorCode": "main_thread_busy",+ "busyCommand": busy_cmd,+ "elapsedSeconds": elapsed,+ "_hint": (+ f"A long-running command ({busy_cmd}) is in progress from another session. "+ "Do NOT retry add-in commands — they will all be rejected until it finishes. "+ "Do NOT press Escape — the add-in is working, not stuck on a dialog. "+ "Commands that still work: fusion_window_info, fusion_screenshot_fusion, "+ "fusion_click_fusion, fusion_send_key, fusion_close_window."+ ),+ }+ if walk:+ resp["progress"] = walk+ return resp++ # Orchestrated multi-step commands (handled at bridge level)+ if command in _OPEN_WITH_SCREENSHOT:+ return _handle_open_with_screenshot(fusion_info, args, command)+ if command == "open_lbr":+ return _orchestrate_open_lbr(args)+ if command == "save_lbr":+ return _orchestrate_save_lbr(args)++ # Add-in proxy commands+ if command in ADDIN_COMMANDS:+ addin_cmd = ADDIN_COMMAND_MAP.get(command, command)+ proxy_timeout = ADDIN_COMMAND_TIMEOUTS.get(command, 30)+ # Wrap long-running commands in set/clear so the gate knows they're active.+ # Pre-dismiss blocking dialogs: modal dialogs steal Fusion's event loop,+ # preventing fireCustomEvent from being processed. Without this, the+ # walk appears "busy" but never actually starts — _walk_progress stays+ # None indefinitely while the command sits in the event queue.+ if command in LONG_RUNNING_COMMANDS:+ try:+ info = get_fusion_window_info()+ if info.get("dialogs"):+ for d in info["dialogs"]:+ try:+ # Use WM_CLOSE via PostMessage — doesn't steal foreground+ # (unlike send_key which uses SendInput + SetForegroundWindow).+ # WM_CLOSE is also more reliable than Escape for Qt dialogs.+ close_window(d.get("hwnd"))+ except Exception:+ pass+ import time as _time+ _time.sleep(0.5) # give Fusion a moment to process the dismiss+ except Exception:+ pass+ _set_long_command(command)+ try:+ return _proxy_to_addin(addin_cmd, args, timeout=proxy_timeout)+ finally:+ _clear_long_command()+ return _proxy_to_addin(addin_cmd, args, timeout=proxy_timeout)++ # Unknown command — check installation/running status for helpful errors+ if not fusion_info.get("installed"):+ return {+ "success": False,+ "error": f"Fusion 360 is not installed on this machine. (command: {command})",+ "errorCode": "fusion_not_installed",+ "_hint": "Fusion 360 cannot be installed programmatically. Report this to the user and ask them to install it from https://www.autodesk.com/products/fusion-360, then retry.",+ }+ if not _is_fusion_running():+ return {+ "success": False,+ "error": f"Fusion 360 is installed but not running. (command: {command})",+ "errorCode": "fusion_not_running",+ "_hint": "Call fusion_start to start Fusion 360 and wait for the add-in to be ready, then retry this command.",+ }+ return {+ "success": False,+ "error": f"Unknown command: {command}",+ "_hint": "Run `adom-desktop help` or check cli/src/commands.rs to see the list of available fusion_* commands. This command name may be misspelled or not yet implemented.",+ }+++def _build_status() -> dict:+ """Build the /status payload — the endpoint AD declares as healthEndpoint.++ MUST return HTTP 2xx whenever the server is up: AD's health check polls this+ path and a 404 (the classic manifest-vs-server mismatch) makes AD wait the+ full startup grace then report a generic "not reachable". We also self-report+ the GUI chip fields {led, summary, tooltip} — AD renders them verbatim (the+ bridge owns its color; AD owns only the unreachable→gray state).+ """+ import platform++ info = fusion_info or {}+ installed = bool(info.get("installed"))+ running = _is_fusion_running() if installed else False+ addin = _probe_addin() if running else None+ addin_ok = bool(addin)++ if platform.system() != "Windows":+ led, summary = "yellow", "Unsupported OS"+ tooltip = ("Bridge running, but this host is not Windows. Fusion 360 verbs "+ "need a Windows host with Fusion 360 installed.")+ elif not installed:+ led, summary = "yellow", "Fusion not installed"+ tooltip = ("Bridge running, but Fusion 360 is not installed. Install it from "+ "autodesk.com/products/fusion-360, then call fusion_start.")+ elif not running:+ led, summary = "yellow", "Fusion not running"+ tooltip = "Fusion 360 is installed but not running. Call fusion_start to launch it."+ elif not addin_ok:+ led, summary = "yellow", "Add-in not connected"+ tooltip = ("Fusion 360 is running but the AdomBridge add-in isn't responding yet "+ "(still loading, or needs enabling via UTILITIES > ADD-INS).")+ else:+ led, summary = "green", "Fusion ready"+ tooltip = "Fusion 360 running, AdomBridge add-in connected. All fusion_* verbs available."++ return {+ "status": "ok",+ "led": led,+ "summary": summary,+ "tooltip": tooltip,+ "bridgeVersion": BRIDGE_VERSION,+ "fusion": {**info, "running": running},+ "addin": addin,+ }+++class FusionBridgeHandler(BaseHTTPRequestHandler):+ """HTTP request handler for the Fusion 360 bridge server."""++ def do_GET(self):+ # /status is the manifest's healthEndpoint; /health is kept as a+ # back-compat alias (older callers + the add-in-probe code path). Both+ # return the same 2xx payload — new chip fields are purely additive.+ if self.path in ("/status", "/health"):+ self._respond(200, _build_status())+ else:+ self._respond(404, {"error": "Not found"})++ def do_POST(self):+ if self.path != "/command":+ self._respond(404, {"error": "Not found"})+ return++ content_length = int(self.headers.get("Content-Length", 0))+ body = self.rfile.read(content_length)++ try:+ request = json.loads(body)+ except json.JSONDecodeError as e:+ self._respond(400, {"success": False, "error": f"Invalid JSON: {e}"})+ return++ command = request.get("command", "")+ args = request.get("args", {})++ print(f"[Fusion Bridge] Command: {command} | Args: {json.dumps(args)}")++ try:+ result = dispatch_command(command, args)+ self._respond(200, result)+ except Exception as e:+ print(f"[Fusion Bridge] ERROR: {e}")+ traceback.print_exc()+ self._respond(500, {+ "success": False,+ "error": f"Internal error: {e}",+ })++ def _respond(self, status: int, data: dict):+ self.send_response(status)+ self.send_header("Content-Type", "application/json")+ self.end_headers()+ self.wfile.write(json.dumps(data).encode("utf-8"))++ def log_message(self, format, *args):+ print(f"[Fusion Bridge] {args[0]} {args[1]} {args[2]}")+++def main():+ global fusion_info++ # Line-buffer stdout/stderr. AD spawns the bridge console-less and captures+ # our stdout+stderr into ~/.adom/bridge-logs/fusion360.log. Python BLOCK-buffers+ # stdout when it's not a TTY, so without this the buffer never flushes while+ # serve_forever() runs → the log stays 0 bytes (and a spawn-crash traceback+ # would be lost). Line buffering flushes every print()/traceback on newline.+ try:+ sys.stdout.reconfigure(line_buffering=True)+ sys.stderr.reconfigure(line_buffering=True)+ except Exception:+ pass++ port = DEFAULT_PORT+ if "--port" in sys.argv:+ idx = sys.argv.index("--port")+ if idx + 1 < len(sys.argv):+ port = int(sys.argv[idx + 1])++ # Early banner BEFORE detection — so the log proves we started even if+ # detect_fusion() is slow or wedges. (KiCad bridge prints an equivalent line.)+ print(f"[Fusion Bridge] starting - version {BRIDGE_VERSION}, port {port}, "+ f"healthEndpoint /status, pid {os.getpid()}", flush=True)++ fusion_info = detect_fusion()++ print(f"[Fusion Bridge] Fusion 360 detection result:")+ print(f" Installed: {fusion_info.get('installed', False)}")+ if fusion_info.get("installed"):+ print(f" Exe path: {fusion_info.get('exe_path')}")+ print(f" AddIns: {fusion_info.get('addins_dir')}")+ print(f" Add-in: {'installed' if fusion_info.get('addin_installed') else 'not installed'}")+ print(f" Running: {fusion_info.get('running')}")++ # Auto-install the add-in if Fusion is installed but the add-in isn't+ if not fusion_info.get("addin_installed"):+ print(f"[Fusion Bridge] Auto-installing AdomBridge add-in...")+ try:+ install_addin()+ fusion_info = detect_fusion() # Re-detect after install+ print(f" Add-in: {'installed' if fusion_info.get('addin_installed') else 'FAILED'}")+ except Exception as e:+ print(f" Add-in install failed: {e}")+ else:+ print(f" WARNING: Fusion 360 not found. Some commands will fail.")++ addin_health = _probe_addin()+ if addin_health:+ print(f" Add-in server: running on port {ADDIN_PORT}")+ else:+ print(f" Add-in server: not running (port {ADDIN_PORT})")++ all_commands = list(COMMAND_HANDLERS.keys()) + sorted(ADDIN_COMMANDS)+ print(f"[Fusion Bridge] Available commands: {', '.join(all_commands)}")++ server = ThreadingHTTPServer(("127.0.0.1", port), FusionBridgeHandler)+ server.daemon_threads = True+ print(f"[Fusion Bridge] Listening on http://127.0.0.1:{port}")+ print(f"[Fusion Bridge] Health check: http://127.0.0.1:{port}/status (alias /health)")+ print(f"[Fusion Bridge] Press Ctrl+C to stop.")++ try:+ server.serve_forever()+ except KeyboardInterrupt:+ print("\n[Fusion Bridge] Shutting down.")+ server.server_close()+++if __name__ == "__main__":+ # Surface any fatal startup error to stderr (which AD captures into+ # ~/.adom/bridge-logs/fusion360.log) so a spawn-crash is debuggable, then+ # re-raise for a non-zero exit.+ try:+ main()+ except Exception:+ print("[Fusion Bridge] FATAL: bridge failed to start", flush=True)+ traceback.print_exc()+ sys.stderr.flush()+ raise