Adom Desktop - KiCad Bridge

"""plugin_install.py — auto-install adom_bridge into KiCad USER_SITE.

The Phase 2 reverse bridge plugin (`adom_bridge.py`) needs to live at
KiCad's USER_SITE directory so that KiCad's bundled Python loads it via
`usercustomize.py` at every GUI process startup. See PLAN.md v4 for the
architectural story.

This module does the deployment, with extensive feedback for the AI that
triggered the install. The pattern is:
- On every bridge boot, the first incoming kicad_* command triggers
  `ensure_plugin_installed_once()`.
- The result (whether install was needed, what was copied, what failed)
  is cached in module state and attached as `pluginAutoInstall` on that
  first response.
- Subsequent commands do nothing (cached short-circuit) so steady-state
  command flow has zero overhead.

Failure modes the AI needs to know about
========================================
* `appdata_missing` — %USERPROFILE% (or $HOME on Mac) unresolved.
* `no_kicad_installed` — kicad_detect found zero installs, nothing to deploy to.
* `payload_missing` — plugins/kicad/plugin_payload/ wasn't bundled into
  the Tauri app (Tauri bundling regression, like the v1.3.30..v1.3.37
  recorder.html bug — see verify_bundle.py).
* `file_locked` — the destination .py was open (rare; might happen if
  KiCad already has it imported and OS locks the file on Windows).
* `permission_denied` — user-site dir uncreatable (very rare).
* `kicad_running` — KiCad was running when we installed; the running
  process WON'T pick up the new plugin until restarted.

Every error case returns a `_hint` field telling the calling AI exactly
what action it should take to recover.
"""

from __future__ import annotations

import hashlib
import json
import os
import platform
import shutil
import sys
import threading
import traceback
from datetime import datetime
from pathlib import Path
from typing import Optional

# Bridge version we're shipping. Bumped together with adom_bridge.py's
# __version__ — version drift between this file and the payload would
# cause an infinite reinstall loop; the runtime asserts they match.
PAYLOAD_VERSION = "0.7.0"

# Module-level cache: install runs at most ONCE per bridge process boot.
# Multiple concurrent kicad_* commands hitting the bridge during startup
# might all check `ensure_plugin_installed_once()` — the lock guarantees
# only the first one actually does the file copy.
_install_lock = threading.Lock()
_install_result: Optional[dict] = None
_response_attached = False


def _payload_dir() -> Path:
    """Find plugin_payload/ relative to this file. Works in dev (`plugins/kicad/`)
    AND in the bundled Tauri release (`<resources>/plugins/kicad/`)."""
    return Path(__file__).parent / "plugin_payload"


def _user_site_dir_windows(kicad_version: str) -> Optional[Path]:
    """Compute the KiCad USER_SITE path on Windows.

    KiCad's bundled sitecustomize.py builds this path from `CSIDL_PERSONAL`
    (== `%USERPROFILE%\\Documents`) joined with `KiCad\\<version>\\3rdparty\\Python<XY>\\site-packages`.
    We're matching that exactly so adom_bridge.py lands where KiCad's
    embedded interpreter will look for usercustomize.py.

    Returns None if USERPROFILE isn't set (extremely unusual on Windows).
    """
    userprofile = os.environ.get("USERPROFILE")
    if not userprofile:
        return None
    # KiCad 10 uses Python 3.11 — the path includes the no-dot version
    # ("311"). KiCad 9 also used 3.x; we hardcode 311 for now and add a
    # version detection step in Phase 2.x.
    return Path(userprofile) / "Documents" / "KiCad" / kicad_version / "3rdparty" / "Python311" / "site-packages"


def _user_site_dir_mac(kicad_version: str) -> Optional[Path]:
    """Best-effort macOS USER_SITE. NEEDS EMPIRICAL VERIFICATION
    (Phase 1 macOS parity follow-up). Documented here so future Kyle-
    session work can confirm/correct."""
    home = os.environ.get("HOME")
    if not home:
        return None
    return Path(home) / "Library" / "Preferences" / "kicad" / kicad_version / "3rdparty" / "Python311" / "site-packages"


