omnigent-ai--omnigent/omnigent/server/performance_metrics.py

"""Lightweight server performance metrics for the FastAPI app."""

from __future__ import annotations

import asyncio
import contextvars
import logging
import os
import sys
import time

try:
    import resource  # POSIX-only; absent on Windows.
except ImportError:  # pragma: no cover - exercised only on Windows
    resource = None  # type: ignore[assignment]
from collections import deque
from dataclasses import dataclass
from threading import Lock
from typing import Protocol

from opentelemetry import metrics as otel_metrics
from opentelemetry.util.types import Attributes
from uvicorn.logging import AccessFormatter

_DEFAULT_WINDOWS_SECONDS = (1.0, 10.0, 30.0)
_BYTES_PER_MIB = 1024 * 1024
_OTEL_METER_NAME = "omnigent.server.performance"
_REQUEST_DURATION_CONTEXT: contextvars.ContextVar[float | None] = contextvars.ContextVar(
    "omnigent_request_duration_seconds",
    default=None,
)
_REQUEST_ID_CONTEXT: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "omnigent_request_id",
    default=None,
)
_REQUEST_USER_AGENT_CONTEXT: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "omnigent_request_user_agent",
    default=None,
)
_REQUEST_SESSION_ID_CONTEXT: contextvars.ContextVar[str | None] = contextvars.ContextVar(
    "omnigent_request_session_id",
    default=None,
)


class FloatSampler(Protocol):
    """
    Protocol for dependency-injected float samplers.
    """

    def __call__(self) -> float:
        """
        Return the sampled float value.

        :returns: A sampled float value.
        """
        ...


class RssSampler(Protocol):
    """
    Protocol for dependency-injected resident-memory samplers.
    """

    def __call__(self) -> int:
        """
        Return resident memory in bytes.

        :returns: Resident memory in bytes.
        """
        ...


class LoadAverageSampler(Protocol):
    """
    Protocol for dependency-injected load-average samplers.
    """

    def __call__(self) -> SystemLoadAverage | None:
        """
        Return system load averages.

        :returns: System load averages, or ``None`` when unavailable.
        """
        ...


class GaugeInstrument(Protocol):
    """
    Protocol for OpenTelemetry synchronous gauge instruments.
    """

    def set(self, amount: int | float, attributes: Attributes = None) -> None:
        """
        Set the gauge to a point-in-time value.

        :param amount: Gauge value to record.
        :param attributes: Optional OpenTelemetry metric attributes.
        """
        ...


class CounterInstrument(Protocol):
    """
    Protocol for OpenTelemetry counter instruments.
    """

    def add(self, amount: int | float, attributes: Attributes = None) -> None:
        """
        Add a non-negative value to the counter.

        :param amount: Counter delta to record.
        :param attributes: Optional OpenTelemetry metric attributes.
        """
        ...


class HistogramInstrument(Protocol):
    """
    Protocol for OpenTelemetry histogram instruments.
    """

    def record(self, amount: int | float, attributes: Attributes = None) -> None:
        """
        Record one histogram sample.

        :param amount: Histogram sample value.
        :param attributes: Optional OpenTelemetry metric attributes.
        """
        ...


