thoxmesh_node/server.py

"""OpenAI-compatible HTTP surface for a mesh node.

Only the subset ThoxRoute and the ThoxOS apps actually call is implemented:
``/v1/chat/completions``, ``/v1/models`` and health. Implementing the surface
narrowly and honestly is deliberate — a node that advertises streaming or
embeddings it cannot serve would be selected by the router and then fail, which
is worse for the mesh than not advertising them.

Every request is timed and recorded into the ``TelemetryRecorder``, because that
recording *is* how the node earns or loses its place in the router's ranking.
"""

from __future__ import annotations

import logging
import time
import uuid
from typing import Any, Literal

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from pydantic import BaseModel, Field, field_validator

from .backends import Backend
from .errors import BackendError
from .telemetry import TelemetryRecorder

logger = logging.getLogger("thox.server")


class ChatMessage(BaseModel):
    """One message in an OpenAI-style chat request."""

    role: Literal["system", "user", "assistant", "tool"]
    content: str = ""

    @field_validator("content", mode="before")
    @classmethod
    def _coerce_content(cls, value: Any) -> str:
        """Accept the content-parts array form and flatten it to text.

        Newer OpenAI clients send ``[{"type": "text", "text": "..."}]``. A node
        that 422s on that shape looks broken to a caller that is, by the spec,
        well-behaved.
        """
        if value is None:
            return ""
        if isinstance(value, str):
            return value
        if isinstance(value, list):
            parts = [
                str(part.get("text", ""))
                for part in value
                if isinstance(part, dict) and part.get("type") in (None, "text")
            ]
            return "".join(parts)
        return str(value)


class ChatCompletionRequest(BaseModel):
    """Request body for ``/v1/chat/completions``."""

    model: str | None = None
    messages: list[ChatMessage] = Field(min_length=1)
    max_tokens: int | None = Field(default=None, ge=1)
    temperature: float = Field(default=0.7, ge=0.0, le=2.0)
    stop: list[str] | str | None = None
    stream: bool = False


def create_app(
    *,
    backend: Backend,
    recorder: TelemetryRecorder,
    model_id: str,
    context_window: int,
    max_tokens_cap: int,
    node_state: dict[str, Any] | None = None,
) -> FastAPI:
    """Build the ASGI app.

    ``node_state`` is a live dict owned by the agent; the health endpoint reads
    it so operators (and the controller's health loop) can see registration
    state without a second channel.
    """
    app = FastAPI(title="THOX Mesh Node", version="1.0.0")
    state = node_state if node_state is not None else {}

    @app.get("/health")
    def health() -> dict[str, Any]:
        """Liveness plus current mesh registration and telemetry."""
        return {
            "status": "ok",
            "model_id": model_id,
            "context_window": context_window,
            "mesh": {
                "transport": state.get("transport"),
                "endpoint_id": state.get("endpoint_id"),
                "registered": bool(state.get("endpoint_id")),
                "public_url": state.get("public_url"),
                "last_heartbeat_at": state.get("last_heartbeat_at"),
                "last_error": state.get("last_error"),
            },
            "telemetry": recorder.snapshot(),
        }

    @app.get("/v1/models")
    def list_models() -> dict[str, Any]:
        """Model list in OpenAI's shape, so generic clients can discover us."""
        return {
            "object": "list",
            "data": [
                {
                    "id": model_id,
                    "object": "model",
                    "owned_by": "thox",
                    "created": int(state.get("started_at", time.time())),
                    "context_window": context_window,
                }
            ],
        }

    @app.post("/v1/chat/completions")
    def chat_completions(body: ChatCompletionRequest) -> JSONResponse:
        """Generate one completion and record its outcome as telemetry."""
        if body.stream:
            # Refused explicitly rather than silently returning a non-stream body,
            # which would leave a streaming client hanging on a parse it can never finish.
            raise HTTPException(
                status_code=400,
                detail="streaming is not supported by this node; retry with stream=false",
            )

        requested = body.max_tokens or max_tokens_cap
        max_tokens = max(1, min(requested, max_tokens_cap))
        stop = [body.stop] if isinstance(body.stop, str) else body.stop

        recorder.request_started()
        started = time.perf_counter()
        try:
            completion = backend.generate(
                [m.model_dump() for m in body.messages],
                max_tokens=max_tokens,
                temperature=body.temperature,
                stop=stop,
            )
        except BackendError as exc:
            recorder.request_finished(latency_ms=(time.perf_counter() - started) * 1000, ok=False)
            logger.error("generation failed: %s", exc)
            raise HTTPException(status_code=503, detail=str(exc)) from exc
        except Exception as exc:  # noqa: BLE001 - never leak a 500 without telemetry
            recorder.request_finished(latency_ms=(time.perf_counter() - started) * 1000, ok=False)
            logger.exception("unexpected generation error")
            raise HTTPException(status_code=500, detail="internal generation error") from exc

        latency_ms = (time.perf_counter() - started) * 1000
        recorder.request_finished(
            latency_ms=latency_ms,
            ok=True,
            completion_tokens=completion.completion_tokens,
        )
        return JSONResponse(
            {
                "id": f"chatcmpl-{uuid.uuid4().hex[:24]}",
                "object": "chat.completion",
                "created": int(time.time()),
                "model": body.model or model_id,
                "choices": [
                    {
                        "index": 0,
                        "message": {"role": "assistant", "content": completion.text},
                        "finish_reason": "stop",
                    }
                ],
                "usage": {
                    "prompt_tokens": completion.prompt_tokens,
                    "completion_tokens": completion.completion_tokens,
                    "total_tokens": completion.total_tokens,
                },
                "thox_node": {
                    "latency_ms": round(latency_ms, 1),
                    "endpoint_id": state.get("endpoint_id"),
                    "node_kind": state.get("node_kind"),
                },
            }
        )

    @app.get("/")
    def root(request: Request) -> dict[str, Any]:
        """Human-facing landing payload; also what a Space's health probe hits."""
        return {
            "service": "THOX Mesh Node",
            "model_id": model_id,
            "openai_base_url": str(request.base_url).rstrip("/") + "/v1",
            "endpoints": ["/v1/chat/completions", "/v1/models", "/health"],
            "registered": bool(state.get("endpoint_id")),
        }

    return app