wave-48 backend deploy

apex/observability.py

@@ -40,6 +40,9 @@ from __future__ import annotations
 
 import logging
 import os
+import time
+
+from apex import observability_metrics
 
 logger = logging.getLogger(__name__)
 
@@ -162,14 +165,36 @@ def setup_observability(app):
         @app.middleware("http")
         async def apex_otel_request_middleware(request, call_next):
             path = request.url.path
-            if path in {"/healthz", "/favicon.ico"}:
+            # Skip readiness + the observability panel's own polling so the
+            # live dashboard reflects real product traffic, not self-polling.
+            if path in {"/healthz", "/favicon.ico", "/api/observability/summary"}:
                 return await call_next(request)
+            started = time.perf_counter()
             with tracer.start_as_current_span(f"{request.method} {path}") as span:
                 span.set_attribute("http.method", request.method)
                 span.set_attribute("http.url", str(request.url))
                 span.set_attribute("http.route", path)
                 response = await call_next(request)
                 span.set_attribute("http.status_code", response.status_code)
+                # Mirror into the in-process live-metrics aggregator with the
+                # REAL Honeycomb trace_id so /api/observability/summary can
+                # deep-link each recent request to its trace waterfall.
+                try:
+                    span_ctx = span.get_span_context()
+                    trace_id_hex = (
+                        format(span_ctx.trace_id, "032x")
+                        if span_ctx and span_ctx.trace_id
+                        else None
+                    )
+                except Exception:  # pragma: no cover - defensive
+                    trace_id_hex = None
+                observability_metrics.record_request(
+                    route=path,
+                    method=request.method,
+                    status=response.status_code,
+                    duration_ms=(time.perf_counter() - started) * 1000.0,
+                    trace_id=trace_id_hex,
+                )
                 return response
 
         logger.info(

apex/observability_metrics.py

@@ -0,0 +1,157 @@
+"""In-process live request metrics for the APEX observability cockpit.
+
+wave-53 production-observability surface. The /judges page renders a LIVE
+telemetry panel fed by `GET /api/observability/summary`. This module holds
+the in-process aggregator that the request middleware writes to and the
+summary route reads from. Pure stdlib, thread-safe, zero new dependencies.
+
+Honesty contract (see project memory feedback_conceptual_stack_vs_shipped_
+stack): every number here is REAL traffic this process has served since
+boot. A cold start (zero requests) is reported truthfully, never faked. The
+same spans are exported to Honeycomb over OTLP by `apex/observability.py`;
+this aggregator is the in-product mirror so judges can see live metrics
+without a Honeycomb login, and each recent request carries its real
+Honeycomb `trace_id` for a deep-link into the actual trace waterfall.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections import deque
+from typing import Any
+
+_MAX_RECENT = 200          # ring buffer used for latency percentiles
+_MAX_RECENT_TRACES = 12    # recent requests surfaced to the panel (deep-linked)
+_STATUS_CLASSES = ("2xx", "3xx", "4xx", "5xx")
+
+
+def _percentile(sorted_values: list[float], pct: float) -> float:
+    """Nearest-rank percentile over an already-sorted list (honest, simple)."""
+    if not sorted_values:
+        return 0.0
+    idx = int(round((pct / 100.0) * (len(sorted_values) - 1)))
+    idx = max(0, min(len(sorted_values) - 1, idx))
+    return round(sorted_values[idx], 2)
+
+
+class LiveMetrics:
+    """Thread-safe in-process request aggregator."""
+
+    def __init__(self) -> None:
+        self._lock = threading.Lock()
+        self._reset_locked()
+
+    def _reset_locked(self) -> None:
+        self._start = time.time()
+        self._total = 0
+        self._status_class: dict[str, int] = {c: 0 for c in _STATUS_CLASSES}
+        self._by_route: dict[str, dict[str, float]] = {}
+        self._recent: deque = deque(maxlen=_MAX_RECENT)
+
+    def reset(self) -> None:
+        with self._lock:
+            self._reset_locked()
+
+    def record(
+        self,
+        *,
+        route: str,
+        method: str,
+        status: int,
+        duration_ms: float,
+        trace_id: str | None = None,
+    ) -> None:
+        cls = f"{int(status) // 100}xx"
+        entry = {
+            "ts": time.time(),
+            "route": route,
+            "method": method,
+            "status": int(status),
+            "duration_ms": round(float(duration_ms), 2),
+            "trace_id": trace_id,
+        }
+        with self._lock:
+            self._total += 1
+            self._status_class[cls] = self._status_class.get(cls, 0) + 1
+            bucket = self._by_route.setdefault(route, {"count": 0.0, "sum_ms": 0.0})
+            bucket["count"] += 1.0
+            bucket["sum_ms"] += float(duration_ms)
+            self._recent.append(entry)
+
+    def snapshot(self) -> dict[str, Any]:
+        now = time.time()
+        with self._lock:
+            recent = list(self._recent)
+            total = self._total
+            status_class = dict(self._status_class)
+            by_route = {
+                route: {
+                    "count": int(v["count"]),
+                    "avg_ms": round(v["sum_ms"] / v["count"], 2) if v["count"] else 0.0,
+                }
+                for route, v in self._by_route.items()
+            }
+            uptime_s = now - self._start
+        durations = sorted(e["duration_ms"] for e in recent)
+        requests_last_60s = sum(1 for e in recent if now - e["ts"] <= 60.0)
+        recent_traces = [
+            {
+                "trace_id": e["trace_id"],
+                "route": e["route"],
+                "method": e["method"],
+                "status": e["status"],
+                "duration_ms": e["duration_ms"],
+                "ts": e["ts"],
+            }
+            for e in reversed(recent)
+            if e["trace_id"]
+        ][:_MAX_RECENT_TRACES]
+        return {
+            "total_requests": total,
+            "status_class": status_class,
+            "by_route": by_route,
+            "latency_ms": {
+                "p50": _percentile(durations, 50),
+                "p95": _percentile(durations, 95),
+                "p99": _percentile(durations, 99),
+                "window": len(durations),
+            },
+            "uptime_s": round(uptime_s, 1),
+            "eps_1m": round(requests_last_60s / 60.0, 3),
+            "recent_traces": recent_traces,
+        }
+
+
+_LIVE = LiveMetrics()
+
+
+def record_request(
+    *,
+    route: str,
+    method: str,
+    status: int,
+    duration_ms: float,
+    trace_id: str | None = None,
+) -> None:
+    """Record one served request into the process-global live aggregator."""
+    _LIVE.record(
+        route=route,
+        method=method,
+        status=status,
+        duration_ms=duration_ms,
+        trace_id=trace_id,
+    )
+
+
+def snapshot() -> dict[str, Any]:
+    """Return the current live-metrics snapshot (real traffic since boot)."""
+    return _LIVE.snapshot()
+
+
+def reset_metrics() -> None:
+    """Reset the process-global aggregator (used by tests)."""
+    _LIVE.reset()
+
+
+__all__ = ["LiveMetrics", "record_request", "snapshot", "reset_metrics"]

apex/server.py

@@ -14,6 +14,7 @@ Exposes:
   - GET  /api/judges/coa-diff     (wave-49 V14 paired COA verdict diff)
   - GET  /api/tire-degradation    (wave-49 Phase 7.2 wear extrapolation)
   - POST /api/critics/verdict     (wave-49 D-018 tri-agent Mellea IVR critic)
+  - GET  /api/observability/summary (wave-53 live OTel telemetry cockpit)
   - GET  /healthz                 (container readiness probe)
 
 Deploy target: any Docker host (Modal / Fly.io / Vercel functions /
@@ -32,12 +33,14 @@ import json
 import os
 import shutil
 import tempfile
+import time
 from pathlib import Path
 from typing import Any, Final
 
 from fastapi import FastAPI, File, HTTPException, Request, UploadFile
 from fastapi.middleware.cors import CORSMiddleware
 
+from apex import observability_metrics
 from apex.critics import run_tri_agent_critics
 from apex.instruct.narrator import Narrator
 from apex.instruct.openrouter_generator import build_openrouter_generator
@@ -146,6 +149,30 @@ app.add_middleware(
 _tracer = setup_observability(app)
 
 
+# Live-metrics fallback for the /judges observability cockpit. When OTel is
+# active (production HF Space), apex/observability.py's span middleware
+# records each request WITH its Honeycomb trace_id. When OTel is off
+# (CI/local/dev), this lightweight middleware still feeds the live panel so
+# `GET /api/observability/summary` is never empty in development. Gated on
+# `_tracer is None` to avoid double-counting when the OTel middleware runs.
+if _tracer is None:
+
+    @app.middleware("http")
+    async def _apex_metrics_fallback_middleware(request: Request, call_next):
+        path = request.url.path
+        if path in {"/healthz", "/favicon.ico", "/api/observability/summary"}:
+            return await call_next(request)
+        started = time.perf_counter()
+        response = await call_next(request)
+        observability_metrics.record_request(
+            route=path,
+            method=request.method,
+            status=response.status_code,
+            duration_ms=(time.perf_counter() - started) * 1000.0,
+        )
+        return response
+
+
 @app.get("/healthz", response_model=HealthzResp)
 def healthz() -> HealthzResp:
     """Container readiness probe. Returns 200 once the singletons load."""
@@ -650,4 +677,39 @@ async def post_critics_verdict(request: Request) -> dict[str, Any]:
     return run_tri_agent_critics(report_summary)
 
 
+# ---- GET /api/observability/summary -----------------------------------
+
+@app.get("/api/observability/summary")
+def get_observability_summary() -> dict[str, Any]:
+    """Live in-process request telemetry for the /judges observability cockpit.
+
+    Honest in-product mirror of the same spans this service exports to
+    Honeycomb over OTLP (see apex/observability.py). Every number is REAL
+    traffic served since process boot; a cold start reports zeroes, never
+    fabricated values. Each recent request carries its Honeycomb trace_id so
+    the frontend can deep-link into the real trace waterfall.
+    """
+    otel_enabled = os.environ.get("APEX_OTEL_ENABLED", "").strip().lower() in {
+        "1",
+        "true",
+        "yes",
+    }
+    otlp_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT", "").strip()
+    otlp_active = bool(otel_enabled and otlp_endpoint)
+    if otlp_active:
+        exporter = "otlp"
+    elif otel_enabled:
+        exporter = "console"
+    else:
+        exporter = "none"
+    return {
+        "service_name": os.environ.get("OTEL_SERVICE_NAME", "apex-backend"),
+        "otel_enabled": otel_enabled,
+        "otlp_active": otlp_active,
+        "exporter": exporter,
+        "captured_at": time.time(),
+        **observability_metrics.snapshot(),
+    }
+
+
 __all__ = ["app"]