hermes-chatto-plugin/vendor_path.py

"""Locate the vendored dependency tree matching this interpreter's platform.

``vendor/`` is split into a platform-independent part and one directory per
supported platform, because three of the vendored packages (pyqwest, protobuf,
protobuf-py-ext) ship compiled extension modules::

    vendor/
      common/                    pure-Python packages, shared by all platforms
      platform/
        linux-x86_64/            compiled extensions for that platform only
        linux-aarch64/
        macos-arm64/

The two directories enter ``sys.path`` differently, on purpose:

* ``platform/<tag>`` is **prepended**. It holds the compiled extensions that
  must match the vendored chattolib (pyqwest, protobuf, protobuf-py-ext);
  falling back to a host copy of a different version would break the generated
  protobuf code.
* ``common`` is **appended**. Most of it (httpx, anyio, httpcore, h11, idna,
  certifi, typing_extensions) is also shipped by the Hermes agent itself, and
  a plugin has no business shadowing the host's pinned versions — Hermes pins
  certifi exactly, and we would otherwise substitute our own CA bundle for the
  whole process. The host's copies win; ours only serve as a fallback, which
  keeps out-of-process use (the cron sender) working on its own.

Run ``./vendor_chattolib.sh`` to (re)build the tree.
"""

from __future__ import annotations

import platform
import sys
from pathlib import Path
from typing import List

VENDOR_DIR = Path(__file__).parent / "vendor"

# Directory name per (system, machine) — keep in sync with PLATFORMS in
# vendor_chattolib.sh.
_MACHINE_ALIASES = {
    "x86_64": "x86_64",
    "amd64": "x86_64",
    "aarch64": "aarch64",
    "arm64": "aarch64",
}


def _is_musl() -> bool:
    """True when running against musl libc (Alpine) rather than glibc."""
    try:
        libc, _version = platform.libc_ver()
    except OSError:
        return False
    # glibc reports "glibc"; musl reports an empty string.
    return libc != "glibc"


def platform_tag() -> str:
    """Return the vendor/platform/ subdirectory name for this interpreter."""
    system = platform.system().lower()
    machine = _MACHINE_ALIASES.get(platform.machine().lower(), platform.machine().lower())

    if system == "linux":
        return f"linux-{machine}-musl" if _is_musl() else f"linux-{machine}"
    if system == "darwin":
        return f"macos-{'arm64' if machine == 'aarch64' else machine}"
    if system == "windows":
        return f"windows-{'amd64' if machine == 'x86_64' else machine}"
    return f"{system}-{machine}"


def available_platforms() -> List[str]:
    """Platform directories that were actually vendored into this checkout."""
    platform_root = VENDOR_DIR / "platform"
    if not platform_root.is_dir():
        return []
    return sorted(p.name for p in platform_root.iterdir() if p.is_dir())


def setup_vendor_path() -> List[str]:
    """Put the vendored packages for this platform on ``sys.path``.

    The compiled extensions are prepended, the pure-Python packages appended;
    see the module docstring for why. Returns the directories that were added.
    Raises ImportError when this platform was not vendored — a clear message
    beats the dlopen error you would otherwise get from a foreign-architecture
    .so file.
    """
    tag = platform_tag()
    platform_dir = VENDOR_DIR / "platform" / tag
    common_dir = VENDOR_DIR / "common"

    # Legacy flat layout (single platform, everything directly in vendor/).
    if not platform_dir.is_dir() and not common_dir.is_dir():
        if (VENDOR_DIR / "chattolib").is_dir():
            prepend, append = [VENDOR_DIR], []
        else:
            raise ImportError(
                f"Chatto: no vendored dependencies found in {VENDOR_DIR}. "
                f"Run ./vendor_chattolib.sh to build them."
            )
    elif not platform_dir.is_dir():
        raise ImportError(
            f"Chatto: no vendored dependencies for platform '{tag}'. "
            f"Available: {', '.join(available_platforms()) or 'none'}. "
            f"Run 'PLATFORMS={tag} ./vendor_chattolib.sh' to add it."
        )
    else:
        prepend, append = [platform_dir], [common_dir]

    added = []
    for path in prepend:
        entry = str(path)
        if entry not in sys.path:
            sys.path.insert(0, entry)
            added.append(entry)
    for path in append:
        entry = str(path)
        if entry not in sys.path:
            sys.path.append(entry)
            added.append(entry)
    return added