fusion - the Fusion 360 bridge (macOS)

"""Auto-detect the Fusion 360 installation on macOS.

This bridge is macOS-only (Kyle, 2026-07-31): the Windows detection —
Fusion360.exe / FusionLauncher.exe markers, tasklist walks, Win32 focus
guards — is gone. The hard-won webdeploy lessons (John, 2026-07-14) carry
over with mac markers:

  * Fusion streams into ~/Library/Application Support/Autodesk/webdeploy/
    production/<hash>/, one hash dir per content build; old builds are never
    pruned, so several stale dirs are NORMAL.
  * "Installed" is gated on the real app binary
    (<hash>/Autodesk Fusion.app/Contents/MacOS/Autodesk Fusion, ~1.6 MB
    measured) existing at full size — the bundle dir itself appears early
    in a stream and must never be the completeness marker.
"""
from __future__ import annotations  # PEP-604 hints must parse on py3.9 (mac stock)

import json
import os
import subprocess
import time
import urllib.error
import urllib.request
from pathlib import Path

from handlers import mac_ui

ADDIN_PORT = 8774

# Fusion's webdeploy root on macOS. Kept as a list for parity with callers
# that iterate (and in case Autodesk ever adds a second root).
_WEBDEPLOY_BASES = [
    Path.home() / "Library" / "Application Support" / "Autodesk" / "webdeploy" / "production",
]
WEBDEPLOY_BASE = _WEBDEPLOY_BASES[0]

# EVERY known per-user add-in dir, newest convention first. Fusion MOVED this
# across versions and silently ignores the others (issue #63: an add-in in the
# legacy API/AddIns dir never loads on 2025+ Fusion, which scans FusionAddins -
# proven live on a fresh install). Track ALL of them; never assume one path.
ADDINS_DIR_CANDIDATES = [
    Path.home() / "Library" / "Application Support" / "Autodesk" / "FusionAddins",
    Path.home() / "Library" / "Application Support" / "Autodesk" / "Autodesk Fusion" / "API" / "AddIns",
    Path.home() / "Library" / "Application Support" / "Autodesk" / "Autodesk Fusion 360" / "API" / "AddIns",
]


def detect_fusion() -> dict:
    """Detect the Fusion 360 installation. Returns a dict with paths and status."""
    app_path = _find_fusion_app()
    addins_dir = _find_addins_dir()
    addin_installed = _check_addin_installed(addins_dir) if addins_dir else False
    running = _is_fusion_running()

    if app_path:
        return {
            "installed": True,
            "exe_path": str(app_path),
            "addins_dir": str(addins_dir) if addins_dir else None,
            "addin_installed": addin_installed,
            "running": running,
        }

    return {
        "installed": False,
        "exe_path": None,
        "addins_dir": str(addins_dir) if addins_dir else None,
        "addin_installed": addin_installed,
        "running": running,
        "error": "Fusion 360 not found",
    }


# ---------------------------------------------------------------------------------------------------
# INSTALL-COMPLETENESS DETECTION
#
# The reliable signal is the real app binary inside the bundle. It only exists at full size once the
# payload finished streaming. Gate "installed" on it, NEVER on the bundle dir appearing (that lands
# early, mid-stream — the mac analog of the FusionLauncher.exe trap that broke the fresh VM,
# John 2026-07-14).
# ---------------------------------------------------------------------------------------------------
_FUSION_BUNDLE = "Autodesk Fusion.app"
_FUSION_APP_EXE = os.path.join(_FUSION_BUNDLE, "Contents", "MacOS", "Autodesk Fusion")
_MIN_APP_EXE_BYTES = 200_000  # guards against a placeholder the streamer writes early


def _app_dir_complete(subdir: Path) -> bool:
    """A production hash dir is a COMPLETE Fusion install ONLY when it contains the bundle's real
    binary at full size. Process-independent and version-stable."""
    exe = subdir / _FUSION_APP_EXE
    try:
        return exe.exists() and exe.stat().st_size >= _MIN_APP_EXE_BYTES
    except OSError:
        return False


def _any_app_complete() -> bool:
    """True if ANY webdeploy hash dir holds a complete Fusion. This is the single source of truth
    for 'Fusion is installed'."""
    for base in _WEBDEPLOY_BASES:
        try:
            if base.exists():
                for subdir in base.iterdir():
                    if subdir.is_dir() and _app_dir_complete(subdir):
                        return True
        except OSError:
            continue
    return False