class MeterLike(Protocol):
    """
    Protocol for the OpenTelemetry meter methods used by this module.
    """

    def create_counter(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> CounterInstrument:
        """
        Create a monotonic counter instrument.

        :param name: Metric name, e.g.
            ``"omnigent.server.http.requests.started"``.
        :param unit: UCUM unit string, e.g. ``"{request}"``.
        :param description: Human-readable metric description.
        :returns: OpenTelemetry counter instrument.
        """
        ...

    def create_gauge(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> GaugeInstrument:
        """
        Create a synchronous gauge instrument.

        :param name: Metric name, e.g.
            ``"omnigent.server.http.requests.in_flight"``.
        :param unit: UCUM unit string, e.g. ``"{request}"``.
        :param description: Human-readable metric description.
        :returns: OpenTelemetry gauge instrument.
        """
        ...

    def create_histogram(
        self,
        name: str,
        unit: str = "",
        description: str = "",
    ) -> HistogramInstrument:
        """
        Create a histogram instrument.

        :param name: Metric name, e.g.
            ``"omnigent.server.http.request.duration"``.
        :param unit: UCUM unit string, e.g. ``"s"``.
        :param description: Human-readable metric description.
        :returns: OpenTelemetry histogram instrument.
        """
        ...


def _sanitize_access_log_value(value: str) -> str:
    """
    Make a client-controlled string safe to embed in an access-log line.

    The ``User-Agent`` header and the session ID parsed from the URL path
    are both attacker-controlled. Replaces control characters (CR, LF,
    NUL, ANSI escape sequences, ...) and the double-quote delimiter with
    ``?`` so a crafted value cannot forge additional log records, corrupt
    the quoted ``ua`` field, or smuggle terminal escapes into log viewers
    (CWE-117).

    :param value: Raw, untrusted string to embed in the access log.
    :returns: Value containing only printable, non-quote characters.
    """
    return "".join(char if char.isprintable() and char != '"' else "?" for char in value)


class RequestDurationAccessFormatter(AccessFormatter):
    """
    Uvicorn access formatter that appends Omnigent request duration.

    Uvicorn owns the actual access log emission. The application
    middleware records request duration in a context variable, and
    Uvicorn formats the access record later in the same request task.
    The suffix intentionally stays terse, e.g. ``"2.5ms"``, so the
    line keeps Uvicorn's standard access-log shape.
    """

    _MAX_USER_AGENT_LENGTH = 80

    def formatMessage(self, record: logging.LogRecord) -> str:
        """
        Format one Uvicorn access record with request context.

        Appends duration, request ID, User-Agent, and session ID when
        available.  Example output::

            GET /v1/sessions/conv_abc HTTP/1.1 200 2.5ms rid=a1b2c3d4 ua="httpx/0.27" sid=conv_abc

        :param record: Logging record emitted by Uvicorn's access
            logger.
        :returns: Enriched access message.
        """
        message = super().formatMessage(record)
        parts: list[str] = [message]
        try:
            duration_seconds = _REQUEST_DURATION_CONTEXT.get()
            if duration_seconds is not None:
                parts.append(f"{duration_seconds * 1000.0:.1f}ms")

            request_id = _REQUEST_ID_CONTEXT.get()
            if request_id is not None:
                parts.append(f"rid={request_id}")

            user_agent = _REQUEST_USER_AGENT_CONTEXT.get()
            if user_agent is not None:
                truncated = user_agent[: self._MAX_USER_AGENT_LENGTH]
                parts.append(f'ua="{_sanitize_access_log_value(truncated)}"')

            session_id = _REQUEST_SESSION_ID_CONTEXT.get()
            if session_id is not None:
                parts.append(f"sid={_sanitize_access_log_value(session_id)}")
        finally:
            _REQUEST_DURATION_CONTEXT.set(None)
            _REQUEST_ID_CONTEXT.set(None)
            _REQUEST_USER_AGENT_CONTEXT.set(None)
            _REQUEST_SESSION_ID_CONTEXT.set(None)
        return " ".join(parts)


@dataclass(frozen=True)
class SystemLoadAverage:
    """
    System load averages for standard Unix windows.

    :param one_minute: One-minute system load average.
    :param five_minutes: Five-minute system load average.
    :param fifteen_minutes: Fifteen-minute system load average.
    """

    one_minute: float
    five_minutes: float
    fifteen_minutes: float


@dataclass
class _CounterState:
    """
    Last emitted cumulative counter values for OTEL delta publishing.

    :param total_started: Last ``total_started`` snapshot value.
    :param total_completed: Last ``total_completed`` snapshot value.
    :param total_failed: Last ``total_failed`` snapshot value.
    """

    total_started: int = 0
    total_completed: int = 0
    total_failed: int = 0


@dataclass(frozen=True)
class _RequestOtelInstruments:
    """
    OpenTelemetry instruments for HTTP and WebSocket request state.

    :param started: Monotonic counter for started HTTP requests.
    :param completed: Monotonic counter for completed HTTP requests.
    :param failed: Monotonic counter for failed HTTP requests.
    :param duration: Histogram for completed HTTP request durations.
    :param in_flight: Gauge for currently processing HTTP requests.
    :param last_1s: Gauge for requests started in the last second.
    :param last_10s: Gauge for requests started in the last ten
        seconds.
    :param last_30s: Gauge for requests started in the last thirty
        seconds.
    :param active_websockets: Gauge for currently open accepted
        WebSocket connections.
    """

    started: CounterInstrument
    completed: CounterInstrument
    failed: CounterInstrument
    duration: HistogramInstrument
    in_flight: GaugeInstrument
    last_1s: GaugeInstrument
    last_10s: GaugeInstrument
    last_30s: GaugeInstrument
    active_websockets: GaugeInstrument


@dataclass(frozen=True)
class _ProcessingOtelInstruments:
    """
    OpenTelemetry gauges for HTTP request processing durations.

    :param avg_ms: Gauge for all-time average request processing
        duration in milliseconds.
    :param max_ms: Gauge for all-time maximum request processing
        duration in milliseconds.
    :param avg_1s_ms: Gauge for one-second rolling average request
        processing duration in milliseconds.
    :param avg_10s_ms: Gauge for ten-second rolling average request
        processing duration in milliseconds.
    :param avg_30s_ms: Gauge for thirty-second rolling average request
        processing duration in milliseconds.
    """

    avg_ms: GaugeInstrument
    max_ms: GaugeInstrument
    avg_1s_ms: GaugeInstrument
    avg_10s_ms: GaugeInstrument
    avg_30s_ms: GaugeInstrument


@dataclass(frozen=True)
class _ResourceOtelInstruments:
    """
    OpenTelemetry gauges for process and system resource state.

    :param cpu_percent: Gauge for process CPU percent.
    :param load_1m: Gauge for one-minute system load average.
    :param load_5m: Gauge for five-minute system load average.
    :param load_15m: Gauge for fifteen-minute system load average.
    :param rss_bytes: Gauge for resident memory bytes.
    """

    cpu_percent: GaugeInstrument
    load_1m: GaugeInstrument
    load_5m: GaugeInstrument
    load_15m: GaugeInstrument
    rss_bytes: GaugeInstrument


@dataclass(frozen=True)
class CompletedRequestTiming:
    """
    Timing for one completed HTTP request.

    :param completed_at: Monotonic timestamp when the request left the
        server.
    :param duration_seconds: Request processing duration in seconds.
    """

    completed_at: float
    duration_seconds: float


@dataclass(frozen=True)
class RequestMetricValues:
    """
    Request metrics copied while holding the tracker lock.

    :param in_flight: HTTP requests currently executing.
    :param total_started: Total HTTP requests started since process
        start.
    :param total_completed: Total HTTP requests completed since
        process start.
    :param total_failed: Total HTTP requests failed since process
        start.
    :param requests_last_1s: Requests started in the last second.
    :param requests_last_10s: Requests started in the last ten
        seconds.
    :param requests_last_30s: Requests started in the last thirty
        seconds.
    :param active_websockets: Accepted WebSocket connections currently
        open in the ASGI app.
    :param request_processing_avg_ms: All-time average request
        processing duration in milliseconds.
    :param request_processing_max_ms: All-time maximum request
        processing duration in milliseconds.
    :param request_processing_avg_1s_ms: Average processing duration
        for requests completed in the last second.
    :param request_processing_avg_10s_ms: Average processing duration
        for requests completed in the last ten seconds.
    :param request_processing_avg_30s_ms: Average processing duration
        for requests completed in the last thirty seconds.
    :param process_cpu_percent: Process CPU percent since the prior
        snapshot.
    """

    in_flight: int
    total_started: int
    total_completed: int
    total_failed: int
    requests_last_1s: int
    requests_last_10s: int
    requests_last_30s: int
    active_websockets: int
    request_processing_avg_ms: float
    request_processing_max_ms: float
    request_processing_avg_1s_ms: float
    request_processing_avg_10s_ms: float
    request_processing_avg_30s_ms: float
    process_cpu_percent: float


@dataclass(frozen=True)
class ServerMetricsSnapshot:
    """
    Point-in-time server performance counters.

    :param in_flight: HTTP requests currently executing in the
        FastAPI app.
    :param total_started: Total HTTP requests accepted by the
        metrics middleware since process start.
    :param total_completed: Total HTTP requests that left the
        metrics middleware since process start.
    :param total_failed: Total HTTP requests that raised or returned
        a 5xx status since process start.
    :param requests_last_1s: HTTP requests started in the last
        second.
    :param requests_last_10s: HTTP requests started in the last ten
        seconds.
    :param requests_last_30s: HTTP requests started in the last
        thirty seconds.
    :param active_websockets: Accepted WebSocket connections currently
        open in the FastAPI app.
    :param request_processing_avg_ms: Average processing duration for
        completed HTTP requests since process start, in milliseconds.
    :param request_processing_max_ms: Maximum processing duration for
        completed HTTP requests since process start, in milliseconds.
    :param request_processing_avg_1s_ms: Average processing duration
        for requests completed in the last second, in milliseconds.
    :param request_processing_avg_10s_ms: Average processing duration
        for requests completed in the last ten seconds, in
        milliseconds.
    :param request_processing_avg_30s_ms: Average processing duration
        for requests completed in the last thirty seconds, in
        milliseconds.
    :param process_cpu_percent: Process CPU use since the previous
        snapshot, expressed as ``process_cpu_seconds / wall_seconds *
        100``. The first snapshot reports ``0.0``.
    :param load_average_1m: System load average over one minute, or
        ``None`` when the platform does not expose load averages.
    :param load_average_5m: System load average over five minutes, or
        ``None`` when the platform does not expose load averages.
    :param load_average_15m: System load average over fifteen minutes,
        or ``None`` when the platform does not expose load averages.
    :param rss_bytes: Resident set size in bytes. On Linux this is
        current RSS from ``/proc/self/status``; elsewhere it falls
        back to ``resource.getrusage().ru_maxrss``.
    """

    in_flight: int
    total_started: int
    total_completed: int
    total_failed: int
    requests_last_1s: int
    requests_last_10s: int
    requests_last_30s: int
    active_websockets: int
    request_processing_avg_ms: float
    request_processing_max_ms: float
    request_processing_avg_1s_ms: float
    request_processing_avg_10s_ms: float
    request_processing_avg_30s_ms: float
    process_cpu_percent: float
    load_average_1m: float | None
    load_average_5m: float | None
    load_average_15m: float | None
    rss_bytes: int

    @property
    def rss_mib(self) -> float:
        """
        Return resident memory as mebibytes.

        :returns: ``rss_bytes`` divided by 1024².
        """
        return self.rss_bytes / _BYTES_PER_MIB


class ServerPerformanceMetrics:
    """
    Process-local metrics tracker for HTTP server requests.

    The tracker is intentionally in-memory and dependency-free. It
    records request start timestamps for rolling-window request
    counters, maintains current in-flight request count, and samples
    process/system resource counters when a snapshot is taken.

    :param clock: Monotonic wall-clock function, e.g.
        ``time.monotonic``. Tests pass a deterministic fake clock.
    :param process_time_fn: Process CPU clock function, e.g.
        ``time.process_time``.
    :param rss_bytes_fn: Function that returns resident memory in
        bytes.
    :param load_avg_fn: Function that returns the one-, five-, and
        fifteen-minute load averages.
    """

    def __init__(
        self,
        *,
        clock: FloatSampler = time.monotonic,
        process_time_fn: FloatSampler = time.process_time,
        rss_bytes_fn: RssSampler | None = None,
        load_avg_fn: LoadAverageSampler | None = None,
    ) -> None:
        """
        Initialize the metrics tracker.

        :param clock: Monotonic wall-clock function, e.g.
            ``time.monotonic``.
        :param process_time_fn: Process CPU clock function, e.g.
            ``time.process_time``.
        :param rss_bytes_fn: Optional resident memory sampler. ``None``
            uses the module's stdlib sampler.
        :param load_avg_fn: Optional system load sampler. ``None``
            uses ``os.getloadavg`` when available.
        """
        self._clock = clock
        self._process_time_fn = process_time_fn
        self._rss_bytes_fn = rss_bytes_fn or _current_rss_bytes
        self._load_avg_fn = load_avg_fn or _load_average
        self._lock = Lock()
        self._request_starts: deque[float] = deque()
        self._request_timings: deque[CompletedRequestTiming] = deque()
        self._in_flight = 0
        self._total_started = 0
        self._total_completed = 0
        self._total_failed = 0
        self._active_websockets = 0
        self._total_processing_seconds = 0.0
        self._max_processing_seconds = 0.0
        self._last_snapshot_wall = clock()
        self._last_snapshot_cpu = process_time_fn()

    def request_started(self) -> float:
        """
        Record that one HTTP request entered the server.

        :returns: Monotonic start timestamp to pass to
            :meth:`request_finished`.
        """
        now = self._clock()
        with self._lock:
            self._request_starts.append(now)
            self._in_flight += 1
            self._total_started += 1
            self._prune_locked(now)
        return now

    def request_finished(self, *, started_at: float, failed: bool = False) -> float:
        """
        Record that one HTTP request left the server.

        :param started_at: Monotonic start timestamp returned by
            :meth:`request_started`.
        :param failed: Whether the request failed, e.g. raised an
            exception or returned a 5xx response.
        :returns: Request processing duration in seconds.
        """
        now = self._clock()
        duration_seconds = max(0.0, now - started_at)
        with self._lock:
            self._in_flight = max(0, self._in_flight - 1)
            self._total_completed += 1
            if failed:
                self._total_failed += 1
            self._total_processing_seconds += duration_seconds
            self._max_processing_seconds = max(
                self._max_processing_seconds,
                duration_seconds,
            )
            self._request_timings.append(
                CompletedRequestTiming(
                    completed_at=now,
                    duration_seconds=duration_seconds,
                )
            )
            self._prune_locked(now)
        return duration_seconds

    def websocket_connected(self) -> None:
        """
        Record that one WebSocket connection was accepted.
        """
        with self._lock:
            self._active_websockets += 1

    def websocket_disconnected(self) -> None:
        """
        Record that one accepted WebSocket connection closed.
        """
        with self._lock:
            self._active_websockets = max(0, self._active_websockets - 1)

    def snapshot(self) -> ServerMetricsSnapshot:
        """
        Capture request counters and process resource usage.

        :returns: A point-in-time immutable metrics snapshot.
        """
        now = self._clock()
        process_cpu = self._process_time_fn()
        with self._lock:
            self._prune_locked(now)
            request_values = self._request_metric_values_locked(now, process_cpu)

        load = self._load_avg_fn()
        if load is None:
            load_average_1m = None
            load_average_5m = None
            load_average_15m = None
        else:
            load_average_1m = load.one_minute
            load_average_5m = load.five_minutes
            load_average_15m = load.fifteen_minutes
        return ServerMetricsSnapshot(
            in_flight=request_values.in_flight,
            total_started=request_values.total_started,
            total_completed=request_values.total_completed,
            total_failed=request_values.total_failed,
            requests_last_1s=request_values.requests_last_1s,
            requests_last_10s=request_values.requests_last_10s,
            requests_last_30s=request_values.requests_last_30s,
            active_websockets=request_values.active_websockets,
            request_processing_avg_ms=request_values.request_processing_avg_ms,
            request_processing_max_ms=request_values.request_processing_max_ms,
            request_processing_avg_1s_ms=request_values.request_processing_avg_1s_ms,
            request_processing_avg_10s_ms=request_values.request_processing_avg_10s_ms,
            request_processing_avg_30s_ms=request_values.request_processing_avg_30s_ms,
            process_cpu_percent=request_values.process_cpu_percent,
            load_average_1m=load_average_1m,
            load_average_5m=load_average_5m,
            load_average_15m=load_average_15m,
            rss_bytes=self._rss_bytes_fn(),
        )

    def _request_metric_values_locked(
        self,
        now: float,
        process_cpu: float,
    ) -> RequestMetricValues:
        """
        Copy request counters and timing values under the tracker lock.

        :param now: Current monotonic time in seconds.
        :param process_cpu: Current process CPU time in seconds.
        :returns: Request metrics that do not require further lock
            access.
        """
        return RequestMetricValues(
            in_flight=self._in_flight,
            total_started=self._total_started,
            total_completed=self._total_completed,
            total_failed=self._total_failed,
            requests_last_1s=self._count_since_locked(now - 1.0),
            requests_last_10s=self._count_since_locked(now - 10.0),
            requests_last_30s=self._count_since_locked(now - 30.0),
            active_websockets=self._active_websockets,
            request_processing_avg_ms=self._overall_processing_avg_ms_locked(),
            request_processing_max_ms=self._max_processing_seconds * 1000.0,
            request_processing_avg_1s_ms=self._processing_avg_since_locked(now - 1.0),
            request_processing_avg_10s_ms=self._processing_avg_since_locked(now - 10.0),
            request_processing_avg_30s_ms=self._processing_avg_since_locked(now - 30.0),
            process_cpu_percent=self._cpu_percent_locked(now, process_cpu),
        )

    def _prune_locked(self, now: float) -> None:
        """
        Drop request starts older than the largest reporting window.

        :param now: Current monotonic time in seconds.
        """
        cutoff = now - max(_DEFAULT_WINDOWS_SECONDS)
        while self._request_starts and self._request_starts[0] < cutoff:
            self._request_starts.popleft()
        while self._request_timings and self._request_timings[0].completed_at < cutoff:
            self._request_timings.popleft()

    def _count_since_locked(self, cutoff: float) -> int:
        """
        Count request starts newer than ``cutoff``.

        :param cutoff: Exclusive lower-bound monotonic timestamp.
        :returns: Number of retained request starts at or after
            ``cutoff``.
        """
        return sum(1 for started_at in self._request_starts if started_at >= cutoff)

    def _overall_processing_avg_ms_locked(self) -> float:
        """
        Return all-time average completed request processing time.

        :returns: Average processing duration in milliseconds, or
            ``0.0`` when no request has completed.
        """
        if self._total_completed == 0:
            return 0.0
        return (self._total_processing_seconds / self._total_completed) * 1000.0

    def _processing_avg_since_locked(self, cutoff: float) -> float:
        """
        Return average processing time for recently completed requests.

        :param cutoff: Exclusive lower-bound completion timestamp.
        :returns: Average processing duration in milliseconds, or
            ``0.0`` when no retained request completed after
            ``cutoff``.
        """
        total_seconds = 0.0
        count = 0
        for timing in self._request_timings:
            if timing.completed_at >= cutoff:
                total_seconds += timing.duration_seconds
                count += 1
        if count == 0:
            return 0.0
        return (total_seconds / count) * 1000.0

    def _cpu_percent_locked(self, now: float, process_cpu: float) -> float:
        """
        Compute CPU percent since the previous snapshot.

        :param now: Current monotonic time in seconds.
        :param process_cpu: Current process CPU time in seconds.
        :returns: Percent CPU used since the prior snapshot.
        """
        wall_delta = max(0.0, now - self._last_snapshot_wall)
        cpu_delta = max(0.0, process_cpu - self._last_snapshot_cpu)
        self._last_snapshot_wall = now
        self._last_snapshot_cpu = process_cpu
        if wall_delta <= 0:
            return 0.0
        return (cpu_delta / wall_delta) * 100.0


def _create_request_otel_instruments(meter: MeterLike) -> _RequestOtelInstruments:
    """
    Create OpenTelemetry instruments for request and WebSocket state.

    :param meter: Meter used to create instruments.
    :returns: Grouped request instruments.
    """
    return _RequestOtelInstruments(
        started=meter.create_counter(
            "omnigent.server.http.requests.started",
            unit="{request}",
            description="HTTP requests started by the Omnigent server.",
        ),
        completed=meter.create_counter(
            "omnigent.server.http.requests.completed",
            unit="{request}",
            description="HTTP requests completed by the Omnigent server.",
        ),
        failed=meter.create_counter(
            "omnigent.server.http.requests.failed",
            unit="{request}",
            description="HTTP requests failed by exception or 5xx status.",
        ),
        duration=meter.create_histogram(
            "omnigent.server.http.request.duration",
            unit="s",
            description="HTTP request processing duration.",
        ),
        in_flight=meter.create_gauge(
            "omnigent.server.http.requests.in_flight",
            unit="{request}",
            description="HTTP requests currently being processed.",
        ),
        last_1s=meter.create_gauge(
            "omnigent.server.http.requests.last_1s",
            unit="{request}",
            description="HTTP requests started in the last second.",
        ),
        last_10s=meter.create_gauge(
            "omnigent.server.http.requests.last_10s",
            unit="{request}",
            description="HTTP requests started in the last ten seconds.",
        ),
        last_30s=meter.create_gauge(
            "omnigent.server.http.requests.last_30s",
            unit="{request}",
            description="HTTP requests started in the last thirty seconds.",
        ),
        active_websockets=meter.create_gauge(
            "omnigent.server.websocket.connections.active",
            unit="{connection}",
            description="Accepted WebSocket connections currently open.",
        ),
    )


def _create_processing_otel_instruments(
    meter: MeterLike,
) -> _ProcessingOtelInstruments:
    """
    Create OpenTelemetry instruments for request processing durations.

    :param meter: Meter used to create instruments.
    :returns: Grouped processing-duration instruments.
    """
    return _ProcessingOtelInstruments(
        avg_ms=meter.create_gauge(
            "omnigent.server.http.request.processing.avg",
            unit="ms",
            description="Average HTTP request processing duration since process start.",
        ),
        max_ms=meter.create_gauge(
            "omnigent.server.http.request.processing.max",
            unit="ms",
            description="Maximum HTTP request processing duration since process start.",
        ),
        avg_1s_ms=meter.create_gauge(
            "omnigent.server.http.request.processing.avg_1s",
            unit="ms",
            description=("Average processing duration for requests completed in the last second."),
        ),
        avg_10s_ms=meter.create_gauge(
            "omnigent.server.http.request.processing.avg_10s",
            unit="ms",
            description=(
                "Average processing duration for requests completed in the last ten seconds."
            ),
        ),
        avg_30s_ms=meter.create_gauge(
            "omnigent.server.http.request.processing.avg_30s",
            unit="ms",
            description=(
                "Average processing duration for requests completed in the last thirty seconds."
            ),
        ),
    )


def _create_resource_otel_instruments(meter: MeterLike) -> _ResourceOtelInstruments:
    """
    Create OpenTelemetry instruments for process and system resources.

    :param meter: Meter used to create instruments.
    :returns: Grouped resource instruments.
    """
    return _ResourceOtelInstruments(
        cpu_percent=meter.create_gauge(
            "omnigent.server.process.cpu.percent",
            unit="%",
            description="Process CPU utilization since the previous server metrics snapshot.",
        ),
        load_1m=meter.create_gauge(
            "omnigent.server.system.load_average.1m",
            unit="{load}",
            description="One-minute system load average.",
        ),
        load_5m=meter.create_gauge(
            "omnigent.server.system.load_average.5m",
            unit="{load}",
            description="Five-minute system load average.",
        ),
        load_15m=meter.create_gauge(
            "omnigent.server.system.load_average.15m",
            unit="{load}",
            description="Fifteen-minute system load average.",
        ),
        rss_bytes=meter.create_gauge(
            "omnigent.server.process.memory.rss",
            unit="By",
            description="Resident memory used by the server process.",
        ),
    )


class ServerMetricsOtelPublisher:
    """
    Publish server performance metrics through OpenTelemetry instruments.

    Periodic snapshot values are exported from a single background
    publisher. That avoids an independent observable callback
    perturbing the CPU-delta state inside
    :class:`ServerPerformanceMetrics`.

    :param meter: Optional meter used to create instruments. ``None``
        uses ``opentelemetry.metrics.get_meter`` with the package's
        server performance meter name.
    """

    def __init__(self, meter: MeterLike | None = None) -> None:
        """
        Initialize OpenTelemetry instruments.

        :param meter: Optional meter used by tests to inject real
            recording stubs. ``None`` uses the global OpenTelemetry
            meter provider.
        """
        effective_meter = meter or otel_metrics.get_meter(_OTEL_METER_NAME)
        self._counter_state = _CounterState()
        self._request = _create_request_otel_instruments(effective_meter)
        self._processing = _create_processing_otel_instruments(effective_meter)
        self._resource = _create_resource_otel_instruments(effective_meter)

    def record_request_duration(
        self,
        *,
        duration_seconds: float,
        failed: bool,
        method: str,
        route: str,
        status_code: int | None,
    ) -> None:
        """
        Record one completed HTTP request duration.

        :param duration_seconds: Request processing duration in
            seconds.
        :param failed: Whether the request failed by exception or
            5xx status.
        :param method: HTTP request method, e.g. ``"GET"``.
        :param route: FastAPI route template, e.g.
            ``"/v1/sessions/{session_id}"``.
        :param status_code: HTTP response status code, e.g. ``200``.
            ``None`` when no response status was available.
        """
        attributes: dict[str, str | bool | int] = {
            "failed": failed,
            "http.request.method": method,
            "http.route": route,
        }
        if status_code is not None:
            attributes["http.response.status_code"] = status_code
        self._request.duration.record(
            duration_seconds,
            attributes=attributes,
        )

    def publish(self, snapshot: ServerMetricsSnapshot) -> None:
        """
        Publish one server metrics snapshot through OTEL instruments.

        :param snapshot: Snapshot produced by
            :meth:`ServerPerformanceMetrics.snapshot`.
        """
        self._publish_counter_deltas(snapshot)
        self._request.in_flight.set(snapshot.in_flight)
        self._request.last_1s.set(snapshot.requests_last_1s)
        self._request.last_10s.set(snapshot.requests_last_10s)
        self._request.last_30s.set(snapshot.requests_last_30s)
        self._request.active_websockets.set(snapshot.active_websockets)
        self._processing.avg_ms.set(snapshot.request_processing_avg_ms)
        self._processing.max_ms.set(snapshot.request_processing_max_ms)
        self._processing.avg_1s_ms.set(snapshot.request_processing_avg_1s_ms)
        self._processing.avg_10s_ms.set(snapshot.request_processing_avg_10s_ms)
        self._processing.avg_30s_ms.set(snapshot.request_processing_avg_30s_ms)
        self._resource.cpu_percent.set(snapshot.process_cpu_percent)
        if snapshot.load_average_1m is not None:
            self._resource.load_1m.set(snapshot.load_average_1m)
        if snapshot.load_average_5m is not None:
            self._resource.load_5m.set(snapshot.load_average_5m)
        if snapshot.load_average_15m is not None:
            self._resource.load_15m.set(snapshot.load_average_15m)
        self._resource.rss_bytes.set(snapshot.rss_bytes)

    def _publish_counter_deltas(self, snapshot: ServerMetricsSnapshot) -> None:
        """
        Publish cumulative snapshot counters as monotonic deltas.

        :param snapshot: Snapshot containing cumulative totals.
        """
        self._add_delta(
            self._request.started,
            snapshot.total_started - self._counter_state.total_started,
        )
        self._add_delta(
            self._request.completed,
            snapshot.total_completed - self._counter_state.total_completed,
        )
        self._add_delta(
            self._request.failed,
            snapshot.total_failed - self._counter_state.total_failed,
        )
        self._counter_state.total_started = snapshot.total_started
        self._counter_state.total_completed = snapshot.total_completed
        self._counter_state.total_failed = snapshot.total_failed

    def _add_delta(self, counter: CounterInstrument, amount: int) -> None:
        """
        Add a non-negative delta to an OpenTelemetry counter.

        :param counter: Counter instrument to update.
        :param amount: Delta since the prior published snapshot.
        """
        if amount > 0:
            counter.add(amount)


def set_request_duration_for_access_log(duration_seconds: float | None) -> None:
    """
    Store the current request duration for Uvicorn access formatting.

    :param duration_seconds: Request processing duration in seconds,
        or ``None`` to clear the value for a test or non-request
        context.
    """
    _REQUEST_DURATION_CONTEXT.set(duration_seconds)


def set_request_id_for_access_log(request_id: str | None) -> None:
    """
    Store the request correlation ID for Uvicorn access formatting.

    :param request_id: UUID hex identifying this request, or ``None``
        to clear.
    """
    _REQUEST_ID_CONTEXT.set(request_id)


def set_request_user_agent_for_access_log(user_agent: str | None) -> None:
    """
    Store the User-Agent header for Uvicorn access formatting.

    :param user_agent: Raw ``User-Agent`` header value, or ``None``
        to clear.
    """
    _REQUEST_USER_AGENT_CONTEXT.set(user_agent)


def set_request_session_id_for_access_log(session_id: str | None) -> None:
    """
    Store the session ID for Uvicorn access formatting.

    :param session_id: Session/conversation ID extracted from the
        request path, or ``None`` when the path does not target a
        session.
    """
    _REQUEST_SESSION_ID_CONTEXT.set(session_id)


async def publish_server_metrics_periodically(
    metrics: ServerPerformanceMetrics,
    *,
    otel_publisher: ServerMetricsOtelPublisher,
    interval_seconds: float = 10.0,
) -> None:
    """
    Publish server metrics snapshots to OpenTelemetry until cancelled.

    :param metrics: Metrics tracker to snapshot.
    :param otel_publisher: OpenTelemetry publisher that emits
        snapshot values as metrics.
    :param interval_seconds: Delay between OTEL snapshots in seconds,
        e.g. ``10.0``.
    """
    while True:
        await asyncio.sleep(interval_seconds)
        otel_publisher.publish(metrics.snapshot())


def _load_average() -> SystemLoadAverage | None:
    """
    Return system load averages when the platform supports them.

    :returns: One-, five-, and fifteen-minute load averages, or
        ``None`` on platforms without ``os.getloadavg`` support.
    """
    try:
        one_minute, five_minutes, fifteen_minutes = os.getloadavg()
        return SystemLoadAverage(
            one_minute=one_minute,
            five_minutes=five_minutes,
            fifteen_minutes=fifteen_minutes,
        )
    except (AttributeError, OSError):
        return None


def _current_rss_bytes() -> int:
    """
    Return process resident memory in bytes.

    Linux exposes current RSS in ``/proc/self/status``. macOS falls
    back to ``resource.getrusage().ru_maxrss`` (maximum RSS, the best
    stdlib-only option there). Windows has no ``resource`` module, so it
    falls back to :mod:`psutil` (a core dependency), which reports true
    current RSS.

    :returns: Resident memory in bytes.
    """
    proc_status = "/proc/self/status"
    try:
        with open(proc_status, encoding="utf-8") as handle:
            for line in handle:
                if line.startswith("VmRSS:"):
                    parts = line.split()
                    if len(parts) >= 2:
                        return int(parts[1]) * 1024
    except (FileNotFoundError, OSError, ValueError):
        pass

    if resource is not None:
        usage = resource.getrusage(resource.RUSAGE_SELF)
        if sys.platform == "darwin":
            return int(usage.ru_maxrss)
        return int(usage.ru_maxrss) * 1024

    import psutil

    return int(psutil.Process().memory_info().rss)