def _user_site_dir(kicad_version: str) -> Optional[Path]:
    if sys.platform == "win32":
        return _user_site_dir_windows(kicad_version)
    if sys.platform == "darwin":
        return _user_site_dir_mac(kicad_version)
    # Linux KiCad install paths vary wildly (flatpak vs apt vs build-
    # from-source). Caller will get "skipped_unsupported_platform".
    return None


def _file_hash(path: Path) -> Optional[str]:
    try:
        with open(path, "rb") as f:
            return hashlib.sha256(f.read()).hexdigest()
    except OSError:
        return None


def _file_state(dest: Path) -> dict:
    """Inspect a destination file and return state for the feedback report.

    Returns a dict with `exists`, `size`, `sha256`, `mtime` (or `_error`).
    """
    if not dest.exists():
        return {"exists": False}
    try:
        st = dest.stat()
        return {
            "exists": True,
            "size": st.st_size,
            "sha256": _file_hash(dest),
            "mtime": datetime.utcfromtimestamp(st.st_mtime).isoformat() + "Z",
        }
    except OSError as e:
        return {"exists": True, "_error": f"stat failed: {e}"}


def _is_kicad_running() -> list[dict]:
    """Best-effort: return PIDs of running KiCad GUI exes. So the AI can
    warn the user that the freshly-installed plugin won't be active until
    they restart KiCad."""
    if sys.platform != "win32":
        return []
    try:
        import subprocess
        out = subprocess.run(
            ["tasklist", "/FO", "CSV", "/NH"],
            capture_output=True, text=True, timeout=10,
        )
        running: list[dict] = []
        targets = {"kicad.exe", "pcbnew.exe", "eeschema.exe", "kicad-cli.exe"}
        for line in out.stdout.splitlines():
            parts = [p.strip('"') for p in line.split(",")]
            if len(parts) < 2:
                continue
            name = parts[0].lower()
            if name in targets:
                try:
                    running.append({"exeName": name, "pid": int(parts[1])})
                except ValueError:
                    pass
        return running
    except (OSError, subprocess.SubprocessError):
        return []


