Fusion - the Fusion 360 Bridge

@@ -1376,15 +1376,21 @@ def _handle_fusion_readiness(fusion_info: dict, args: dict) -> dict:         # ONCE. They cannot share a token (Autodesk DPAPI-encrypts Fusion's store), but they CAN         # share the warm browser SSO session - so APS consent is silent right after a Fusion login.         _aps = _aps_quick_state()-        _updmsg = ""+        # MCP state, reported beside APS on every launch (both are search/drive surfaces the+        # AI should know about without probing). Cheap: a 0.4s loopback socket.+        _mcpmsg = (" Autodesk's MCP server is LIVE (fusion_mcp_* verbs work)."+                   if _probe_mcp_port() else+                   " Autodesk's MCP server is OFF - fusion_mcp_enable can turn it on for the "+                   "user (~10s, drives the Preferences dialog).")+        _updmsg = _mcpmsg         if _upd.get("applied"):-            _updmsg = (" AUTO-UPDATE: Fusion was offering an update and I clicked %r for the user "+            _updmsg += (" AUTO-UPDATE: Fusion was offering an update and I clicked %r for the user "                        "IN THE BACKGROUND - it downloads now and the new build comes up on the next "                        "launch. TELL THE USER it is handled; they never need to click the nag. "                        "They can opt out with fusion_set_auto_update {enabled:false} (sticky)."                        % _upd.get("button"))         elif not _auto_update_enabled():-            _updmsg = (" AUTO-UPDATE is OFF (the user opted out, sticky) - do NOT click Fusion's "+            _updmsg += (" AUTO-UPDATE is OFF (the user opted out, sticky) - do NOT click Fusion's "                        "update prompts for them.")         if _aps.get("signedIn"):             hint = ("READY - drive Fusion (fusion_open_lbr / build_library_3d / etc.). APS cloud "@@ -1636,6 +1642,88 @@ _APS_ABE = ( )  +# ── SEARCH ROUTER: one live-state answer to "which file-search should the AI use?" ───────────+# John, 2026-07-24: "you now have a lot of ways to search for files, so even the ai will get+# confused by which verb to use... are your hints smart, i.e. they know if aps is configured or+# not so the hint is dynamic and does a bunch of lookup work... and give hints where you can+# tell it ideas on things you could enable and the complexity of enabling it?"+#+# There are three ways to find a cloud file, and the right one DEPENDS ON LIVE STATE:+#   1. fusion_aps_search        - APS Data Management. Needs APS signed in. Works with Fusion+#                                 CLOSED, headless, no subscription. ~2s.+#   2. fusion_mcp_call read/document/search - Autodesk's MCP server. Needs Fusion RUNNING ++#                                 subscription + the MCP toggle on. ~2s.+#   3. fusion_search_cloud_files - DISABLED (30+ min walk, crashed Fusion). Never.+#+# _search_router() probes all of it live (cheap: APS token file read + a 0.4s loopback socket)+# and every relevant hint renders the SAME ranked answer, including what could be ENABLED and+# what enabling costs (APS signin = silent on a warm SSO session, else a browser consent;+# MCP = fusion_mcp_enable drives the Preferences dialog, ~10s, brief announced foreground).+def _probe_mcp_port(timeout: float = 0.4) -> bool:+    """Is Autodesk's MCP server listening on loopback right now? The bridge runs ON the box, so+    this is a direct, cheap socket probe - no relay round-trip."""+    import socket+    try:+        s = socket.socket()+        s.settimeout(timeout)+        s.connect(("127.0.0.1", 27182))+        s.close()+        return True+    except Exception:+        return False+++def _search_router() -> dict:+    """Live state of every file-search path + a ranked recommendation. Never raises."""+    aps_st = _aps_quick_state()                      # {configured, signedIn} - token-aware+    mcp_on = _probe_mcp_port()+    fusion_up = mcp_on  # the MCP server lives inside Fusion; port open implies running+    if not mcp_on:+        try:+            fusion_up = _is_fusion_running()+        except Exception:+            fusion_up = False++    aps_ready = bool(aps_st.get("signedIn"))+    options = []+    if aps_ready:+        options.append(("fusion_aps_search", "ready NOW (~2s, works even with Fusion closed)"))+    if mcp_on:+        options.append(("fusion_mcp_call {tool:'fusion_mcp_read', arguments:{queryType:'document',"+                        "operation:'search', name:'...'}}", "ready NOW (~2s, MCP server is live)"))++    enable = []+    if not aps_ready:+        if aps_st.get("configured"):+            enable.append("APS: one fusion_aps_signin - SILENT if run right after a Fusion "+                          "sign-in (warm browser SSO), else a one-click browser consent")+        else:+            enable.append("APS: needs one-time setup - fusion_aps_set_client_id (once per "+                          "company) then fusion_aps_signin (once per user)")+    if not mcp_on:+        if fusion_up:+            enable.append("MCP: fusion_mcp_enable turns it on FOR the user (drives the "+                          "Preferences dialog, ~10s, brief announced foreground); needs a "+                          "Fusion subscription")+        else:+            enable.append("MCP: needs Fusion running first (fusion_start), then "+                          "fusion_mcp_enable; needs a Fusion subscription")++    if options:+        best = options[0][0].split(" ")[0].split("{")[0]+        rec = "BEST SEARCH RIGHT NOW: " + "; also ".join("%s - %s" % o for o in options) + "."+    else:+        best = None+        rec = ("NO file search is ready right now. Quickest to enable: "+               + (enable[0] if enable else "none") + ".")+    if enable and options:+        rec += " Could also enable: " + " | ".join(enable) + "."++    return {"apsReady": aps_ready, "apsConfigured": bool(aps_st.get("configured")),+            "mcpLive": mcp_on, "fusionRunning": fusion_up,+            "bestSearch": best, "searchHint": rec}++ def _aps_state_hint(extra: str = "") -> dict:     """Live APS readiness + a self-teaching hint bundle for the AI. Never raises."""     try:@@ -1655,12 +1743,16 @@ def _aps_state_hint(extra: str = "") -> dict:         stage, todo = "token_expired", "Token expired/refresh failed — fusion_aps_signin again."     else:         stage, todo = "ready", "Ready — fusion_aps_search {\"query\":\"...\"} or fusion_aps_open {\"name\":\"...\"}."-    parts = [f"APS {stage}.", todo, "VERBS: " + _APS_VERB_MAP, _APS_ABE, _APS_WHY]+    # SMART ROUTING (John 2026-07-24): an APS problem must not dead-end the AI when another+    # search path is live right now. The router probes MCP + Fusion state and says what is+    # usable NOW vs what could be enabled and at what cost.+    router = _search_router()+    parts = [f"APS {stage}.", todo, router["searchHint"], "VERBS: " + _APS_VERB_MAP, _APS_ABE, _APS_WHY]     if extra:         parts.insert(0, extra)     return {         "apsStage": stage, "apsConfigured": configured, "apsSignedIn": signed_in,-        "apsTokenLive": live, "_hint": "  ".join(parts),+        "apsTokenLive": live, "searchRouter": router, "_hint": "  ".join(parts),     }  @@ -3923,7 +4015,7 @@ def _mcp_rpc(method: str, params: dict = None, timeout: int = 60) -> dict:         return {"_mcpError": "unparseable response", "raw": (r.get("body") or "")[:400]}  -_MCP_OFF_HINT = (+_MCP_OFF_STATIC = (     "Fusion's MCP server is not reachable on 127.0.0.1:27182. It is OFF by default. The bridge "     "can turn it on FOR the user: call fusion_mcp_enable, which opens Preferences, expands "     "General, selects API, ticks 'Fusion MCP Server (runs locally on this device)' and clicks "@@ -3934,12 +4026,27 @@ _MCP_OFF_HINT = ( )  +def _mcp_off_hint() -> str:+    """The MCP-unreachable hint, made state-aware: never tell the AI to go enable MCP when APS+    is signed in and could serve the request RIGHT NOW with zero setup."""+    r = _search_router()+    lead = ""+    if r.get("apsReady"):+        lead = ("NOTE FIRST: APS is signed in, so for FILE SEARCH you do not need MCP at all - "+                "fusion_aps_search works right now. Only enable MCP if you need its other surface "+                "(script text-to-CAD, screenshots, electronics object-model reads). ")+    elif r.get("apsConfigured"):+        lead = ("NOTE: for FILE SEARCH, APS is configured and only needs fusion_aps_signin "+                "(silent on a warm SSO session) - that may be less disruptive than enabling MCP. ")+    return lead + _MCP_OFF_STATIC++ def _handle_mcp_status(fusion_info: dict, args: dict) -> dict:     """Is Autodesk's Fusion MCP server up, and what does it expose?"""     s = _mcp_new_session()     if s.get("error"):         return {"success": True, "enabled": False, "url": _MCP_URL, "error": s.get("error"),-                "_hint": _MCP_OFF_HINT, "statusVerb": "fusion_mcp_status"}+                "_hint": _mcp_off_hint(), "statusVerb": "fusion_mcp_status"}     tools = []     tl = _mcp_rpc("tools/list")     if not tl.get("_mcpError"):@@ -3962,7 +4069,7 @@ def _handle_mcp_tools(fusion_info: dict, args: dict) -> dict:     """Full tool list WITH input schemas."""     tl = _mcp_rpc("tools/list")     if tl.get("_mcpError"):-        return {"success": False, "enabled": False, "error": tl["_mcpError"], "_hint": _MCP_OFF_HINT}+        return {"success": False, "enabled": False, "error": tl["_mcpError"], "_hint": _mcp_off_hint()}     tools = (tl.get("result") or {}).get("tools") or []     return {"success": True, "count": len(tools), "tools": tools,             "_hint": "Call one with fusion_mcp_call {\"tool\":\"<name>\",\"arguments\":{...}}.",@@ -3978,7 +4085,7 @@ def _handle_mcp_call(fusion_info: dict, args: dict) -> dict:     r = _mcp_rpc("tools/call", {"name": tool, "arguments": (args or {}).get("arguments") or {}},                  timeout=int((args or {}).get("timeout") or 120))     if r.get("_mcpError"):-        return {"success": False, "error": r["_mcpError"], "_hint": _MCP_OFF_HINT}+        return {"success": False, "error": r["_mcpError"], "_hint": _mcp_off_hint()}     res = r.get("result") or {}     return {"success": not res.get("isError", False), "tool": tool, "result": res,             "_hint": ("Autodesk MCP tool result. Content is usually a list of {type,text} blocks. "@@ -3993,12 +4100,12 @@ def _handle_mcp_resources(fusion_info: dict, args: dict) -> dict:     if uri:         r = _mcp_rpc("resources/read", {"uri": uri})         if r.get("_mcpError"):-            return {"success": False, "error": r["_mcpError"], "_hint": _MCP_OFF_HINT}+            return {"success": False, "error": r["_mcpError"], "_hint": _mcp_off_hint()}         return {"success": True, "uri": uri, "result": r.get("result"),                 "statusVerb": "fusion_mcp_status"}     r = _mcp_rpc("resources/list")     if r.get("_mcpError"):-        return {"success": False, "error": r["_mcpError"], "_hint": _MCP_OFF_HINT}+        return {"success": False, "error": r["_mcpError"], "_hint": _mcp_off_hint()}     res = (r.get("result") or {}).get("resources") or []     return {"success": True, "count": len(res),             "resources": [x.get("uri") for x in res],@@ -4033,7 +4140,7 @@ def _handle_mcp_enable(fusion_info: dict, args: dict) -> dict:             dlg = w.get("hwnd")     if not dlg:         return {"success": False, "error": "Preferences dialog did not open.",-                "steps": steps, "_hint": _MCP_OFF_HINT}+                "steps": steps, "_hint": _mcp_off_hint()}     steps.append("opened Preferences")      def shot():