feat(proxy): add POST /v1/callbacks/logs to replay logging payloads through callbacks

backend/routes/allowlist.py

@@ -83,6 +83,7 @@
     # Callbacks / hooks
     "/active/callbacks",
     "/callbacks",
+    "/v1/callbacks/",
     "/team_callback",
     # Alerting / email / IP allowlist
     "/alerting/",

litellm-rust/crates/ai-gateway/ARCHITECTURE.md

@@ -0,0 +1,100 @@
+# ai-gateway — architecture
+
+The Rust gateway owns the **data plane** (the realtime WebSocket connection and the
+LLM splice). It is **never** in the request path for logging: when a session ends it
+hands a finished `StandardLoggingPayload` to a callback, and a background worker ships
+that to the **LiteLLM Python service** over HTTP. Python does spend logs + the full
+callback fan-out. This keeps the gateway non-blocking under high traffic — Python
+slowness, batching, or downtime can never stall or crash a live session.
+
+The realtime logging wrapper is a 1:1 port of Python's `RealTimeStreaming`
+(`litellm/litellm_core_utils/realtime_streaming.py`): accumulate O(1) per-session
+state during the call, emit **one** payload at close.
+
+## Data plane → logging plane → Python
+
+```mermaid
+flowchart LR
+  subgraph DP["Data plane — Rust ai-gateway (hot path, never blocks)"]
+    direction TB
+    C["client WS<br/>(voice / text)"]
+    BR["routes/realtime — axum bridge<br/>socket ⇄ typed RealtimeEvent"]
+    SP["providers::realtime::realtime<br/>tokio::select! splice ⇄ OpenAI"]
+    RT["RealTimeStreaming.observe(event)<br/>O(1): model, usage, tool tallies<br/>drop the frame"]
+    UP["OpenAI realtime WS"]
+    C <--> BR <--> SP <--> UP
+    SP -. each event .-> RT
+  end
+
+  RT ==>|"on close: log_messages() builds ONE StandardLoggingPayload"| CB
+
+  subgraph LP["Logging plane — decoupled, async"]
+    direction TB
+    CB["CustomLogger fan-out<br/>(list of Arc dyn CustomLogger)"]
+    PX["LiteLLMPythonProxyAPILogger<br/>sink.try_send (non-blocking)"]
+    CH[["bounded byte-capped channel<br/>full → drop + counter"]]
+    WK["egress worker<br/>batch + gzip, pooled client"]
+    CB --> PX --> CH --> WK
+  end
+
+  WK ==>|"POST /v1/callbacks/logs (batched, keep-alive)"| PY
+
+  subgraph PY["LiteLLM Python service (scales horizontally)"]
+    direction TB
+    EP["/v1/callbacks/logs"]
+    RP["CallbackLogsReplayer →<br/>async_success_handler"]
+    OUT["spend logs + every callback<br/>(Langfuse, Datadog, …)"]
+    EP --> RP --> OUT
+  end
+```
+
+## Why it holds up under high traffic
+
+The request path does **zero** logging I/O. Each event costs one `observe()` call —
+a `match` on the event type that drops the frame (audio deltas hit no branch). The
+only network hop for logs is the worker's, off the hot path:
+
+```mermaid
+flowchart LR
+  G1["gateway pod 1"] --> SVC
+  G2["gateway pod 2"] --> SVC
+  G3["gateway pod N"] --> SVC
+  SVC["LiteLLM Python service<br/>POST /v1/callbacks/logs<br/>(horizontally scaled)"]
+```
+
+- **Non-blocking emit** — handlers only `try_send`; if the channel is full they drop
+  the record and bump a metric, then keep serving. Python latency/downtime degrades
+  to dropped log lines + a rising counter, never a stalled or crashed gateway.
+- **Bounded memory** — the channel is capped by **bytes** (not record count), and
+  `RealTimeStreaming` keeps **O(1)** per-session state (no buffered frames), so a
+  long voice call can't OOM the pod.
+- **Batching + gzip** — the worker coalesces records and compresses, so many gateway
+  pods fan into a Python service that scales out independently.
+
+## Map to Python
+
+| Python (`realtime_streaming.py` / `integrations/`) | Rust (this crate + `integrations/`) |
+| --- | --- |
+| `RealTimeStreaming` (wrapper, assemble-then-emit-once) | `RealTimeStreaming` (1:1) |
+| `store_message` / `_collect_*` (accumulate) | `RealTimeStreaming.observe(&event)` (O(1)) |
+| `log_messages()` at WS close (once) | `RealTimeStreaming::log_messages()` (once) |
+| `CustomLogger` base class | `trait CustomLogger` (sync, typed args, `-> Result`) |
+| `async_log_success_event` / `async_log_failure_event` | `log_success_event` / `log_failure_event` |
+| a concrete integration (datadog, langfuse, …) | `LiteLLMPythonProxyAPILogger` (POSTs to the proxy) |
+| in-process `async_success_handler` | `POST /v1/callbacks/logs` → same `async_success_handler` |
+
+## Where the code lives
+
+```
+crates/ai-gateway/
+  routes/realtime/        # axum surface + splice service (data plane)
+  integrations/           # the callback layer (mirrors litellm/integrations/)
+    custom_logger.rs       #   trait CustomLogger — base contract
+    types.rs               #   typed StandardLoggingPayload, Usage, RequestMetadata, LoggingError
+    litellm_python_proxy_api.rs  #   LiteLLMPythonProxyAPILogger: bounded channel + egress worker
+  realtime/streaming.rs   # RealTimeStreaming — per-session O(1) collector
+```
+
+The Python ingest endpoint (`POST /v1/callbacks/logs`) ships in
+[BerriAI/litellm#31134](https://github.com/BerriAI/litellm/pull/31134); the realtime
+splice in [#31135](https://github.com/BerriAI/litellm/pull/31135).

litellm-rust/crates/ai-gateway/Cargo.toml

@@ -14,7 +14,9 @@ litellm-core.workspace = true
 litellm-providers.workspace = true
 axum = { workspace = true, features = ["ws"] }
 futures-util.workspace = true
-tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "time"] }
+tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "time", "sync"] }
+# reqwest (rustls + json) ships realtime logs to the Python proxy callbacks API.
+reqwest = { workspace = true }
 serde.workspace = true
 serde_json.workspace = true
 subtle.workspace = true