def _install_one_version(version: str, version_info: dict) -> dict:
    """Install / refresh the plugin for ONE KiCad version. Returns a
    dict describing what happened."""
    report: dict = {
        "kicadVersion": version,
        "installPath": version_info.get("base_dir"),
    }
    payload = _payload_dir()
    if not payload.exists():
        report["action"] = "failed"
        report["errorCode"] = "payload_missing"
        report["error"] = f"plugin payload directory not found at {payload}"
        report["_hint"] = (
            "The plugin_payload/ directory wasn't shipped with the bundle. This is a "
            "build regression — see verify_bundle.py in src-tauri/. Report to the user "
            "and ask them to upgrade adom-desktop to a build that includes the payload "
            "(verify with: adom-desktop --version, expecting v1.7.5 or newer)."
        )
        return report

    user_site = _user_site_dir(version)
    if user_site is None:
        report["action"] = "skipped"
        report["errorCode"] = "user_site_unavailable"
        report["error"] = (
            f"could not compute USER_SITE path on {sys.platform} for KiCad {version} "
            f"(missing USERPROFILE / HOME?)"
        )
        report["_hint"] = (
            f"Plugin auto-install is only supported on Windows in v1.7.x; "
            f"this is {sys.platform}. macOS parity is Phase 1 follow-up. "
            "Other kicad_* commands still work via the legacy UI-automation path; "
            "you just won't have the in-process reverse bridge."
        )
        return report

    report["userSitePath"] = str(user_site)
    src_bridge = payload / "adom_bridge.py"
    src_loader = payload / "usercustomize.py"
    dest_bridge = user_site / "adom_bridge.py"
    dest_loader = user_site / "usercustomize.py"

    # Pre-flight: inspect current state so feedback shows before/after
    pre_state = {
        "adom_bridge.py": _file_state(dest_bridge),
        "usercustomize.py": _file_state(dest_loader),
    }
    report["preState"] = pre_state

    # Compare hashes — short-circuit if already up to date
    src_bridge_hash = _file_hash(src_bridge)
    src_loader_hash = _file_hash(src_loader)
    dest_bridge_hash = pre_state["adom_bridge.py"].get("sha256")
    dest_loader_hash = pre_state["usercustomize.py"].get("sha256")

    if dest_bridge_hash == src_bridge_hash and dest_loader_hash == src_loader_hash:
        report["action"] = "already-current"
        report["payloadVersion"] = PAYLOAD_VERSION
        report["message"] = (
            f"plugin already deployed at {user_site} and content matches the shipped "
            f"payload (sha256 identical for both files). No write performed."
        )
        return report

    # Mkdir the user-site if missing. The KiCad sitecustomize.py creates
    # this on first launch too, so we're paving the same path it would.
    try:
        user_site.mkdir(parents=True, exist_ok=True)
    except OSError as e:
        report["action"] = "failed"
        report["errorCode"] = "permission_denied"
        report["error"] = f"could not create {user_site}: {e}"
        report["_hint"] = (
            "Filesystem permission issue creating the KiCad user-site directory. "
            "This is rare on Windows. Possible causes: (1) Documents folder is on "
            "OneDrive with sync errors; (2) antivirus blocking. Report exact error "
            "to the user and have them try: `mkdir -p '<userSitePath>'` manually, "
            "then retry the kicad_* command. If that also fails, the user has a "
            "broken filesystem state."
        )
        return report

    # Copy. shutil.copy2 preserves timestamps which is harmless. We do
    # one file at a time so a half-failure still leaves a coherent state.
    written: list[dict] = []
    for src, dest in ((src_bridge, dest_bridge), (src_loader, dest_loader)):
        try:
            shutil.copy2(src, dest)
            written.append({
                "name": src.name,
                "src": str(src),
                "dest": str(dest),
                "size": dest.stat().st_size,
                "sha256": _file_hash(dest),
            })
        except OSError as e:
            report["action"] = "failed"
            report["errorCode"] = "copy_failed"
            report["error"] = f"copying {src.name} -> {dest}: {e}"
            report["filesWritten"] = written
            report["_hint"] = (
                "Failed mid-copy. Most likely cause on Windows: the destination .py "
                "is currently imported by a running KiCad process which has Windows-"
                "level locks on the file. Fix: ask the user to close ALL KiCad windows "
                "(Project Manager, PCB Editor, Schematic Editor), then retry the "
                "kicad_* command. If the file isn't held, this could also be an "
                "antivirus quarantine — check Windows Defender's protection history."
            )
            return report

    report["action"] = "installed" if not pre_state["adom_bridge.py"]["exists"] else "updated"
    report["payloadVersion"] = PAYLOAD_VERSION
    report["filesWritten"] = written
    report["message"] = (
        f"plugin {'installed' if report['action'] == 'installed' else 'updated to v' + PAYLOAD_VERSION} "
        f"at {user_site}. KiCad GUI processes launched AFTER this point will auto-load it via "
        f"sitecustomize.py + usercustomize.py."
    )
    return report


