emma-hermes/hermes_cli/managed_uv.py

"""Managed uv — one path, no guessing.

Hermes owns its own uv binary at ``$HERMES_HOME/bin/uv`` (or ``uv.exe`` on
Windows).  Every code path that needs uv resolves it from that single location.
If the binary is missing, ``ensure_uv()`` bootstraps it via the official
standalone installer with ``UV_UNMANAGED_INSTALL`` / ``UV_INSTALL_DIR`` pointed
at ``$HERMES_HOME/bin`` so the installer writes directly there — no PATH
probing, no conda guards, no multi-location resolution chains.

When ``ensure_uv()`` bootstraps uv for the first time (i.e. there was no
managed uv before), it returns ``(path, True)`` instead of just ``path``.
Callers in the update path use that signal to nuke and recreate the venv
with the now-current managed uv, guaranteeing a Python with FTS5.
"""

from __future__ import annotations

import logging
import os
import platform
import shutil
import subprocess
import tempfile
from pathlib import Path
from typing import Optional, Tuple

from hermes_constants import get_hermes_home

logger = logging.getLogger(__name__)

# ---------------------------------------------------------------------------
# Public helpers
# ---------------------------------------------------------------------------

def managed_uv_path() -> Path:
    """Return the path where Hermes keeps *its* uv binary.

    ``$HERMES_HOME/bin/uv`` on POSIX, ``$HERMES_HOME\\bin\\uv.exe`` on
    Windows.  The directory may not exist yet — callers should use
    ``ensure_uv()`` to bootstrap it.
    """
    home = get_hermes_home()
    if platform.system() == "Windows":
        return home / "bin" / "uv.exe"
    return home / "bin" / "uv"


def resolve_uv() -> Optional[str]:
    """Return the managed uv path if it exists, else ``None``.

    No side effects — pure lookup.
    """
    p = managed_uv_path()
    if p.is_file() and os.access(p, os.X_OK):
        return str(p)
    return None


def ensure_uv() -> Tuple[Optional[str], bool]:
    """Return the managed uv path, installing it first if necessary.

    Returns ``(path, freshly_bootstrapped)`` where *freshly_bootstrapped* is
    ``True`` when we just installed managed uv for the first time (there was
    no managed uv before this call).  Callers can use that signal to rebuild
    the venv so Python is guaranteed to have FTS5.

    On failure returns ``(None, False)`` (never raises) so callers can fall
    back to pip gracefully.
    """
    existing = resolve_uv()
    if existing:
        return (existing, False)

    target = managed_uv_path()
    target.parent.mkdir(parents=True, exist_ok=True)

    print(f"  → Installing managed uv into {target.parent} ...")

    try:
        _install_uv(target)
    except Exception as exc:
        logger.warning("Managed uv install failed: %s", exc)
        print(f"  ✗ Failed to install managed uv: {exc}")
        return (None, False)

    # Verify
    result = resolve_uv()
    if result:
        version = subprocess.run(
            [result, "--version"],
            capture_output=True,
            text=True,
            check=False,
        ).stdout.strip()
        print(f"  ✓ Managed uv installed ({version})")
    else:
        print("  ✗ Managed uv install appeared to succeed but binary not found")
    return (result, result is not None)