litellm-rust/crates/ai-gateway/README.md

@@ -7,6 +7,7 @@ dials OpenAI upstream, and splices the two sockets frame-by-frame.
 - **Client endpoint:** `wss://<host>/v1/realtime?model=<model>` (WebSocket)
 - **Auth:** `Authorization: Bearer $LITELLM_MASTER_KEY` (fails closed if unset)
 - **Health:** `GET /health/readiness`, `GET /health/liveness`, `GET /health/gil`
+- **Request logs:** piped to the LiteLLM control plane via `POST /v1/callbacks/logs` (see [Request logging](#request-logging--litellm-control-plane))
 
 > **Realtime serving is pure Rust.** Python is used at **load time only** — to
 > read the config once at boot. The realtime hot path never touches Python.
@@ -53,6 +54,7 @@ overridden at deploy time (e.g. a Render secret file mounted at the same path).
 | `OPENAI_API_KEY` | yes | — | Upstream OpenAI key. Referenced by config.yaml as `os.environ/OPENAI_API_KEY` for the gateway→OpenAI dial. |
 | `HOST` | no | `127.0.0.1` | **Set to `0.0.0.0` in any container/deploy** or external traffic is refused. |
 | `PORT` | no | `4001` | Listen port. Render and most PaaS inject this automatically. |
+| `LITELLM_PROXY_BASE_URL` | no | `http://localhost:4000` | Base URL of the LiteLLM control plane to pipe request logs to. See [Request logging](#request-logging--litellm-control-plane). |
 
 > Secrets (`LITELLM_MASTER_KEY`, `OPENAI_API_KEY`) are never baked into the image
 > or `render.yaml` — inject them at deploy time only.
@@ -71,6 +73,63 @@ This mode links no libpython and needs no config file, but it only supports one
 hard-coded OpenAI deployment. **config.yaml is the recommended path** — use the
 stand-in only for the leanest possible build.
 
+## Request logging → LiteLLM control plane
+
+The gateway owns the realtime connection, but it does **not** run spend logic
+itself. Instead, when a realtime session ends it emits **one**
+`StandardLoggingPayload` (model, cumulative usage, timing, session id) and ships
+it to the **LiteLLM control plane** — your existing LiteLLM proxy — which replays
+it through its normal callback fan-out. Spend logs, Langfuse, Datadog, and every
+other configured callback fire exactly as they would for a Python-served request.
+
+The gateway POSTs to **`{LITELLM_PROXY_BASE_URL}/v1/callbacks/logs`** (an
+admin-only endpoint on the proxy) authenticated with the master key.
+
+### Configure it — two env vars
+
+| Var | Default | Purpose |
+|---|---|---|
+| `LITELLM_PROXY_BASE_URL` | `http://localhost:4000` | Base URL of the LiteLLM control plane (your proxy). The gateway POSTs request logs to `<this>/v1/callbacks/logs`. |
+| `LITELLM_MASTER_KEY` | — | Sent as `Authorization: Bearer <key>` on the log POST. The `/v1/callbacks/logs` route is admin-only, so this must be the proxy's master key. (Same var the client auth uses.) |
+
+```bash
+# Pipe this gateway's realtime request logs to a LiteLLM proxy at logs.acme.com
+LITELLM_PROXY_BASE_URL=https://logs.acme.com \
+LITELLM_MASTER_KEY=sk-...                     \
+OPENAI_API_KEY=sk-...                         \
+  ./litellm-ai-gateway
+```
+
+On Render / Docker, set the same two vars (the master key is already required for
+client auth, so logging adds only `LITELLM_PROXY_BASE_URL`). Point it at the proxy
+that owns your DB and callbacks — it can be a different host/region than the
+gateway.
+
+### How it behaves
+
+- **Non-blocking.** Handlers only `try_send` the finished payload to a bounded,
+  byte-capped channel; a background worker batches and POSTs. The realtime hot
+  path never awaits the network — if the control plane is slow or down, logs are
+  dropped (with a counter) rather than stalling or crashing the gateway.
+- **One payload per session.** Per-session state is O(1) (no buffered frames), so
+  long voice calls can't OOM the gateway.
+- **`request_id` == the OpenAI realtime session id** (`sess_…`), so a logged row
+  maps 1:1 to OpenAI's session.
+
+The receiving endpoint (`POST /v1/callbacks/logs`) ships with the LiteLLM proxy;
+no extra setup beyond a reachable proxy URL + the master key.
+
+### Tuning (optional)
+
+The egress worker's defaults suit most deployments; override per env var if
+needed:
+
+| Var | Default | Purpose |
+|---|---|---|
+| `LITELLM_LOG_CHANNEL_CAPACITY` | `4096` | Bounded in-memory queue depth. When full, logs are dropped (with a counter) rather than blocking the splice — raise it to absorb larger bursts. |
+| `LITELLM_LOG_BATCH_SIZE` | `256` | Max records POSTed per request to the control plane. |
+| `LITELLM_LOG_FLUSH_INTERVAL_MS` | `500` | How often a partial batch is flushed even if not full. |
+
 ## Build & run with Docker
 
 The image is built `--features python-config` and installs litellm **from this

litellm-rust/crates/ai-gateway/src/integrations/custom_logger.rs

@@ -0,0 +1,24 @@
+//! The `CustomLogger` trait — the Rust mirror of Python
+//! `litellm/integrations/custom_logger.py::CustomLogger`.
+//!
+//! Synchronous (no `async_trait`): callbacks are O(1) enqueue-and-return so the
+//! realtime splice never blocks on a logger. Default bodies are no-ops so a
+//! logger can implement only the events it cares about.
+
+use crate::integrations::types::{LogError, LoggingError, StandardLoggingPayload};
+
+pub trait CustomLogger: Send + Sync {
+    /// Record a successful call. Default: no-op.
+    fn log_success_event(&self, _payload: &StandardLoggingPayload) -> Result<(), LogError> {
+        Ok(())
+    }
+
+    /// Record a failed call. Default: no-op.
+    fn log_failure_event(
+        &self,
+        _payload: &StandardLoggingPayload,
+        _error: &LoggingError,
+    ) -> Result<(), LogError> {
+        Ok(())
+    }
+}