binprovider_uv.py

#!/usr/bin/env python3
__package__ = "abxpkg"
import json
import os
import shutil
import stat
import subprocess
import sys
import urllib.error
import urllib.parse
import urllib.request
from collections.abc import Mapping
from datetime import UTC
from importlib.metadata import distributions
from pathlib import Path
from typing import ClassVar, Self

from platformdirs import user_cache_path
from pydantic import Field, TypeAdapter, computed_field, model_validator

from .base_types import (
    BinName,
    BinProviderName,
    HostBinPath,
    InstallArgs,
    PATHStr,
    abxpkg_cache_dir_default,
    abxpkg_install_root_default,
    bin_abspath,
)
from .binary import Binary
from .binprovider import (
    BinProvider,
    EnvProvider,
    env_flag_is_true,
    log_method_call,
    remap_kwargs,
)
from .config import load_derived_cache
from .logging import format_command, format_subprocess_output, get_logger
from .semver import SemVer

USER_CACHE_PATH = user_cache_path("uv", "abxpkg")
logger = get_logger(__name__)


class UvProvider(BinProvider):
    """Standalone ``uv`` package manager provider.
    Has two modes, picked based on whether ``install_root`` is set:
    1. **Hermetic venv mode** (``install_root=Path(...)``): creates a dedicated
       venv at the requested path via ``uv venv`` and installs packages
       into it via ``uv pip install --python <venv>/bin/python``, the same
       way ``PipProvider`` does when configured with ``install_root``. This is
       the idiomatic "install a Python library + its CLI entrypoints into
       an isolated environment" path.
    2. **Global tool mode** (``install_root=None``): delegates to
       ``uv tool install`` which lays out a fresh venv under
       ``UV_TOOL_DIR`` per tool and writes shims into ``UV_TOOL_BIN_DIR``.
       This is the idiomatic "install a CLI tool globally" path.
    Security:
    - ``--no-build`` for ``postinstall_scripts=False`` (wheels only).
    - ``--exclude-newer=<ISO8601>`` for ``min_release_age``.
    """

    name: BinProviderName = "uv"
    _log_emoji = "🚀"
    INSTALLER_BIN: BinName = "uv"
    INSTALLER_BINPROVIDERS: ClassVar[tuple[BinProviderName, ...] | None] = (
        "env",
        "pip",
    )
    PATH: PATHStr = ""  # Starts empty; setup_PATH() lazily uses install_root/venv/bin in venv mode, or UV_TOOL_BIN_DIR/~/.local/bin in tool mode.
    postinstall_scripts: bool | None = Field(
        default_factory=lambda: env_flag_is_true("ABXPKG_POSTINSTALL_SCRIPTS"),
        repr=False,
    )
    min_release_age: float | None = Field(
        default_factory=lambda: float(os.environ.get("ABXPKG_MIN_RELEASE_AGE", "7")),
        repr=False,
    )
    # None = global ``uv tool`` mode, otherwise install_root is the provider root.
    # In install_root mode the actual virtualenv lives at install_root/venv
    # so provider metadata like derived.env can stay next to it.
    # Default: ABXPKG_UV_ROOT > ABXPKG_LIB_DIR/uv > None.
    install_root: Path | None = Field(
        default_factory=lambda: abxpkg_install_root_default("uv"),
        validation_alias="uv_venv",
    )
    # Managed venv mode fills this with ``<install_root>/venv/bin`` in detect_euid_to_use();
    # tool mode may accept an explicit ``uv_tool_bin_dir`` override or leave it unset.
    bin_dir: Path | None = Field(default=None, validation_alias="uv_tool_bin_dir")

    @staticmethod
    def _runtime_site_packages(venv_root: Path) -> Path:
        python_lib = f"python{sys.version_info.major}.{sys.version_info.minor}"
        return venv_root / "lib" / python_lib / "site-packages"

    def env_projection_target(self, source_path: Path) -> Path | None:
        """Uv-managed venv entrypoints are safe to expose through env/bin."""
        if self.install_root is None:
            return None
        source_path = source_path.expanduser().absolute()
        for venv_root in self._managed_venv_roots(source_path.name):
            bin_dir = venv_root / "bin"
            try:
                source_path.relative_to(bin_dir)
            except ValueError:
                continue
            if source_path.is_file() and os.access(source_path, os.X_OK):
                return source_path
        return None

    @computed_field
    @property
    def ENV(self) -> "dict[str, str]":
        env: dict[str, str] = {
            "UV_ACTIVE": "1",
            "UV_CACHE_DIR": str(self.cache_dir),
            "PYTHONDONTWRITEBYTECODE": "1",
        }
        if self.install_root:
            venv_root = self.install_root / "venv"
            env["VIRTUAL_ENV"] = str(venv_root)
            site_packages = self._runtime_site_packages(venv_root)
            if site_packages.is_dir():
                env["PYTHONPATH"] = ":" + str(site_packages)
            # `uv` can install side dependencies into package-scoped venvs under
            # the same provider root. They are still part of this provider's
            # runtime surface, so expose them here instead of making callers know
            # or rebuild the provider's filesystem layout.
            package_paths: list[str] = []
            packages_root = self.install_root / "packages"
            if packages_root.is_dir():
                for package_root in sorted(packages_root.iterdir()):
                    package_venv = package_root / "venv"
                    package_bin = package_venv / "bin"
                    if package_bin.is_dir():
                        env["PATH"] = f"{env.get('PATH', '')}:{package_bin}"
                    site_packages = self._runtime_site_packages(package_venv)
                    if site_packages.is_dir():
                        package_paths.append(str(site_packages))
            if package_paths:
                env["PYTHONPATH"] = (
                    f"{env.get('PYTHONPATH', '')}:{os.pathsep.join(package_paths)}"
                )
            return env
        env["UV_TOOL_DIR"] = str(self.tool_dir)
        if self.bin_dir:
            env["UV_TOOL_BIN_DIR"] = str(self.bin_dir)
        return env

    def supports_min_release_age(self, action, no_cache: bool = False) -> bool:
        return action in ("install", "update")

    def supports_postinstall_disable(self, action, no_cache: bool = False) -> bool:
        return action in ("install", "update")

    def _cached_installer_binary(self, no_cache: bool = False):
        if not no_cache and self._INSTALLER_BINARY and self._INSTALLER_BINARY.is_valid:
            return self._INSTALLER_BINARY

        derived_env_path = self.derived_env_path
        if no_cache or not derived_env_path or not derived_env_path.is_file():
            return None

        cache = load_derived_cache(derived_env_path)
        for cached_record in list(cache.values()):
            if not isinstance(cached_record, dict):
                continue
            if cached_record.get("provider_name") != self.name or cached_record.get(
                "bin_name",
            ) != str(self.INSTALLER_BIN):
                continue
            cached_abspath = cached_record.get("abspath")
            if not isinstance(cached_abspath, str):
                continue
            cached_context = cached_record.get("cache_context")
            cached_context_hash = cached_record.get("cache_context_hash")
            loaded = self.load_cached_binary(
                self.INSTALLER_BIN,
                Path(cached_abspath),
                cache_context=cached_context
                if isinstance(cached_context, str)
                else None,
                cache_context_hash=(
                    cached_context_hash
                    if isinstance(cached_context_hash, str)
                    else None
                ),
            )
            if loaded and loaded.loaded_abspath:
                self._INSTALLER_BINARY = loaded
                return loaded
        return None

    def _installer_provider_root(self) -> Path:
        lib_dir = os.environ.get("ABXPKG_LIB_DIR")
        if (
            self.install_root is not None
            and lib_dir
            and str(self.install_root).startswith(lib_dir.rstrip("/") + "/")
        ):
            return Path(lib_dir) / "pip" / "packages" / "uv"
        if self.install_root is not None:
            return self.install_root / "pip"
        return self.cache_dir / "pip"

    def _load_installer_at(self, abspath: Path, no_cache: bool = False):
        loaded = EnvProvider(
            PATH=str(abspath.parent),
            install_root=None,
            bin_dir=None,
        ).load(bin_name=self.INSTALLER_BIN, no_cache=True)
        if loaded and loaded.loaded_abspath:
            if loaded.loaded_version and loaded.loaded_sha256:
                self.write_cached_binary(
                    self.INSTALLER_BIN,
                    loaded.loaded_abspath,
                    loaded.loaded_version,
                    loaded.loaded_sha256,
                    resolved_provider_name=(
                        loaded.loaded_binprovider.name
                        if loaded.loaded_binprovider is not None
                        else self.name
                    ),
                    resolved_provider=loaded.loaded_binprovider,
                    cache_kind="dependency",
                )
            self._INSTALLER_BINARY = loaded
            return loaded
        return None

    def _install_installer_binary(self, no_cache: bool = False):
        from .binprovider_pip import PipProvider

        pip_root = self._installer_provider_root()
        loaded = Binary(
            name=self.INSTALLER_BIN,
            binproviders=[
                PipProvider(
                    install_root=pip_root,
                    postinstall_scripts=True,
                    min_release_age=0,
                ),
            ],
            postinstall_scripts=True,
            min_release_age=0,
        ).install(no_cache=no_cache)
        if loaded and loaded.loaded_abspath:
            if loaded.loaded_version and loaded.loaded_sha256:
                self.write_cached_binary(
                    self.INSTALLER_BIN,
                    loaded.loaded_abspath,
                    loaded.loaded_version,
                    loaded.loaded_sha256,
                    resolved_provider_name=(
                        loaded.loaded_binprovider.name
                        if loaded.loaded_binprovider is not None
                        else self.name
                    ),
                    resolved_provider=loaded.loaded_binprovider,
                    cache_kind="dependency",
                )
            self._INSTALLER_BINARY = loaded
        return loaded

    def INSTALLER_BINARY(self, no_cache: bool = False):
        cached = self._cached_installer_binary(no_cache=no_cache)
        if cached is not None:
            return cached

        env_var = f"{self.INSTALLER_BIN.upper()}_BINARY"
        manual = os.environ.get(env_var)
        if manual and os.path.isabs(manual) and Path(manual).is_file():
            loaded = self._load_installer_at(Path(manual), no_cache=no_cache)
            if loaded is not None:
                return loaded

        host_installer = bin_abspath(
            self.INSTALLER_BIN,
            PATH=os.environ.get("PATH", ""),
        )
        if host_installer:
            loaded = self._load_installer_at(host_installer, no_cache=no_cache)
            if loaded is not None:
                return loaded

        local_installer = (
            self._installer_provider_root()
            / "venv"
            / "bin"
            / str(
                self.INSTALLER_BIN,
            )
        )
        if local_installer.is_file() and os.access(local_installer, os.X_OK):
            loaded = self._load_installer_at(local_installer, no_cache=no_cache)
            if loaded is not None:
                return loaded

        return self._install_installer_binary(no_cache=no_cache)

    @computed_field
    @property
    def is_valid(self) -> bool:
        if self.install_root:
            venv_python = self.install_root / "venv" / "bin" / "python"
            if venv_python.exists() and not (
                venv_python.is_file() and os.access(venv_python, os.X_OK)
            ):
                return False
        return super().is_valid

    @model_validator(mode="after")
    def detect_euid_to_use(self) -> Self:
        """Derive uv's managed virtualenv bin_dir from install_root when configured."""
        if self.bin_dir is None and self.install_root is not None:
            self.bin_dir = self.install_root / "venv" / "bin"
        return self

    @property
    def cache_dir(self) -> Path:
        """Return uv's shared download/build cache dir."""
        return abxpkg_cache_dir_default("uv") or Path(
            os.environ.get("UV_CACHE_DIR") or USER_CACHE_PATH,
        )

    def _cache_args(self, *, no_cache: bool = False) -> list[str]:
        return ["--no-cache"] if no_cache else []

    def _cache_context(self, bin_name: BinName) -> str:
        provider_config = json.loads(super()._cache_context(bin_name))
        provider_config["runtime_python"] = {
            "executable": str(Path(sys.executable).absolute()),
            "cache_tag": sys.implementation.cache_tag,
        }
        return json.dumps(
            provider_config,
            default=str,
            separators=(",", ":"),
            sort_keys=True,
        )

    @property
    def tool_dir(self) -> Path:
        """Return uv's global tool install root used in ``uv tool`` mode."""
        return Path(
            os.environ.get("UV_TOOL_DIR")
            or (Path("~").expanduser() / ".local" / "share" / "uv" / "tools"),
        )

    def setup_PATH(self, no_cache: bool = False) -> None:
        """Populate PATH on first use from install_root/venv/bin in venv mode, or UV tool bin dirs in tool mode."""
        if self.install_root:
            bin_dir = self.bin_dir
            assert bin_dir is not None
            self.PATH = self._merge_PATH(
                bin_dir,
                PATH=self.PATH,
                prepend=True,
            )
            packages_root = self.install_root / "packages"
            if packages_root.is_dir():
                for package_root in sorted(packages_root.iterdir(), reverse=True):
                    package_bin_dir = package_root / "venv" / "bin"
                    if package_bin_dir.is_dir():
                        self.PATH = self._merge_PATH(
                            package_bin_dir,
                            PATH=self.PATH,
                            prepend=True,
                        )
        elif self.bin_dir:
            self.PATH = self._merge_PATH(
                self.bin_dir,
                PATH=self.PATH,
                prepend=True,
            )
        else:
            default_bin = Path(
                os.environ.get("UV_TOOL_BIN_DIR")
                or (Path("~").expanduser() / ".local" / "bin"),
            )
            self.PATH = self._merge_PATH(default_bin, PATH=self.PATH, prepend=True)
        super().setup_PATH(no_cache=no_cache)

    @log_method_call()
    def setup(
        self,
        *,
        postinstall_scripts: bool | None = None,
        min_release_age: float | None = None,
        min_version: SemVer | None = None,
        no_cache: bool = False,
    ) -> None:
        if self.euid is None:
            self.euid = self._managed_install_euid()
        if self.install_root:
            self._ensure_venv(no_cache=no_cache)
        else:
            self.tool_dir.mkdir(parents=True, exist_ok=True)
            if self.bin_dir:
                self.bin_dir.mkdir(parents=True, exist_ok=True)

    def detect_euid(
        self,
        owner_paths=(),
        preserve_root: bool = False,
    ) -> int:
        if self.install_root is not None:
            return self._managed_install_euid()
        return super().detect_euid(
            owner_paths=owner_paths,
            preserve_root=preserve_root,
        )

    def _managed_install_euid(self) -> int:
        """Return the uid that should own uv-managed package environments."""
        owner_paths = (self.install_root,)
        if self.install_root is not None and self.install_root.is_dir():
            existing_owner = super().detect_euid(
                owner_paths=owner_paths,
                preserve_root=True,
            )
            if existing_owner != 0:
                return existing_owner
        sudo_uid = self._sudo_managed_install_euid()
        if sudo_uid is not None:
            return sudo_uid
        return super().detect_euid(
            owner_paths=owner_paths,
            preserve_root=True,
        )

    def _sudo_managed_install_euid(
        self,
        *,
        current_euid: int | None = None,
        environ: Mapping[str, str] | None = None,
    ) -> int | None:
        """Return SUDO_UID for root-invoked managed installs, when usable."""
        if self.install_root is None:
            return None
        if current_euid is None:
            current_euid = os.geteuid()
        if current_euid != 0:
            return None
        sudo_uid = (environ or os.environ).get("SUDO_UID")
        if not sudo_uid:
            return None
        try:
            uid = int(sudo_uid)
        except ValueError:
            return None
        if uid > 0 and self.uid_has_passwd_entry(uid):
            return uid
        return None

    def _managed_package_root(
        self,
        bin_name: BinName,
        install_args: InstallArgs | None = None,
    ) -> Path:
        """Return the mutable install root for a managed Python CLI package."""
        assert self.install_root is not None
        if self.install_root.parent.name == "packages":
            return self.install_root
        if self.install_root.name != "uv":
            return self.install_root
        package_name = self.get_package_names(bin_name, install_args)[0]
        return self.install_root / "packages" / package_name

    def _managed_venv_roots(self, bin_name: BinName | None = None) -> list[Path]:
        """Return candidate managed uv venvs, with package-specific roots first."""
        roots: list[Path] = []
        if self.install_root is None:
            return roots
        if bin_name is not None:
            roots.append(self._managed_package_root(bin_name) / "venv")
        packages_root = self.install_root / "packages"
        if packages_root.is_dir():
            roots.extend(
                package_root / "venv"
                for package_root in sorted(packages_root.iterdir())
                if (package_root / "venv" / "bin").is_dir()
            )
        roots.append(self.install_root / "venv")
        deduped: list[Path] = []
        for root in roots:
            if root not in deduped:
                deduped.append(root)
        return deduped

    def _managed_path_matches_runtime(
        self,
        bin_name: BinName,
        abspath: Path,
    ) -> bool:
        resolved_abspath = abspath.expanduser().resolve(strict=False)
        for venv_root in self._managed_venv_roots(bin_name):
            try:
                resolved_abspath.relative_to(
                    venv_root.resolve(strict=False),
                )
            except ValueError:
                continue
            runtime_site_packages = self._runtime_site_packages(venv_root)
            if not runtime_site_packages.is_dir():
                return False
            if not resolved_abspath.is_relative_to(
                (venv_root / "lib").resolve(strict=False),
            ):
                return True
            return resolved_abspath.is_relative_to(runtime_site_packages.resolve())
        return True

    def _ensure_venv(self, root: Path | None = None, *, no_cache: bool = False) -> None:
        """Create a managed uv virtualenv on first use when install_root is pinned."""
        assert self.install_root is not None
        target_root = root or self.install_root
        venv_root = target_root / "venv"
        venv_python = venv_root / "bin" / "python"
        if venv_python.is_file() and os.access(venv_python, os.X_OK):
            self._ensure_managed_root_access(target_root)
            proc = subprocess.run(
                [
                    str(venv_python),
                    "-c",
                    "import sys; print(f'{sys.version_info.major}.{sys.version_info.minor}')",
                ],
                capture_output=True,
                text=True,
                timeout=5,
                check=False,
            )
            expected = f"{sys.version_info.major}.{sys.version_info.minor}"
            if proc.returncode == 0 and proc.stdout.strip() == expected:
                return
            # Python hook scripts run under the interpreter that launched
            # abxpkg, but provider deps are exposed through PYTHONPATH. A stale
            # managed venv from a different Python minor can point compiled
            # wheels such as pydantic-core at an incompatible ABI, so rebuild it
            # before any package install/load can leak those paths downstream.
            shutil.rmtree(venv_root, ignore_errors=True)
        target_root.mkdir(parents=True, exist_ok=True)
        self._ensure_managed_root_access(target_root)
        installer_bin = self.INSTALLER_BINARY(no_cache=no_cache).loaded_abspath
        assert installer_bin
        proc = self.exec(
            bin_name=installer_bin,
            cmd=[
                *self._cache_args(no_cache=no_cache),
                "venv",
                "--python",
                sys.executable,
                str(venv_root),
            ],
            quiet=True,
            timeout=self.install_timeout,
        )
        if proc.returncode != 0:
            self._raise_proc_error("install", ["uv venv"], proc)
        self._ensure_managed_root_access(target_root)

    def _ensure_managed_root_access(self, root: Path) -> None:
        """Keep uv-created managed venv roots usable by this provider's runtime UID.

        ArchiveBox can invoke installers from a sudo-created process whose real
        uid is still root while the effective/runtime uid is the collection
        owner. Plain ``os.access`` probes can therefore accept files that only
        root can execute. ``UvProvider`` owns these package-scoped venv trees, so
        normalize just the venv root directories and entrypoints after uv mutates
        them. Do not recurse through uv's shared cache; that path is large and
        hot.
        """
        if self.install_root is None:
            return
        try:
            root = root.expanduser().resolve(strict=False)
            install_root = self.install_root.expanduser().resolve(strict=False)
            if root != install_root and not root.is_relative_to(install_root):
                return
        except OSError:
            return

        uid = self.EUID
        try:
            gid = self.get_pw_record(uid).pw_gid
        except KeyError:
            gid = os.getegid()

        paths = [
            root.parent,
            root,
            root / "venv",
            root / "venv" / "bin",
            root / "venv" / "lib",
        ]
        for path in paths:
            if not path.exists():
                continue
            try:
                path_stat = path.stat()
                if (os.geteuid() == 0 or os.getuid() == 0) and (
                    path_stat.st_uid != uid or path_stat.st_gid != gid
                ):
                    os.chown(path, uid, gid)
            except PermissionError:
                pass
            try:
                current_mode = path.stat().st_mode
                required_mode = (
                    current_mode
                    | stat.S_IRUSR
                    | stat.S_IWUSR
                    | stat.S_IXUSR
                    | stat.S_IRGRP
                    | stat.S_IXGRP
                    | stat.S_IROTH
                    | stat.S_IXOTH
                )
                if required_mode != current_mode:
                    path.chmod(required_mode)
            except PermissionError:
                pass

        bin_dir = root / "venv" / "bin"
        if not bin_dir.is_dir():
            return
        for entrypoint in bin_dir.iterdir():
            if not (entrypoint.is_file() or entrypoint.is_symlink()):
                continue
            try:
                if os.geteuid() == 0 or os.getuid() == 0:
                    entrypoint_stat = (
                        entrypoint.lstat()
                        if entrypoint.is_symlink()
                        else entrypoint.stat()
                    )
                    if entrypoint_stat.st_uid != uid or entrypoint_stat.st_gid != gid:
                        if entrypoint.is_symlink():
                            os.lchown(entrypoint, uid, gid)
                        else:
                            os.chown(entrypoint, uid, gid)
            except (NotImplementedError, PermissionError):
                pass
            if entrypoint.is_symlink():
                continue
            try:
                current_mode = entrypoint.stat().st_mode
                required_mode = (
                    current_mode
                    | stat.S_IRUSR
                    | stat.S_IXUSR
                    | stat.S_IRGRP
                    | stat.S_IXGRP
                    | stat.S_IROTH
                    | stat.S_IXOTH
                )
                if required_mode != current_mode:
                    entrypoint.chmod(required_mode)
            except PermissionError:
                pass

    @staticmethod
    def _release_age_cutoff(min_release_age: float | None) -> str | None:
        """Translate ``min_release_age`` days into uv's ``--exclude-newer`` timestamp."""
        if min_release_age is None or min_release_age <= 0:
            return None
        from datetime import datetime, timedelta

        return (datetime.now(UTC) - timedelta(days=min_release_age)).strftime(
            "%Y-%m-%dT%H:%M:%SZ",
        )

    def _pip_flags(
        self,
        *,
        install_args: InstallArgs,
        postinstall_scripts: bool,
        min_release_age: float | None,
    ) -> list[str]:
        """Build the shared ``uv pip`` security flags list."""
        combined = tuple(install_args)
        flags: list[str] = []
        if not postinstall_scripts and not any(
            arg == "--no-build" or arg.startswith("--no-build=") for arg in combined
        ):
            flags.append("--no-build")
        cutoff = self._release_age_cutoff(min_release_age)
        if cutoff and not any(
            arg == "--exclude-newer" or arg.startswith("--exclude-newer=")
            for arg in combined
        ):
            flags.append(f"--exclude-newer={cutoff}")
        return flags

    def cached_binary_state_mismatch(
        self,
        bin_name: BinName,
        cached_record: Mapping[str, object],
    ) -> bool:
        cached_abspath = cached_record.get("abspath")
        return (
            super().cached_binary_state_mismatch(bin_name, cached_record)
            or not isinstance(cached_abspath, str)
            or not self._managed_path_matches_runtime(
                bin_name,
                Path(cached_abspath),
            )
        )

    def get_cache_info(
        self,
        bin_name: BinName,
        abspath: HostBinPath,
    ) -> dict[str, list[Path]] | None:
        cache_info = super().get_cache_info(bin_name, abspath)
        if cache_info is None or self.install_root is None:
            return cache_info

        package_name = self.get_package_names(str(bin_name))[0]
        normalized_name = package_name.lower().replace("-", "_")
        venv_roots = self._managed_venv_roots(str(bin_name))
        owning_venv = next(
            (
                venv_root
                for venv_root in venv_roots
                if Path(abspath)
                .resolve(strict=False)
                .is_relative_to(
                    venv_root.resolve(strict=False),
                )
            ),
            venv_roots[0] if venv_roots else self.install_root / "venv",
        )
        venv_python = owning_venv / "bin" / "python"
        if venv_python.exists():
            cache_info["fingerprint_paths"].append(venv_python)
        metadata_files = sorted(
            (owning_venv / "lib").glob(
                f"python*/site-packages/{normalized_name}*.dist-info/METADATA",
            ),
        ) or sorted(
            (owning_venv / "lib").glob(
                f"python*/site-packages/{normalized_name}*.dist-info/PKG-INFO",
            ),
        )
        if metadata_files:
            cache_info["fingerprint_paths"].append(metadata_files[0])
        return cache_info

    def _version_from_uv_metadata(
        self,
        package_name: str,
        timeout: int | None = None,
        no_cache: bool = False,
        abspath: HostBinPath | None = None,
    ) -> SemVer | None:
        """Read installed distribution metadata or global ``uv tool list``."""
        if self.install_root:
            if abspath is not None:
                owning_venv = next(
                    (
                        parent
                        for parent in abspath.resolve().parents
                        if (parent / "pyvenv.cfg").is_file()
                    ),
                    None,
                )
                venv_roots = [owning_venv] if owning_venv is not None else []
            else:
                venv_roots = self._managed_venv_roots(package_name)
            for venv_root in venv_roots:
                # Installed metadata is readable without invoking uv, whose
                # interpreter discovery writes to its download/build cache.
                for distribution in distributions(
                    name=package_name,
                    path=[str(self._runtime_site_packages(venv_root))],
                ):
                    version = SemVer.parse(distribution.version)
                    if version:
                        return version
            return None
        try:
            uv_abspath = self.INSTALLER_BINARY(no_cache=no_cache).loaded_abspath
            assert uv_abspath
        except (AssertionError, OSError, RuntimeError, ValueError):
            return None
        proc = self.exec(
            bin_name=uv_abspath,
            cmd=[*self._cache_args(no_cache=no_cache), "tool", "list"],
            timeout=timeout,
            quiet=True,
        )
        if proc.returncode != 0:
            return None
        for line in proc.stdout.splitlines():
            line = line.strip()
            if not line or line.startswith("-"):
                continue
            parts = line.split(" v", 1)
            if len(parts) == 2 and parts[0] == package_name:
                return SemVer.parse(parts[1])
        return None

    def _clear_venv_site_packages_pycache(self) -> None:
        if self.install_root is None:
            return
        for site_packages in ((self.install_root / "venv") / "lib").glob(
            "python*/site-packages",
        ):
            for pycache_dir in site_packages.rglob("__pycache__"):
                if pycache_dir.exists():
                    logger.info(
                        "$ %s",
                        format_command(["rm", "-rf", str(pycache_dir)]),
                    )
                shutil.rmtree(pycache_dir, ignore_errors=True)

    def default_search_handler(
        self,
        bin_name: BinName,
        min_version: SemVer | None = None,
        min_release_age: float | None = None,
        timeout: int | None = None,
        **context,
    ) -> list:
        """Resolve the latest published version for an exact PyPI package name via the PyPI JSON API.

        ``uv`` has no ``uv search`` subcommand and ``uv pip`` no longer
        ships ``index versions``, so we hit the same PyPI JSON endpoint
        ``uv pip install`` uses to resolve versions.
        """
        from .binary import Binary

        url = f"https://pypi.org/pypi/{urllib.parse.quote(str(bin_name))}/json"
        try:
            with urllib.request.urlopen(
                url,
                timeout=timeout or self.version_timeout,
            ) as resp:
                data = json.loads(resp.read().decode("utf-8"))
        except urllib.error.HTTPError:
            return []
        info = data.get("info", {}) or {}
        pkg_name = info.get("name", str(bin_name))
        version_str = info.get("version", "")
        summary = info.get("summary", "") or pkg_name
        return [
            Binary(
                name=pkg_name,
                description=f"{version_str} - {summary}".strip(" -"),
                binproviders=[self],
                overrides={self.name: {"install_args": [pkg_name]}},
            ),
        ]

    @remap_kwargs({"packages": "install_args"})
    def default_install_handler(
        self,
        bin_name: str,
        install_args: InstallArgs | None = None,
        postinstall_scripts: bool | None = None,
        min_release_age: float | None = None,
        min_version: SemVer | None = None,
        no_cache: bool = False,
        timeout: int | None = None,
    ) -> str:
        installer_bin = self.INSTALLER_BINARY(no_cache=no_cache).loaded_abspath
        assert installer_bin
        postinstall_scripts = (
            False if postinstall_scripts is None else postinstall_scripts
        )
        min_release_age = 7.0 if min_release_age is None else min_release_age
        install_args = install_args or self.get_install_args(bin_name)
        if min_version:
            install_args = [
                f"{arg}>={min_version}"
                if arg
                and not arg.startswith("-")
                and not any(c in arg for c in ">=<!=~")
                else arg
                for arg in install_args
            ]
        flags = self._pip_flags(
            install_args=install_args,
            postinstall_scripts=postinstall_scripts,
            min_release_age=min_release_age,
        )
        target_root: Path | None = None
        if self.install_root:
            target_root = self._managed_package_root(
                bin_name,
                install_args,
            )
            target_python = target_root / "venv" / "bin" / "python"
            self._ensure_venv(target_root, no_cache=no_cache)
            if min_version:
                tool_names = self.get_package_names(bin_name, install_args)
                installed_versions = [
                    self._version_from_uv_metadata(
                        package_name,
                        timeout=timeout,
                        no_cache=no_cache,
                    )
                    for package_name in tool_names
                ]
                if any(
                    version and version < min_version for version in installed_versions
                ):
                    uninstall_proc = self.exec(
                        bin_name=installer_bin,
                        cmd=[
                            "pip",
                            "uninstall",
                            "--python",
                            str(target_python),
                            *tool_names,
                        ],
                        timeout=timeout,
                    )
                    if (
                        uninstall_proc.returncode != 0
                        and "No packages to uninstall"
                        not in (uninstall_proc.stderr or "")
                    ):
                        self._raise_proc_error("install", tool_names, uninstall_proc)
                    # When install() is revalidating a too-old package, uv's
                    # in-place upgrade can leave importable stale bytecode even
                    # though package metadata has advanced. Remove pyc caches
                    # before reinstalling so the loaded CLI and metadata agree.
                    self._clear_venv_site_packages_pycache()
            cmd = [
                *self._cache_args(no_cache=no_cache),
                "pip",
                "install",
                "--python",
                str(target_python),
                *flags,
                *install_args,
            ]
        else:
            cmd = [
                *self._cache_args(no_cache=no_cache),
                "tool",
                "install",
                "--force",
                *flags,
                *install_args,
            ]
        proc = self.exec(bin_name=installer_bin, cmd=cmd, timeout=timeout)
        if proc.returncode != 0:
            self._raise_proc_error("install", install_args, proc)
        if target_root is not None:
            self._ensure_managed_root_access(target_root)
        return format_subprocess_output(proc.stdout, proc.stderr)

    @remap_kwargs({"packages": "install_args"})
    def default_update_handler(
        self,
        bin_name: str,
        install_args: InstallArgs | None = None,
        postinstall_scripts: bool | None = None,
        min_release_age: float | None = None,
        min_version: SemVer | None = None,
        no_cache: bool = False,
        timeout: int | None = None,
    ) -> str:
        installer_bin = self.INSTALLER_BINARY(no_cache=no_cache).loaded_abspath
        assert installer_bin
        postinstall_scripts = (
            False if postinstall_scripts is None else postinstall_scripts
        )
        min_release_age = 7.0 if min_release_age is None else min_release_age
        install_args = install_args or self.get_install_args(bin_name)
        if min_version:
            install_args = [
                f"{arg}>={min_version}"
                if arg
                and not arg.startswith("-")
                and not any(c in arg for c in ">=<!=~")
                else arg
                for arg in install_args
            ]
        flags = self._pip_flags(
            install_args=install_args,
            postinstall_scripts=postinstall_scripts,
            min_release_age=min_release_age,
        )
        target_root: Path | None = None
        if self.install_root:
            target_root = self._managed_package_root(
                bin_name,
                install_args,
            )
            target_python = target_root / "venv" / "bin" / "python"
            self._ensure_venv(target_root, no_cache=no_cache)
            # Do an explicit uninstall + install cycle instead of
            # ``uv pip install --upgrade --reinstall`` so the venv's
            # site-packages is fully repopulated from scratch (uv's
            # in-place upgrade path can leave stale files otherwise).
            tool_names = self.get_package_names(bin_name, install_args)
            uninstall_proc = self.exec(
                bin_name=installer_bin,
                cmd=[
                    *self._cache_args(no_cache=no_cache),
                    "pip",
                    "uninstall",
                    "--python",
                    str(target_python),
                    *tool_names,
                ],
                timeout=timeout,
            )
            # Treat "no packages to uninstall" as a no-op success.
            if uninstall_proc.returncode != 0 and "No packages to uninstall" not in (
                uninstall_proc.stderr or ""
            ):
                self._raise_proc_error("update", tool_names, uninstall_proc)
            # Existing installs may contain bytecode from older abxpkg
            # versions. Remove it before reinstalling so stale bytecode cannot
            # shadow the freshly installed source.
            self._clear_venv_site_packages_pycache()
            cmd = [
                *self._cache_args(no_cache=no_cache),
                "pip",
                "install",
                "--python",
                str(target_python),
                *flags,
                *install_args,
            ]
        else:
            # ``uv tool install --force`` creates a fresh per-tool venv each
            # time, so there's no stale-compiled-artifact hazard.
            cmd = [
                *self._cache_args(no_cache=no_cache),
                "tool",
                "install",
                "--force",
                *flags,
                *install_args,
            ]
        proc = self.exec(bin_name=installer_bin, cmd=cmd, timeout=timeout)
        if proc.returncode != 0:
            self._raise_proc_error("update", install_args, proc)
        if target_root is not None:
            self._ensure_managed_root_access(target_root)
        return format_subprocess_output(proc.stdout, proc.stderr)

    def write_cached_binary(
        self,
        bin_name: BinName,
        abspath: HostBinPath,
        loaded_version: SemVer,
        loaded_sha256: str,
        resolved_provider_name: str | None = None,
        resolved_provider: BinProvider | None = None,
        cache_kind: str = "binary",
    ):
        result = super().write_cached_binary(
            bin_name,
            abspath,
            loaded_version,
            loaded_sha256,
            resolved_provider_name,
            resolved_provider,
            cache_kind,
        )
        if self.install_root is not None:
            self._ensure_managed_root_access(self.install_root)
            derived_env_path = self.derived_env_path
            if os.geteuid() == 0 and derived_env_path and derived_env_path.exists():
                try:
                    pw_record = self.get_pw_record(self.EUID)
                    os.chown(derived_env_path, self.EUID, pw_record.pw_gid)
                except (PermissionError, OSError, KeyError):
                    pass
        return result

    @remap_kwargs({"packages": "install_args"})
    def default_uninstall_handler(
        self,
        bin_name: str,
        install_args: InstallArgs | None = None,
        postinstall_scripts: bool | None = None,
        min_release_age: float | None = None,
        min_version: SemVer | None = None,
        no_cache: bool = False,
        timeout: int | None = None,
    ) -> bool:
        installer_bin = self.INSTALLER_BINARY(no_cache=no_cache).loaded_abspath
        assert installer_bin
        install_args = install_args or self.get_install_args(bin_name)
        # Strip version pins / extras from package specs so both
        # ``uv pip uninstall`` and ``uv tool uninstall`` get bare names.
        tool_names = [
            arg.split("[", 1)[0].split("=", 1)[0].split(">", 1)[0].split("<", 1)[0]
            for arg in install_args
            if arg and not arg.startswith("-")
        ] or [bin_name]
        if self.install_root:
            target_python = (
                self._managed_package_root(bin_name, install_args)
                / "venv"
                / "bin"
                / "python"
            )
            cmd = [
                *self._cache_args(no_cache=no_cache),
                "pip",
                "uninstall",
                "--python",
                str(target_python),
                *tool_names,
            ]
        else:
            cmd = ["tool", "uninstall", *tool_names]
        proc = self.exec(bin_name=installer_bin, cmd=cmd, timeout=timeout)
        if proc.returncode != 0:
            self._raise_proc_error("uninstall", tool_names, proc)
        return True

    def default_abspath_handler(
        self,
        bin_name: BinName | HostBinPath,
        no_cache: bool = False,
        **context,
    ) -> HostBinPath | None:
        try:
            abspath = super().default_abspath_handler(bin_name, **context)
            if abspath and self._managed_path_matches_runtime(
                str(bin_name),
                Path(abspath),
            ):
                return TypeAdapter(HostBinPath).validate_python(abspath)
        except (OSError, RuntimeError, ValueError) as err:
            logger.debug("uv provider host-path probe failed for %s: %s", bin_name, err)
        # Managed libraries may have no console entrypoint.
        if self.install_root:
            tool_name = self.get_package_names(str(bin_name))[0]

            for venv_root in self._managed_venv_roots(str(bin_name)):
                location = self._runtime_site_packages(venv_root)
                if (
                    next(
                        iter(distributions(name=tool_name, path=[str(location)])),
                        None,
                    )
                    is None
                ):
                    continue
                candidate = venv_root / "bin" / str(bin_name)
                if candidate.exists():
                    return TypeAdapter(HostBinPath).validate_python(candidate)
                module_names = [
                    name.replace("-", "_").replace(".", "_")
                    for name in self.get_library_names(str(bin_name))
                ]
                for module_name in module_names:
                    for module_candidate in (
                        location / module_name / "__init__.py",
                        location / f"{module_name}.py",
                    ):
                        if module_candidate.exists():
                            return TypeAdapter(HostBinPath).validate_python(
                                module_candidate,
                            )
        else:
            tool_name = self.get_package_names(str(bin_name))[0]
            candidate = self.tool_dir / tool_name / "bin" / str(bin_name)
            if candidate.exists():
                return TypeAdapter(HostBinPath).validate_python(candidate)
        return None

    def default_docs_url_handler(
        self,
        bin_name: BinName,
        **context,
    ) -> str | None:
        package = self.get_package_names(str(bin_name))[0]
        return f"https://pypi.org/project/{package}"

    def default_version_handler(
        self,
        bin_name: BinName,
        abspath: HostBinPath | None = None,
        timeout: int | None = None,
        no_cache: bool = False,
        **context,
    ) -> SemVer | None:
        tool_name = self.get_package_names(str(bin_name))[0]
        metadata_version = self._version_from_uv_metadata(
            tool_name,
            timeout=timeout,
            no_cache=no_cache,
            abspath=abspath,
        )
        if metadata_version:
            return metadata_version
        if abspath is None or os.access(abspath, os.X_OK):
            try:
                version = self._version_from_exec(
                    bin_name,
                    abspath=abspath,
                    timeout=timeout,
                )
                if version:
                    return version
            except ValueError:
                pass
        return None


if __name__ == "__main__":
    # Usage:
    # ./binprovider_uv.py load black
    # ./binprovider_uv.py install black
    result = uv = UvProvider()
    func = None
    if len(sys.argv) > 1:
        result = func = getattr(uv, sys.argv[1])
    if len(sys.argv) > 2 and callable(func):
        result = func(sys.argv[2])
    print(result)