def _do_install(all_kicad_versions: list[dict]) -> dict:
    """Internal entry. Iterates over every detected KiCad version and
    installs into each. Returns a top-level summary dict."""
    started_at = datetime.utcnow().isoformat() + "Z"

    if not all_kicad_versions:
        return {
            "triggered": True,
            "startedAt": started_at,
            "completedAt": datetime.utcnow().isoformat() + "Z",
            "summary": "no-kicad",
            "errorCode": "no_kicad_installed",
            "perVersion": [],
            "_hint": (
                "Auto-install skipped because kicad_detect found zero KiCad installs. "
                "Report this to the user; recommend installing KiCad from "
                "https://www.kicad.org/download/ (or via `adom-desktop desktop_install_kicad` "
                "which uses winget). After install, the next kicad_* command will retry "
                "the auto-install for you."
            ),
        }

    running = _is_kicad_running()
    per_version = []
    overall_action = "already-current"
    overall_error = None
    for v in all_kicad_versions:
        version = v.get("version") or "10.0"
        try:
            per_version.append(_install_one_version(version, v))
        except Exception as e:  # pylint: disable=broad-except
            per_version.append({
                "kicadVersion": version,
                "action": "failed",
                "errorCode": "unexpected_error",
                "error": f"{type(e).__name__}: {e}",
                "_traceback": traceback.format_exc(),
                "_hint": (
                    "Unexpected exception during plugin install for this version. "
                    "Full traceback in `_traceback`. Report to John on the adom-desktop "
                    "side; this should not happen and is a real bug, not an environment "
                    "issue."
                ),
            })

    # Pick the most-severe action across all versions as the overall summary.
    actions = [pv.get("action") for pv in per_version]
    if any(a == "failed" for a in actions):
        overall_action = "partial-failure"
        overall_error = next(pv for pv in per_version if pv.get("action") == "failed").get("errorCode")
    elif any(a == "installed" for a in actions):
        overall_action = "installed"
    elif any(a == "updated" for a in actions):
        overall_action = "updated"
    elif any(a == "skipped" for a in actions):
        overall_action = "skipped"

    result: dict = {
        "triggered": True,
        "startedAt": started_at,
        "completedAt": datetime.utcnow().isoformat() + "Z",
        "summary": overall_action,
        "payloadVersion": PAYLOAD_VERSION,
        "perVersion": per_version,
        "runningKiCad": running,
    }

    if overall_error:
        result["errorCode"] = overall_error

    # Top-level hint depends on what happened + whether KiCad was running
    if overall_action == "installed":
        if running:
            running_summary = ", ".join(f"{r['exeName']} pid={r['pid']}" for r in running)
            result["_hint"] = (
                f"Plugin successfully INSTALLED for the first time. HOWEVER, KiCad is "
                f"currently running ({running_summary}). Those running processes will "
                f"NOT pick up the new plugin (usercustomize.py only fires at Python "
                f"interpreter init, which happens at KiCad startup). "
                f"\n\nTo use the new reverse bridge, ask the user to: (1) Save any work "
                f"in their open KiCad windows; (2) Close ALL KiCad windows; (3) Reopen "
                f"KiCad. After that, run kicad_bridge_status to verify the plugin came "
                f"online.\n\nIf the user doesn't want to restart KiCad right now, the "
                f"existing non-bridge commands (open_board, install_symbol, etc.) keep "
                f"working through the legacy UI-automation path. The bridge is purely "
                f"additive."
            )
        else:
            result["_hint"] = (
                "Plugin successfully INSTALLED. No KiCad processes were running at "
                "install time, so the next time the user (or your next kicad_* command) "
                "launches KiCad, the reverse-bridge plugin will auto-load. Verify with "
                "kicad_bridge_status once KiCad is up."
            )
    elif overall_action == "updated":
        if running:
            result["_hint"] = (
                "Plugin successfully UPDATED to v" + PAYLOAD_VERSION + ". Running "
                "KiCad processes still have the OLD plugin loaded in-memory. They'll "
                "switch to the new version on next restart. Not urgent — current "
                "session still works."
            )
        else:
            result["_hint"] = (
                "Plugin successfully UPDATED to v" + PAYLOAD_VERSION + ". Next KiCad "
                "launch picks it up automatically."
            )
    elif overall_action == "already-current":
        result["_hint"] = (
            "Plugin already current at v" + PAYLOAD_VERSION + ". No-op. "
            "Future kicad_* commands won't re-check; this status is cached for the "
            "bridge process lifetime."
        )
    elif overall_action == "partial-failure":
        result["_hint"] = (
            "AT LEAST ONE KiCad version's plugin install failed. Inspect perVersion[] "
            "for which exact version's errorCode applies. The bridge still works for "
            "the versions that succeeded; the failed ones fall back to legacy UI "
            "automation. Report the specific errorCode to the user — each one has its "
            "own recovery path documented in the per-version `_hint` field."
        )
    elif overall_action == "skipped":
        result["_hint"] = (
            "Plugin install skipped — see per-version errorCodes. Most common reason: "
            "running on macOS/Linux where Phase 2 hasn't been ported yet. Legacy "
            "kicad_* commands keep working; you just don't have the reverse bridge."
        )

    return result