def _find_fusion_app() -> Path | None:
    """The newest COMPLETE 'Autodesk Fusion.app' bundle to launch (via `open`), or None.

    Completeness is gated per-dir on the bundle binary — returning a bundle whose
    binary hasn't finished streaming reproduces the launch-into-half-written-install
    failure."""
    candidates = []
    for base in _WEBDEPLOY_BASES:
        if not base.exists():
            continue
        try:
            subdirs = [d for d in base.iterdir() if d.is_dir()]
        except OSError:
            continue
        for subdir in subdirs:
            # The bundle usually sits inside the hash dir; tolerate it directly
            # under production/ as well.
            if subdir.name == _FUSION_BUNDLE and (subdir / "Contents").exists():
                app = subdir
            else:
                app = subdir / _FUSION_BUNDLE
            binary = app / "Contents" / "MacOS" / "Autodesk Fusion"
            try:
                if not (binary.exists() and binary.stat().st_size >= _MIN_APP_EXE_BYTES):
                    continue
                candidates.append((subdir.stat().st_mtime, app))
            except OSError:
                continue
    if not candidates:
        return None
    candidates.sort(key=lambda c: c[0], reverse=True)
    return candidates[0][1]


def _incomplete_webdeploy_present() -> bool:
    """Disk-state (process-INDEPENDENT) 'a fresh install is actively streaming' signal: some hash dir
    has the bundle but NO complete binary exists ANYWHERE yet. Why disk-state and not a process
    check: the Autodesk streamer spawns short-lived per-chunk workers, so a process poll reads False
    for most of a live multi-GB stream. Why gate on 'no complete app anywhere': once the app IS
    complete, leftover partial dirs are NORMAL and must NOT read as 'still installing'. Never
    raises."""
    if _any_app_complete():
        return False                                  # a complete install exists -> not installing
    for base in _WEBDEPLOY_BASES:
        try:
            if not base.exists():
                continue
            for subdir in base.iterdir():
                if not subdir.is_dir():
                    continue
                if (subdir / _FUSION_BUNDLE).exists() or subdir.name == _FUSION_BUNDLE:
                    return True                       # payload started, no complete app -> streaming
        except OSError:
            continue
    return False


def _clean_incomplete_webdeploy() -> list:
    """Remove the residue of an INTERRUPTED stream so the next install re-streams clean.
    Only ever called AFTER a launch failure (we already know something is corrupt).

    SAFETY GUARD: if a COMPLETE install exists anywhere, do NOTHING and return [] — partial-looking
    dirs beside a complete app can be a legit part of the install, not corruption. Never raises."""
    import shutil
    if _any_app_complete():
        return []                                     # healthy install present - never touch it
    removed = []
    for base in _WEBDEPLOY_BASES:
        if not base.exists():
            continue
        try:
            subdirs = [d for d in base.iterdir() if d.is_dir()]
        except OSError:
            continue
        for subdir in subdirs:
            has_bundle = (subdir / _FUSION_BUNDLE).exists() or subdir.name == _FUSION_BUNDLE
            if not has_bundle:
                continue
            if not _app_dir_complete(subdir if subdir.name != _FUSION_BUNDLE else subdir.parent):
                try:
                    shutil.rmtree(subdir, ignore_errors=True)
                    removed.append(subdir.name)
                except Exception:
                    pass
    return removed


def fusion_update_in_progress() -> bool:
    """Best-effort: is Fusion ACTIVELY applying an auto-update right now? While it is, Fusion
    crash-restarts ITSELF every ~30-60s, so verbs intermittently see 'not running'. Callers surface
    a distinct non-fatal `updating` status instead of a fatal error so a driving AI waits/retries.

    A bare "2+ production dirs" count is NOT an update — Fusion never prunes old webdeploy builds,
    so a normal machine accumulates many stale dirs (regression caught live 2026-07-06). Require
    ACTIVE streaming: 2+ bundle-bearing builds AND at least one build dir TOUCHED in the last
    ~20 min (the streamer is writing into it). Never raises."""
    now = time.time()
    RECENT_SECS = 20 * 60
    bundle_mtimes = []
    for base in _WEBDEPLOY_BASES:
        try:
            if not base.exists():
                continue
            for d in base.iterdir():
                try:
                    if not d.is_dir():
                        continue
                    if (d / _FUSION_BUNDLE).exists() or d.name == _FUSION_BUNDLE:
                        bundle_mtimes.append(d.stat().st_mtime)
                except OSError:
                    continue
        except OSError:
            continue
    if len(bundle_mtimes) < 2:
        return False
    return any((now - mt) < RECENT_SECS for mt in bundle_mtimes)


def _find_addins_dir() -> Path | None:
    """Find the Fusion 360 add-ins directory."""
    for candidate in ADDINS_DIR_CANDIDATES:
        if candidate.exists():
            return candidate
    # Return the first candidate even if it doesn't exist yet (for install_addin)
    return ADDINS_DIR_CANDIDATES[0] if ADDINS_DIR_CANDIDATES else None


