Source code for taimoe.platform.routes

"""Optional FastAPI routes for runtime discovery.

Exposes three endpoints a managed runtime advertises so the Taimoe Platform
can probe it:

- ``GET /.well-known/taimoe-runtime.json`` — RFC 8615 well-known descriptor.
  Carries the runtime manifest and optionally answers a challenge so the
  Platform can verify the runtime is running an actual Taimoe SDK.
- ``GET /agents`` — list of agents currently registered at this runtime.
- ``GET /health`` — runtime health status (uptime, version, etc).

Per RFC 8615 the well-known endpoint must live at the URL root. To make
that hard to get wrong we return two routers — one anchored at root, one
that can be mounted at any prefix — and the helper wires both into a
FastAPI app for you.
"""

from __future__ import annotations

import hashlib
from typing import TYPE_CHECKING, Any

from ._well_known import CHALLENGE_PROTOCOL_VERSION, WELL_KNOWN_PATH

if TYPE_CHECKING:
    from fastapi import APIRouter, FastAPI

    from .registry import TaimoeRegistry




def compute_challenge_response(challenge: str) -> str:
    """Hash ``challenge`` with the protocol version for the well-known reply.

    Anti-spoofing only — see ``_well_known.CHALLENGE_PROTOCOL_VERSION`` for
    why this isn't authentication and how to evolve it.
    """
    return hashlib.sha256(
        f"{challenge}{CHALLENGE_PROTOCOL_VERSION}".encode()
    ).hexdigest()





def create_well_known_router(registry: "TaimoeRegistry") -> "APIRouter":
    """Router carrying only ``GET /.well-known/taimoe-runtime.json``.

    Must be mounted at the FastAPI app root with no prefix — RFC 8615
    requires the well-known URI to be served from origin root.
    """
    APIRouter, _ = _require_fastapi()

    router = APIRouter()

    @router.get(WELL_KNOWN_PATH, response_model=None)
    def well_known(challenge: str | None = None) -> dict[str, Any]:
        # `dict[str, Any]` instead of returning the Pydantic model directly
        # because we conditionally add `challenge_response`. The manifest
        # itself stays schema-checked via model_dump.
        data = registry.runtime_manifest().model_dump(mode="json")
        if challenge:
            data["challenge_response"] = compute_challenge_response(challenge)
        return data

    return router





def create_discovery_router(registry: "TaimoeRegistry") -> "APIRouter":
    """Router carrying ``/agents`` and ``/health``.

    Can be mounted at any prefix the host application prefers
    (e.g. ``prefix="/v1"``).
    """
    APIRouter, _ = _require_fastapi()

    # Lazy import the response models so callers without ``fastapi`` can
    # still import this module (we already lazy-import APIRouter).
    from .types.manifest import AgentsManifest, HealthStatus

    router = APIRouter()

    @router.get("/agents", response_model=AgentsManifest)
    def agents() -> AgentsManifest:
        return registry.agents_manifest()

    @router.get("/health", response_model=HealthStatus)
    def health() -> HealthStatus:
        return registry.health()

    return router





def install_routes(registry: "TaimoeRegistry", app: "FastAPI") -> None:
    """Wire both routers into a FastAPI app with the correct mounts.

    The well-known router goes at the root; the discovery router is
    mounted bare (``/agents`` / ``/health``). Callers that need a
    different prefix for discovery can use :func:`create_discovery_router`
    directly.
    """
    app.include_router(create_well_known_router(registry))
    app.include_router(create_discovery_router(registry))





def create_fastapi_router(registry: "TaimoeRegistry") -> "APIRouter":
    """Backwards-compatible single-router entry point.

    Returns one combined router that includes both the well-known and
    discovery routes. Convenient for simple deployments, but means the
    caller must mount it at the app root or the well-known path will
    break RFC 8615. Prefer :func:`install_routes` for new code.
    """
    APIRouter, _ = _require_fastapi()
    combined = APIRouter()
    combined.include_router(create_well_known_router(registry))
    combined.include_router(create_discovery_router(registry))
    return combined



def _require_fastapi() -> tuple[type, type]:
    """Import FastAPI lazily so non-proxy users don't pay the dep cost."""
    try:
        from fastapi import APIRouter, FastAPI
    except ImportError as exc:  # pragma: no cover - import guard
        raise RuntimeError(
            "Install taimoe-platform[proxy] to use FastAPI discovery routes"
        ) from exc
    return APIRouter, FastAPI