def ensure_plugin_installed_once(all_kicad_versions: list[dict]) -> dict:
    """Install the plugin if needed; cache the result.

    Idempotent: every call after the first returns the cached dict, marked
    with `triggered: False` and `cached: True` so consumers can tell whether
    this call actually did anything.
    """
    global _install_result  # pylint: disable=global-statement
    with _install_lock:
        if _install_result is not None:
            cached = dict(_install_result)
            cached["triggered"] = False
            cached["cached"] = True
            return cached
        # Honor disable env var. Useful for development/diagnostic.
        if os.environ.get("ADOM_KICAD_PLUGIN_AUTO_INSTALL_DISABLE") == "1":
            _install_result = {
                "triggered": False,
                "summary": "disabled",
                "errorCode": "disabled_via_env",
                "_hint": (
                    "ADOM_KICAD_PLUGIN_AUTO_INSTALL_DISABLE=1 in the kicad bridge's "
                    "environment. To install manually, call: "
                    "adom-desktop kicad_install_plugin (Phase 2.1)."
                ),
            }
            return dict(_install_result)
        _install_result = _do_install(all_kicad_versions)
        return _install_result


def attach_to_response_if_first(result: dict) -> dict:
    """Attach `pluginAutoInstall` to a response dict EXACTLY ONCE per
    bridge boot. After the AI has been notified on first command, future
    responses skip the field to keep payloads small.

    bridge_status / bridge_call ALWAYS get the field attached regardless,
    since those verbs explicitly ask about plugin state.
    """
    global _response_attached  # pylint: disable=global-statement
    if not isinstance(result, dict):
        return result
    if _response_attached and _install_result is not None:
        return result
    if _install_result is None:
        return result
    result["pluginAutoInstall"] = dict(_install_result)
    _response_attached = True
    return result


def force_attach(result: dict) -> dict:
    """For bridge_status / bridge_call — always attach the current install
    state, regardless of `_response_attached`. Useful when the AI is
    explicitly inspecting plugin state."""
    if not isinstance(result, dict) or _install_result is None:
        return result
    result["pluginAutoInstall"] = dict(_install_result)
    return result


def get_cached_result() -> Optional[dict]:
    return dict(_install_result) if _install_result is not None else None


def force_install(all_kicad_versions: list[dict]) -> dict:
    """Bypass the once-per-boot cache and re-run the install. The new
    result replaces the cache so subsequent attach_to_response_if_first()
    calls see the fresh state.

    Used by the explicit `kicad_install_plugin` verb."""
    global _install_result, _response_attached  # pylint: disable=global-statement
    with _install_lock:
        _install_result = _do_install(all_kicad_versions)
        # Reset the "attached on first response" flag so future calls
        # surface the fresh install state once more.
        _response_attached = False
        return dict(_install_result)


def handle_install_plugin(kicad_info: dict, args: dict) -> dict:
    """Explicit kicad_install_plugin verb. Bypasses the auto-install
    cache and forces a fresh deploy + status check.

    Args:
      force:  default true. If false, behaves like the auto-install path
              (returns cached result if it's already run this boot).

    Returns the full install report. Useful for: debugging when the
    plugin isn't loading, updating the payload between releases without
    restarting the bridge, recovering from a partial-failure install.
    """
    # Need the full all_kicad_versions list from server.py — caller
    # passed kicad_info (the default version). Re-detect to get all.
    from kicad_detect import detect_all_kicad_versions
    all_versions = detect_all_kicad_versions()

    if args.get("force") is False:
        # Honor explicit force=false by behaving like the auto path
        return {"success": True, "result": ensure_plugin_installed_once(all_versions)}

    result = force_install(all_versions)
    return {
        "success": result.get("summary") not in ("partial-failure",),
        "result": result,
        "_hint": (
            "force-reinstall complete. Future kicad_* commands will show "
            "this fresh install state in their pluginAutoInstall fields. "
            "If KiCad was running during install, ask the user to restart "
            "those processes — the in-memory Python interpreter still has "
            "the OLD plugin loaded until next launch."
        ),
    }