def rebuild_venv(uv_bin: str, venv_dir: Path, python_version: str = "3.11") -> bool:
    """Nuke and recreate the venv with managed uv.

    Called when managed uv is first bootstrapped on an existing install — the
    old venv may point to a Python without FTS5, so we rebuild it with a
    fresh interpreter from the current managed uv.  Returns ``True`` on
    success.

    On Windows, ``shutil.rmtree(..., ignore_errors=True)`` can silently leave
    the venv directory partially intact when another process is holding an
    open handle to a file inside it (typical culprits: a running
    ``hermes.exe`` REPL, the gateway, AV scanners). If we don't notice that
    and just call ``uv venv``, uv refuses with
    ``Caused by: A directory already exists at: venv`` and the *whole
    update* falls back to installing on top of the stale venv — which has
    historically produced partial installs where a freshly added dependency
    (e.g. ``pathspec``) silently fails to land. Retry with ``--clear`` to
    force uv past that condition before giving up.
    """
    if venv_dir.exists():
        print(f"  → Rebuilding venv (old Python may lack FTS5)...")
        shutil.rmtree(venv_dir, ignore_errors=True)

    def _run_uv_venv(extra_args: list[str]) -> subprocess.CompletedProcess[str]:
        return subprocess.run(
            [uv_bin, "venv", str(venv_dir), "--python", python_version, *extra_args],
            capture_output=True,
            text=True,
            check=False,
        )

    result = _run_uv_venv([])

    # If uv refused because the directory still exists (rmtree above was
    # blocked by an open file handle, common on Windows), retry with
    # --clear so uv overwrites it. Match on stderr because uv's exit code
    # alone doesn't distinguish "dir exists" from real failures.
    if result.returncode != 0 and "already exists" in (result.stderr or "").lower():
        print("  → venv dir not fully removed (likely an open file handle); retrying with --clear...")
        result = _run_uv_venv(["--clear"])

    if result.returncode == 0:
        venv_python = venv_dir / ("Scripts" if platform.system() == "Windows" else "bin") / "python"
        py_ver = subprocess.run(
            [str(venv_python), "--version"],
            capture_output=True,
            text=True,
            check=False,
        ).stdout.strip()
        print(f"  ✓ venv rebuilt ({py_ver})")
        return True
    else:
        logger.warning("venv rebuild failed: %s", result.stderr)
        print(f"  ✗ venv rebuild failed: {result.stderr.strip()}")
        return False


def update_managed_uv() -> Optional[str]:
    """Run ``uv self update`` on the managed uv binary.

    Call this during ``hermes update`` so the managed copy stays current.
    Returns the managed path on success, ``None`` if uv isn't available or
    the self-update fails (non-fatal — the old version still works).
    """
    existing = resolve_uv()
    if not existing:
        # Not installed yet — ensure_uv() will handle that elsewhere.
        return None

    result = subprocess.run(
        [existing, "self", "update"],
        capture_output=True,
        text=True,
        check=False,
    )
    if result.returncode == 0:
        version = subprocess.run(
            [existing, "--version"],
            capture_output=True,
            text=True,
            check=False,
        ).stdout.strip()
        print(f"  ✓ Managed uv updated ({version})")
    else:
        # Non-fatal — old uv still works fine.
        logger.debug("uv self update failed (rc=%d): %s", result.returncode, result.stderr)
    return existing


# ---------------------------------------------------------------------------
# Installer internals
# ---------------------------------------------------------------------------

def _install_uv(target: Path) -> None:
    """Bootstrap uv into *target* using the official standalone installer.

    Uses ``UV_UNMANAGED_INSTALL`` (POSIX) or ``UV_INSTALL_DIR`` (Windows)
    so the astral installer writes the binary directly into
    ``$HERMES_HOME/bin/`` instead of ``~/.local/bin/``.
    """
    system = platform.system()
    env = {
        **os.environ,
        # Tell the astral installer to drop the binary in our dir, not
        # ~/.local/bin.  UV_UNMANAGED_INSTALL is the POSIX env var; Windows
        # uses UV_INSTALL_DIR.
        "UV_UNMANAGED_INSTALL": str(target.parent),
        "UV_INSTALL_DIR": str(target.parent),
    }

    if system == "Windows":
        _install_uv_windows(env)
    else:
        _install_uv_posix(env)


def _install_uv_posix(env: dict[str, str]) -> None:
    """Download + sh the POSIX installer (two-stage to avoid curl|sh pitfalls)."""
    with tempfile.NamedTemporaryFile(suffix=".sh", delete=False) as f:
        installer_path = f.name

    try:
        subprocess.run(
            ["curl", "-LsSf", "https://astral.sh/uv/install.sh", "-o", installer_path],
            check=True,
            capture_output=True,
        )
        subprocess.run(
            ["sh", installer_path],
            env=env,
            check=True,
            capture_output=True,
        )
    finally:
        try:
            os.unlink(installer_path)
        except OSError:
            pass


def _install_uv_windows(env: dict[str, str]) -> None:
    """Invoke the PowerShell installer."""
    cmd = (
        'irm https://astral.sh/uv/install.ps1 | iex'
    )
    subprocess.run(
        ["powershell", "-ExecutionPolicy", "Bypass", "-c", cmd],
        env=env,
        check=True,
        capture_output=True,
    )