def _check_addin_installed(addins_dir: Path | None) -> bool:
    """Check if the AdomBridge add-in is installed in ANY known add-in dir.

    Scans every ADDINS_DIR_CANDIDATES entry (not just the preferred one) - the
    copy that counts is whichever dir THIS Fusion version scans, and we can't
    know that a priori, so presence anywhere counts as installed."""
    for candidate in ADDINS_DIR_CANDIDATES:
        if (candidate / "AdomBridge" / "AdomBridge.py").exists():
            return True
    if addins_dir and (Path(addins_dir) / "AdomBridge" / "AdomBridge.py").exists():
        return True
    return False


def _has_fusion_process() -> bool:
    """Is any Fusion 360 process running? (pgrep on the bundle binary path —
    process check is authoritative; window-title matching false-positives on
    browser tabs that mention 'Autodesk Fusion')."""
    return mac_ui.has_fusion_process()


def _is_fusion_running() -> bool:
    return _has_fusion_process()


def is_fusion_running() -> bool:
    """Public API: check if Fusion 360 is currently running (live check)."""
    return _has_fusion_process()


def family_windows() -> list:
    """Visible normal-layer windows owned by Fusion processes, in the shape the
    launch state machine consumes: {hwnd, title, pid, image, width, height,
    left, top, owner}. 'hwnd' is the CGWindowID; CGWindowList has no owner-popup
    concept, so 'owner' is always 0 and find_licensing_dialog relies on the
    shape gate. Never raises."""
    out = []
    try:
        for w in mac_ui._fusion_windows():
            r = w["rect"]
            if r["width"] <= 0 or r["height"] <= 0:
                continue
            out.append({
                "hwnd": int(w["id"]), "title": w["title"] or "", "pid": int(w["pid"]),
                "image": w.get("owner") or "",
                "width": r["width"], "height": r["height"],
                "left": r["left"], "top": r["top"],
                "owner": 0,
            })
    except Exception:
        pass
    return out


def find_licensing_dialog():
    """The seat/licensing or sign-in modal ("Active Sessions Exceeded" / "Suspend Remote Session" /
    a sign-in prompt), or None.

    CGWindowList exposes no owned-popup relationship, so detection is the SHAPE gate proven on the
    Windows line: a seat/sign-in modal is always WIDE (823x262 'Suspend Remote Session', 1235x527
    'Active Sessions Exceeded') — far wider than any docked side panel — and never the LARGEST
    Fusion window (a dialog is never bigger than the app that owns it; matching the main window is
    the false-fire that hung readiness, John 2026-07-14). Require width >= 480 (excludes narrow
    panels), 180 <= height <= 900 (excludes toolbar strips), and not the main window."""
    wins = family_windows()
    if not wins:
        return None
    main_hwnd = max(wins, key=lambda w: w["width"] * w["height"])["hwnd"]
    for w in wins:
        if w["hwnd"] == main_hwnd:
            continue
        ww, hh = w["width"], w["height"]
        if 480 <= ww <= 1500 and 180 <= hh <= 900:
            return w
    return None


def wait_for_addin(timeout: int = 90, dismiss_dialogs: bool = False) -> bool:
    """Wait for the AdomBridge add-in HTTP server to become responsive.

    Polls the add-in health endpoint once per second up to `timeout` seconds.
    Returns True if the add-in responded, False on timeout.

    If dismiss_dialogs is True, also checks for and dismisses blocking dialogs
    (startup picker, recovery dialog) every 5 seconds while waiting. This is
    critical at startup: dialogs block the add-in's main thread, so dismissing
    them is required for the add-in to become responsive.
    """
    for i in range(timeout):
        try:
            req = urllib.request.Request(
                f"http://127.0.0.1:{ADDIN_PORT}/health", method="GET"
            )
            with urllib.request.urlopen(req, timeout=1) as resp:
                return True
        except Exception:
            # Every 5 seconds during the wait, try to dismiss blocking dialogs.
            # Dialogs are the #1 reason the add-in can't respond at startup.
            if dismiss_dialogs and i >= 5 and i % 5 == 0:
                _dismiss_startup_dialogs()
            time.sleep(1)
    return False


def _dismiss_startup_dialogs():
    """Dismiss common Fusion startup dialogs (recovery, startup picker) so the
    add-in's main thread unblocks and it can become responsive."""
    try:
        from handlers.dismiss_recovery import dismiss_recovery_dialog
        dismiss_recovery_dialog(max_wait=2.0)
    except Exception:
        pass

    # Dismiss the startup picker ("What do you want to design?") and other
    # Fusion dialogs by sending Escape to the Fusion main window.
    try:
        from handlers.fusion_ui import send_key_to_fusion
        send_key_to_fusion("escape")
    except Exception:
        pass


def _popen_fusion_background(app_path: str):
    """Launch the Fusion bundle in the BACKGROUND: `open -g` starts it without
    stealing the user's focus (John, 2026-07-06: "new windows ALWAYS only open
    in the background"). The add-in still loads fully unfocused; a verb
    explicitly foregrounds Fusion later when the user should see it."""
    return subprocess.Popen(
        ["open", "-g", str(app_path)],
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )


