hermes-relay/plugin/hermes_relay_bootstrap/_patch.py

"""Import-hook plumbing + Application monkey-patch.

When `aiohttp.web` is imported anywhere in the running interpreter, our
`_AioHttpWebFinder` intercepts the import, lets the real loader finish,
then swaps `web.Application` for a `_PatchedApplication` subclass.

The subclass overrides `__setitem__` so we can detect the moment hermes-agent's
`APIServerAdapter` does `self._app["api_server_adapter"] = self` — that's the
single line in the upstream `connect()` method that gives us a reference to
the adapter while the app is still being built (router not yet frozen).

At that point, we call `_register_routes()` to bind any missing compatibility
handlers to the same router the gateway is in the middle of populating. Route
registration is granular: native upstream routes win per method/path, and the
bootstrap only fills gaps that still do not exist. We also install the
slash-command middleware that intercepts
``/v1/chat/completions`` and ``/v1/runs`` to handle gateway commands like
``/help``, ``/commands``, ``/profile``, and ``/provider`` without forwarding
them to the LLM.  The gateway then continues with its own route registrations
and starts the server normally.

Failure modes are silent-but-logged: if anything in this chain breaks
(unexpected upstream refactor, aiohttp version incompatibility, missing
SessionDB methods, etc.), we log a warning and let the gateway start without
the injected routes. The Android client's capability detection will fall back
to the standard upstream endpoints automatically.
"""

from __future__ import annotations

import logging
import sys

logger = logging.getLogger(__name__)

# Compatibility paths that `_handlers.register_routes()` can fill when native
# upstream support is absent. Registration is method-aware in `_handlers.py`;
# this set is only used for logging/cheap diagnostics in the import hook.
_COMPATIBILITY_PATHS: frozenset[str] = frozenset({
    "/api/sessions",
    "/api/sessions/search",
    "/api/sessions/{session_id}",
    "/api/sessions/{session_id}/messages",
    "/api/sessions/{session_id}/fork",
    "/api/memory",
    "/api/skills",
    "/api/skills/{name}",
    "/api/config",
    "/api/available-models",
})


class _AioHttpWebFinder:
    """sys.meta_path finder that wraps the loader for `aiohttp.web`.

    Removes itself from `sys.meta_path` after firing once, so subsequent
    imports of unrelated modules don't pay the find-spec cost.
    """

    def find_spec(self, fullname, path, target=None):
        if fullname != "aiohttp.web":
            return None

        # Remove ourselves so this only runs on the first import.
        try:
            sys.meta_path.remove(self)
        except ValueError:
            pass

        # Resolve the real spec via the remaining finders. We can't recurse
        # back into our own find_spec because we just removed ourselves.
        for finder in sys.meta_path:
            if not hasattr(finder, "find_spec"):
                continue
            spec = finder.find_spec(fullname, path, target)
            if spec is None:
                continue

            original_loader = spec.loader
            spec.loader = _PatchingLoader(original_loader)
            return spec

        # No finder could resolve aiohttp.web — fall through, normal import
        # error will surface.
        return None


class _PatchingLoader:
    """Loader wrapper that runs `_apply_patch()` after the module is exec'd."""

    def __init__(self, wrapped):
        self._wrapped = wrapped

    def create_module(self, spec):
        if hasattr(self._wrapped, "create_module"):
            return self._wrapped.create_module(spec)
        return None

    def exec_module(self, module):
        self._wrapped.exec_module(module)
        try:
            _apply_patch(module)
        except Exception as exc:
            logger.warning(
                "hermes_relay_bootstrap: failed to install Application patch: %s",
                exc,
            )


def install_finder() -> None:
    """Insert the import-hook finder at the front of `sys.meta_path`."""
    finder = _AioHttpWebFinder()
    sys.meta_path.insert(0, finder)