def ensure_fusion_running(fusion_info: dict, wait_addin: bool = False) -> dict | None:
    """Launch Fusion 360 if not already running.

    Uses a live process check (not the stale fusion_info["running"] flag).
    Returns None on success, or an error dict on failure.
    If wait_addin is True, also waits for the add-in to be responsive.
    """
    # Live check — ignore stale fusion_info["running"]
    if _has_fusion_process():
        if wait_addin:
            # Fusion is running but add-in might not be ready yet
            if not wait_for_addin(timeout=10):
                return {
                    "success": False,
                    "error": "Fusion 360 is running but AdomBridge add-in not responding.",
                    "errorCode": "fusion_addin_not_responding",
                    "_hint": "Fix it YOURSELF - never ask the user: (1) run install_addin() so the "
                             "add-in lands in EVERY Fusion add-in dir (Fusion moved it to "
                             "FusionAddins on 2025+ builds; an add-in in the legacy API/AddIns dir "
                             "is silently ignored - issue #63), then (2) restart Fusion via "
                             "fusion_stop + fusion_start so it rescans.",
                }
        return None  # Already running (and add-in ready if requested)

    exe_path = fusion_info.get("exe_path")
    if not exe_path or not Path(exe_path).exists():
        return {"success": False, "error": "Fusion 360 executable not found"}

    try:
        _popen_fusion_background(exe_path)
    except Exception as e:
        return {"success": False, "error": f"Failed to launch Fusion 360: {e}"}

    # Wait for the Fusion process to appear (up to 30s)
    for _ in range(60):
        time.sleep(0.5)
        if _has_fusion_process():
            break
    else:
        # LAUNCH FAILED - do NOT return a blind "process not found". READ the dialog that may be on
        # screen, and DISTINGUISH the two causes (conflating them mis-blamed a healthy install,
        # John 2026-07-14):
        #  (a) app is COMPLETE but launch failed -> the install is FINE; do NOT re-stream.
        #  (b) app is NOT complete -> a genuinely half-streamed install; clean the partial dirs so
        #      the next install re-streams clean.
        launch_err = None
        try:
            from handlers.dialog_classify import classify_launch_dialogs, close_dialog_bg
            for d in classify_launch_dialogs():
                if d.get("category") == "launch_error" or "streamed application" in (d.get("title") or "").lower():
                    launch_err = d
                    close_dialog_bg(d.get("hwnd"))
        except Exception:
            pass

        if _any_app_complete():
            return {
                "success": False,
                "error": "Fusion is INSTALLED (complete bundle present) but did not start within 30s"
                         + (" - a launch-error dialog was dismissed" if launch_err else "") + ".",
                "errorCode": "fusion_launch_failed",
                "installed": True,
                "dialog": (launch_err or {}).get("title"),
                "_hint": "The install is COMPLETE - do NOT re-stream. Retry fusion_start (a stale "
                         "lock from a just-killed instance clears on a retry). If it keeps failing, "
                         "fusion_kill then wait a few seconds before fusion_start.",
            }
        removed = _clean_incomplete_webdeploy()
        return {
            "success": False,
            "error": "Fusion launch failed: the streamed install is INCOMPLETE (no complete bundle yet).",
            "errorCode": "fusion_incomplete_install",
            "cleanedWebdeployDirs": removed,
            "dialog": (launch_err or {}).get("title"),
            "_hint": "AUTO-RECOVERED: dismissed the launch-error dialog and removed the partial "
                     "webdeploy dir(s). Re-run fusion_install_fusion, then WAIT for fusion_readiness "
                     "to report installed:true BEFORE fusion_start - starting mid-stream is what "
                     "corrupts the install.",
        }

    # Give it time to finish initializing UI
    time.sleep(5.0)

    if wait_addin:
        # dismiss_dialogs=True: while waiting for the add-in, also check for
        # and dismiss blocking dialogs every 5s. This is critical because
        # startup dialogs (picker, recovery) block the add-in's main thread.
        if not wait_for_addin(timeout=60, dismiss_dialogs=True):
            return {
                "success": False,
                "error": "addin_timeout_after_launch",
                "message": "Fusion 360 launched but the AdomBridge add-in didn't respond "
                           "within 60s. A modal dialog (e.g. Document Recovery, update prompt) "
                           "may be blocking startup. Check Fusion and dismiss any dialogs, "
                           "then retry.",
            }

    return None


# Back-compat alias: callers that predate the mac-only rewrite import
# _find_fusion_launcher; on macOS the "launcher" IS the app bundle path.
_find_fusion_launcher = _find_fusion_app


if __name__ == "__main__":
    info = detect_fusion()
    print(json.dumps(info, indent=2))