def _apply_patch(web_module) -> None:
    """Replace `web_module.Application` with `_PatchedApplication`.

    No-op if `web_module` lacks the expected `Application` attribute (e.g.
    aiohttp version skew or stripped-down build).
    """
    original_application = getattr(web_module, "Application", None)
    if original_application is None:
        logger.warning(
            "hermes_relay_bootstrap: aiohttp.web has no Application attribute "
            "— version mismatch? skipping injection."
        )
        return

    if getattr(original_application, "_hermes_relay_patched", False):
        return  # Idempotent — already patched in a previous import.

    class _PatchedApplication(original_application):
        """Subclass of aiohttp.web.Application with adapter-detection hook.

        Hermes-agent's `APIServerAdapter.connect()` does
        `self._app["api_server_adapter"] = self` immediately after building
        the application. We catch that and use the adapter reference to
        register our extra routes on `self.router` while it's still mutable.
        """

        _hermes_relay_patched = True

        def __setitem__(self, key, value):
            super().__setitem__(key, value)
            if key != "api_server_adapter":
                return
            try:
                _maybe_register_routes(self, value)
            except Exception as exc:  # pragma: no cover - defensive logging
                logger.warning(
                    "hermes_relay_bootstrap: route injection failed: %s",
                    exc,
                )
            try:
                _maybe_install_command_middleware(self, value)
            except Exception as exc:  # pragma: no cover - defensive logging
                logger.warning(
                    "hermes_relay_bootstrap: command middleware "
                    "installation failed: %s",
                    exc,
                )

    web_module.Application = _PatchedApplication


def _maybe_register_routes(app, adapter) -> None:
    """Register missing compatibility routes without shadowing upstream.

    Current Hermes core work is split across focused upstream PRs rather than
    one broad session-management branch. For example, native `/api/sessions/*`
    may exist while `/api/config`, `/api/skills`, or `/api/memory` still need
    the relay bootstrap. Do not use all-or-nothing route detection here.
    """
    existing_paths: set[str] = set()
    try:
        for resource in app.router.resources():
            canonical = getattr(resource, "canonical", None)
            if canonical:
                existing_paths.add(canonical)
    except Exception:
        # If router introspection fails, err on the side of NOT injecting —
        # double-registration would crash the whole gateway startup.
        logger.warning(
            "hermes_relay_bootstrap: cannot inspect existing routes; "
            "skipping injection (gateway will run with whatever routes "
            "it natively provides)."
        )
        return

    if existing_paths & _COMPATIBILITY_PATHS:
        logger.info(
            "hermes_relay_bootstrap: detected native /api/* routes; "
            "will inject only missing compatibility routes."
        )

    # Defer the heavy import until we're actually going to register. This
    # keeps the bootstrap cheap for `python -c "1+1"` style invocations
    # that never use aiohttp meaningfully.
    try:
        from . import _handlers
    except Exception as exc:
        logger.warning(
            "hermes_relay_bootstrap: cannot import _handlers (%s); "
            "skipping injection.",
            exc,
        )
        return

    try:
        injected = _handlers.register_routes(app, adapter)
    except Exception as exc:
        logger.warning(
            "hermes_relay_bootstrap: register_routes raised %s; "
            "the gateway will run without injected compatibility API.",
            exc,
        )
        return

    if injected:
        logger.info(
            "hermes_relay_bootstrap: injected %d missing /api/* "
            "compatibility routes onto upstream hermes-agent gateway",
            injected,
        )
    else:
        logger.info(
            "hermes_relay_bootstrap: all compatibility /api/* routes are "
            "already native; no route injection needed."
        )


def _maybe_install_command_middleware(app, adapter) -> None:
    """Install the slash-command middleware on the app if not already present.

    The middleware intercepts ``/v1/chat/completions`` and ``/v1/runs``
    to handle gateway commands (``/help``, ``/commands``, ``/profile``,
    ``/provider``) and return decline notices for stateful commands,
    preventing the LLM from hallucinating responses for them.

    If the upstream ``api_server_slash`` module already exists (meaning
    the PR has been merged or the fork is deployed), the middleware is
    skipped — the native handler handles interception.
    """
    try:
        from . import _command_middleware
    except Exception as exc:
        logger.warning(
            "hermes_relay_bootstrap: cannot import _command_middleware (%s); "
            "skipping middleware installation.",
            exc,
        )
        return

    try:
        _command_middleware.maybe_install_middleware(app, adapter)
    except Exception as exc:
        logger.warning(
            "hermes_relay_bootstrap: maybe_install_middleware raised %s; "
            "slash commands will fall through to the LLM.",
            exc,
        )