diff --git a/.github/workflows/test-unit-proxy-infra.yml b/.github/workflows/test-unit-proxy-infra.yml index 336e53ee3d74..2681400e8b7d 100644 --- a/.github/workflows/test-unit-proxy-infra.yml +++ b/.github/workflows/test-unit-proxy-infra.yml @@ -29,6 +29,7 @@ jobs: tests/test_litellm/proxy/_experimental tests/test_litellm/proxy/experimental tests/test_litellm/proxy/common_utils + tests/test_litellm/proxy/logging_endpoints tests/test_litellm/proxy/test_*.py workers: 2 reruns: 2 diff --git a/.gitignore b/.gitignore index fda3311fe02a..3563d7c8c2d2 100644 --- a/.gitignore +++ b/.gitignore @@ -123,3 +123,6 @@ crash.*.log # and should be committed. .vscode .pin_list.txt + +# pytest coverage data +.coverage diff --git a/backend/routes/allowlist.py b/backend/routes/allowlist.py index 2f65f99c2926..b67f7d421278 100644 --- a/backend/routes/allowlist.py +++ b/backend/routes/allowlist.py @@ -84,6 +84,8 @@ "/active/callbacks", "/callbacks", "/team_callback", + # Rust data-plane gateway → proxy control-plane API (logging today, auth later) + "/v1/rust_control_plane/", # Alerting / email / IP allowlist "/alerting/", "/email/", diff --git a/litellm-rust/CLAUDE.md b/litellm-rust/CLAUDE.md index 0a985b833f3e..7c723e570ef2 100644 --- a/litellm-rust/CLAUDE.md +++ b/litellm-rust/CLAUDE.md @@ -77,6 +77,20 @@ such as `ai-gateway`, router hosts, or standalone servers: - Avoid `expect`/`unwrap` in server startup and request paths unless the panic is impossible by construction and documented. +## Constants + +Magic numbers and fixed strings go in a crate-level `constants.rs`, never +hardcoded inline — the Rust mirror of Python's `litellm/constants.py`. + +- Each crate that needs them has `src/constants.rs` (declared `mod constants;`); + import from it (`use crate::constants::...`). Don't scatter `const` values at + the top of feature modules. +- An env-overridable tunable still lives in `constants.rs` as its `DEFAULT_*` + value; the env read (with fallback to that default) happens at the host/config + resolution layer, not in `core`/`providers`. +- Exception: a value that is purely local to one function and has no meaning + elsewhere may stay inline, but prefer `constants.rs` when in doubt. + ## Checks Run these before pushing Rust changes. The same checks run in GitHub Actions diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock index 3b8c83aac164..6fe84f1cfbc7 100644 --- a/litellm-rust/Cargo.lock +++ b/litellm-rust/Cargo.lock @@ -571,6 +571,7 @@ dependencies = [ "reqwest", "serde", "serde_json", + "sha2", "subtle", "tokio", "tokio-tungstenite", @@ -1129,6 +1130,17 @@ dependencies = [ "digest", ] +[[package]] +name = "sha2" +version = "0.10.9" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" +dependencies = [ + "cfg-if", + "cpufeatures", + "digest", +] + [[package]] name = "shlex" version = "2.0.1" diff --git a/litellm-rust/Cargo.toml b/litellm-rust/Cargo.toml index 785404c38f18..25ee22130406 100644 --- a/litellm-rust/Cargo.toml +++ b/litellm-rust/Cargo.toml @@ -20,6 +20,7 @@ rand = "0.8" reqwest = { version = "0.12", default-features = false, features = ["blocking", "json", "rustls-tls"] } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" +sha2 = "0.10" subtle = "2" thiserror = "2.0" tokio = { version = "1", features = ["rt-multi-thread", "macros", "time"] } diff --git a/litellm-rust/crates/ai-gateway/ARCHITECTURE.md b/litellm-rust/crates/ai-gateway/ARCHITECTURE.md new file mode 100644 index 000000000000..733953bbdb3b --- /dev/null +++ b/litellm-rust/crates/ai-gateway/ARCHITECTURE.md @@ -0,0 +1,12 @@ +# ai-gateway architecture + +The Rust ai-gateway does LLM inference (realtime WebSocket). Spend tracking is an +API callback: it POSTs each finished session to the LiteLLM proxy, which records +spend and runs the usual callbacks. + +```mermaid +flowchart LR + C[client] <--> G[Rust ai-gateway
LLM inference] + G <--> O[OpenAI realtime] + G -. spend tracking callback .-> P[litellm proxy] +``` diff --git a/litellm-rust/crates/ai-gateway/Cargo.toml b/litellm-rust/crates/ai-gateway/Cargo.toml index c829333a4b19..b08fb89d5e87 100644 --- a/litellm-rust/crates/ai-gateway/Cargo.toml +++ b/litellm-rust/crates/ai-gateway/Cargo.toml @@ -15,19 +15,25 @@ required-features = ["server"] [dependencies] litellm-core.workspace = true +# reqwest (rustls + json) is used by io/ocr and ships realtime logs to the +# Python proxy callbacks API. reqwest.workspace = true -tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "time"] } +# `sync` powers the bounded mpsc channel the realtime logger drains. +tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "time", "sync"] } tokio-tungstenite.workspace = true futures-util.workspace = true serde_json.workspace = true axum = { workspace = true, features = ["ws"], optional = true } serde = { workspace = true, optional = true } subtle = { workspace = true, optional = true } +# sha2 hashes the master key into user_api_key_hash (matches the proxy's +# SHA-256 hash_token) so the plaintext credential never enters a log payload. +sha2 = { workspace = true, optional = true } pyo3 = { workspace = true, features = ["auto-initialize"], optional = true } [features] default = [] -server = ["dep:axum", "dep:subtle", "dep:serde"] +server = ["dep:axum", "dep:subtle", "dep:serde", "dep:sha2"] # Build the gateway's config from the proxy YAML via an embedded Python # interpreter (links libpython; requires `litellm` importable at runtime). python-config = ["dep:pyo3"] diff --git a/litellm-rust/crates/ai-gateway/README.md b/litellm-rust/crates/ai-gateway/README.md index 649b9b07942e..f913beff6d56 100644 --- a/litellm-rust/crates/ai-gateway/README.md +++ b/litellm-rust/crates/ai-gateway/README.md @@ -19,6 +19,7 @@ Dependency direction (acyclic): litellm-core ← litellm-ai-gateway ← litellm- - **Client endpoint:** `wss:///v1/realtime?model=` (WebSocket) - **Auth:** `Authorization: Bearer $LITELLM_MASTER_KEY` (fails closed if unset) - **Health:** `GET /health/readiness`, `GET /health/liveness`, `GET /health/gil` +- **Request logs:** POSTed to a LiteLLM proxy at `/v1/rust_control_plane/logs` (see [Request logging](#request-logging)) > **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. @@ -65,6 +66,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` | LiteLLM proxy that request logs are POSTed to. See [Request logging](#request-logging). | > Secrets (`LITELLM_MASTER_KEY`, `OPENAI_API_KEY`) are never baked into the image > or `render.yaml` — inject them at deploy time only. @@ -83,6 +85,18 @@ 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 + +The gateway runs no spend logic. When a session ends it builds one +`StandardLoggingPayload` and POSTs it to `{LITELLM_PROXY_BASE_URL}/v1/rust_control_plane/logs` +(admin-only, bearer = `LITELLM_MASTER_KEY`), and the proxy replays it through its +normal callbacks (spend logs, Langfuse, etc.). The POST is non-blocking: a bounded +channel drained by a background worker, dropping with a counter if the proxy is +down. It sends one payload per session. Both env vars are in the table above. + +Worker tuning, rarely needed: `LITELLM_LOG_CHANNEL_CAPACITY` (4096), +`LITELLM_LOG_BATCH_SIZE` (256), `LITELLM_LOG_FLUSH_INTERVAL_MS` (500). + ## Build & run with Docker The image is built `--features python-config` and installs litellm **from this diff --git a/litellm-rust/crates/ai-gateway/src/auth/mod.rs b/litellm-rust/crates/ai-gateway/src/auth/mod.rs index e2dd51f656d3..438a05130572 100644 --- a/litellm-rust/crates/ai-gateway/src/auth/mod.rs +++ b/litellm-rust/crates/ai-gateway/src/auth/mod.rs @@ -12,10 +12,29 @@ use axum::extract::FromRequestParts; use axum::http::header::AUTHORIZATION; use axum::http::request::Parts; use axum::http::StatusCode; +use sha2::{Digest, Sha256}; use subtle::ConstantTimeEq; use crate::state::AppState; +/// SHA-256 hex digest of a token — the exact transform the Python proxy applies +/// (`litellm.proxy.utils.hash_token`). +/// +/// STRICT REQUIREMENT: a raw key (`LITELLM_MASTER_KEY`, a virtual key, …) must +/// **never** leave this gateway in a log payload. Spend logs and every callback +/// integration receive `user_api_key_hash`, so that field must be this hash, not +/// the credential. Hashing here also means the value matches the key's hash in +/// `LiteLLM_SpendLogs.api_key`, so realtime spend joins with the rest of LiteLLM. +pub fn hash_token(token: &str) -> String { + let digest = Sha256::digest(token.as_bytes()); + let mut hex = String::with_capacity(digest.len() * 2); + for byte in digest { + use std::fmt::Write; + let _ = write!(hex, "{byte:02x}"); + } + hex +} + /// Extractor that requires the configured master key as a bearer token. /// /// Rejections: `500` when no master key is configured (permanent @@ -52,3 +71,23 @@ impl FromRequestParts for RequireMasterKey { } } } + +#[cfg(test)] +mod tests { + use super::hash_token; + + #[test] + fn hash_token_matches_python_sha256_hexdigest() { + // Must equal hashlib.sha256("sk-1234".encode()).hexdigest() — the value + // the proxy stores in LiteLLM_SpendLogs.api_key. + assert_eq!( + hash_token("sk-1234"), + "88dc28d0f030c55ed4ab77ed8faf098196cb1c05df778539800c9f1243fe6b4b" + ); + // 64 lowercase hex chars, and never the raw input. + let h = hash_token("sk-secret"); + assert_eq!(h.len(), 64); + assert!(h.chars().all(|c| c.is_ascii_hexdigit())); + assert_ne!(h, "sk-secret"); + } +} diff --git a/litellm-rust/crates/ai-gateway/src/constants.rs b/litellm-rust/crates/ai-gateway/src/constants.rs new file mode 100644 index 000000000000..3116a4c99326 --- /dev/null +++ b/litellm-rust/crates/ai-gateway/src/constants.rs @@ -0,0 +1,29 @@ +//! Crate-level constants for the ai-gateway. +//! +//! Per `litellm-rust/CLAUDE.md`, magic numbers and fixed strings live here +//! (the Rust mirror of Python's `litellm/constants.py`), not inline in feature +//! modules. Env-overridable tunables keep their `DEFAULT_*` value here; the env +//! read + fallback happens at the host/config layer. + +/// Default LiteLLM control-plane base URL for request-log egress when +/// `LITELLM_PROXY_BASE_URL` is unset. +pub(crate) const DEFAULT_PROXY_BASE_URL: &str = "http://localhost:4000"; + +/// The logs ingest path appended to the proxy base. Not a tunable; it is the +/// proxy's API contract (the rust-control-plane router on the Python proxy). +pub(crate) const RUST_CONTROL_PLANE_LOGS_PATH: &str = "/v1/rust_control_plane/logs"; + +/// Default bounded channel depth for the log-egress worker. +/// Override: `LITELLM_LOG_CHANNEL_CAPACITY`. +pub(crate) const DEFAULT_CHANNEL_CAPACITY: usize = 4096; + +/// Default max records POSTed per request to the control plane. +/// Override: `LITELLM_LOG_BATCH_SIZE`. +pub(crate) const DEFAULT_MAX_BATCH_SIZE: usize = 256; + +/// Default partial-batch flush cadence, in ms. +/// Override: `LITELLM_LOG_FLUSH_INTERVAL_MS`. +pub(crate) const DEFAULT_FLUSH_INTERVAL_MS: u64 = 500; + +/// Provider attributed to realtime sessions in the logging payload. +pub(crate) const DEFAULT_PROVIDER: &str = "openai"; diff --git a/litellm-rust/crates/ai-gateway/src/integrations/custom_logger.rs b/litellm-rust/crates/ai-gateway/src/integrations/custom_logger.rs new file mode 100644 index 000000000000..53b599d8c983 --- /dev/null +++ b/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(()) + } +} diff --git a/litellm-rust/crates/ai-gateway/src/integrations/litellm_python_proxy_api.rs b/litellm-rust/crates/ai-gateway/src/integrations/litellm_python_proxy_api.rs new file mode 100644 index 000000000000..165a90d8dbee --- /dev/null +++ b/litellm-rust/crates/ai-gateway/src/integrations/litellm_python_proxy_api.rs @@ -0,0 +1,209 @@ +//! A `CustomLogger` that ships finished events to the LiteLLM Python proxy's +//! `/v1/rust_control_plane/logs` endpoint. +//! +//! The callback path is non-blocking: `log_success_event` / `log_failure_event` +//! build a `LogRecord` and `try_send` it onto a bounded channel, returning a +//! `LogError` (never panicking, never awaiting) if the channel is full or the +//! worker has gone away. A spawned background worker drains the channel, batches +//! records into `{"records":[...]}`, and POSTs them to the proxy with a pooled +//! `reqwest::Client`. + +use std::sync::Arc; +use std::time::Duration; + +use reqwest::Client; +use tokio::sync::mpsc::{self, Receiver, Sender}; +use tokio::time::interval; + +use crate::constants::{ + DEFAULT_CHANNEL_CAPACITY, DEFAULT_FLUSH_INTERVAL_MS, DEFAULT_MAX_BATCH_SIZE, + DEFAULT_PROXY_BASE_URL, RUST_CONTROL_PLANE_LOGS_PATH, +}; +use crate::integrations::custom_logger::CustomLogger; +use crate::integrations::types::{ + CallbackLogsRequest, LogError, LogRecord, LoggingError, StandardLoggingPayload, +}; + +/// Egress worker tunables. Each field defaults to the matching `DEFAULT_*` const +/// in `crate::constants` and is overridable via an env var (read once at logger +/// construction). +struct EgressTunables { + channel_capacity: usize, + max_batch_size: usize, + flush_interval: Duration, +} + +impl EgressTunables { + fn from_env() -> Self { + Self { + channel_capacity: env_positive( + "LITELLM_LOG_CHANNEL_CAPACITY", + DEFAULT_CHANNEL_CAPACITY, + ), + max_batch_size: env_positive("LITELLM_LOG_BATCH_SIZE", DEFAULT_MAX_BATCH_SIZE), + flush_interval: Duration::from_millis(env_positive( + "LITELLM_LOG_FLUSH_INTERVAL_MS", + DEFAULT_FLUSH_INTERVAL_MS, + )), + } + } +} + +/// Parse a positive integer env var, falling back to `default` on missing, +/// unparseable, or non-positive values. Generic over the integer type so one +/// helper serves both the `usize` capacities and the `u64` interval. +fn env_positive(name: &str, default: T) -> T +where + T: std::str::FromStr + PartialOrd + From, +{ + let zero = T::from(0u8); + std::env::var(name) + .ok() + .and_then(|value| value.trim().parse::().ok()) + .filter(|n| *n > zero) + .unwrap_or(default) +} + +/// Ships realtime logging events to the LiteLLM Python proxy. +pub struct LiteLLMPythonProxyAPILogger { + sink: Sender, +} + +impl LiteLLMPythonProxyAPILogger { + /// Spawn the background worker and return a logger handle. `base` is the + /// proxy base URL (no trailing path); `master_key` is sent as a bearer token. + pub fn start(base: String, master_key: String) -> Arc { + let tunables = EgressTunables::from_env(); + let (sink, receiver) = mpsc::channel::(tunables.channel_capacity); + let url = format!( + "{}{}", + base.trim_end_matches('/'), + RUST_CONTROL_PLANE_LOGS_PATH + ); + let client = Client::new(); + tokio::spawn(worker_loop( + receiver, + client, + url, + master_key, + tunables.max_batch_size, + tunables.flush_interval, + )); + Arc::new(Self { sink }) + } + + /// Build a logger from the environment: `LITELLM_PROXY_BASE_URL` (default + /// `http://localhost:4000`) and `LITELLM_MASTER_KEY`. + /// + /// `LITELLM_PROXY_BASE_URL` is treated as the full base and the route is + /// appended verbatim, so if the proxy runs under a `SERVER_ROOT_PATH` + /// (e.g. served at `https://host/litellm`), include it in the base + /// (`LITELLM_PROXY_BASE_URL=https://host/litellm`) and the POST lands at + /// `https://host/litellm/v1/rust_control_plane/logs`. + pub fn from_env() -> Arc { + let base = std::env::var("LITELLM_PROXY_BASE_URL") + .ok() + .filter(|value| !value.trim().is_empty()) + .unwrap_or_else(|| DEFAULT_PROXY_BASE_URL.to_string()); + let key = std::env::var("LITELLM_MASTER_KEY").unwrap_or_default(); + Self::start(base, key) + } + + fn enqueue(&self, record: LogRecord) -> Result<(), LogError> { + self.sink.try_send(record).map_err(|err| match err { + mpsc::error::TrySendError::Full(_) => LogError::channel_full(), + mpsc::error::TrySendError::Closed(_) => LogError::channel_closed(), + }) + } +} + +impl CustomLogger for LiteLLMPythonProxyAPILogger { + fn log_success_event(&self, payload: &StandardLoggingPayload) -> Result<(), LogError> { + self.enqueue(LogRecord { + status: "success".to_string(), + payload: payload.clone(), + error: None, + }) + } + + fn log_failure_event( + &self, + payload: &StandardLoggingPayload, + error: &LoggingError, + ) -> Result<(), LogError> { + self.enqueue(LogRecord { + status: "failure".to_string(), + payload: payload.clone(), + error: Some(format!("{}: {}", error.kind, error.message)), + }) + } +} + +/// Drain the channel, batching records and POSTing them to the proxy. Exits when +/// the channel is closed (all senders dropped) and drained. +async fn worker_loop( + mut receiver: Receiver, + client: Client, + url: String, + master_key: String, + max_batch_size: usize, + flush_interval: Duration, +) { + let mut ticker = interval(flush_interval); + let mut batch: Vec = Vec::with_capacity(max_batch_size); + + loop { + tokio::select! { + maybe_record = receiver.recv() => { + match maybe_record { + Some(record) => { + batch.push(record); + if batch.len() >= max_batch_size { + flush(&client, &url, &master_key, &mut batch).await; + } + } + None => { + // Channel closed: flush remaining and exit. + flush(&client, &url, &master_key, &mut batch).await; + break; + } + } + } + _ = ticker.tick() => { + flush(&client, &url, &master_key, &mut batch).await; + } + } + } +} + +/// POST the current batch (if any), clearing it. Errors are logged, not fatal. +async fn flush(client: &Client, url: &str, master_key: &str, batch: &mut Vec) { + if batch.is_empty() { + return; + } + let records = std::mem::take(batch) + .into_iter() + .map(LogRecord::into_callback_record) + .collect(); + let body = CallbackLogsRequest { records }; + + let response = client + .post(url) + .bearer_auth(master_key) + .json(&body) + .send() + .await; + + match response { + Ok(resp) if resp.status().is_success() => {} + Ok(resp) => { + eprintln!( + "litellm-ai-gateway: callback logs POST returned {} to {url}", + resp.status() + ); + } + Err(err) => { + eprintln!("litellm-ai-gateway: callback logs POST failed to {url}: {err}"); + } + } +} diff --git a/litellm-rust/crates/ai-gateway/src/integrations/mod.rs b/litellm-rust/crates/ai-gateway/src/integrations/mod.rs new file mode 100644 index 000000000000..8799be0c0402 --- /dev/null +++ b/litellm-rust/crates/ai-gateway/src/integrations/mod.rs @@ -0,0 +1,10 @@ +//! Pure-Rust logging integrations. Names map 1:1 to Python +//! `litellm/integrations/`: +//! - [`custom_logger::CustomLogger`] — the callback trait +//! - [`litellm_python_proxy_api::LiteLLMPythonProxyAPILogger`] — ships events +//! to the Python proxy's `/v1/callbacks/logs` endpoint +//! - [`types`] — the typed `StandardLoggingPayload` wire contract + +pub mod custom_logger; +pub mod litellm_python_proxy_api; +pub mod types; diff --git a/litellm-rust/crates/ai-gateway/src/integrations/types.rs b/litellm-rust/crates/ai-gateway/src/integrations/types.rs new file mode 100644 index 000000000000..d61a1f816a7c --- /dev/null +++ b/litellm-rust/crates/ai-gateway/src/integrations/types.rs @@ -0,0 +1,164 @@ +//! Typed payloads for the LiteLLM `/v1/callbacks/logs` realtime-logging contract. +//! +//! Field names below are the EXACT JSON keys the Python replay path + spend-logs +//! builder read. Note the deliberate mix: +//! - `startTime` / `endTime` are camelCase (epoch f64 seconds) +//! - `response_cost` / `prompt_tokens` / etc. are snake_case +//! +//! Mirrors Python `litellm/integrations/` + the proxy `CallbackLogsRequest` +//! contract 1:1. + +use serde::Serialize; +use serde_json::Value; +use std::collections::HashMap; + +/// Cumulative token usage for a realtime session. +#[derive(Clone, Copy, Debug, Default, PartialEq, Eq)] +pub struct Usage { + pub prompt_tokens: u64, + pub completion_tokens: u64, + pub total_tokens: u64, +} + +/// Cost-attribution metadata threaded from the authenticated request. +#[derive(Clone, Debug, Default)] +pub struct RequestMetadata { + pub user_api_key_hash: Option, + pub user_api_key_user_id: Option, + pub user_api_key_team_id: Option, +} + +/// A logging-callback failure (e.g. a custom logger raised). Mirrors the Python +/// failure-event shape: a message plus an exception kind/class name. +#[derive(Clone, Debug)] +pub struct LoggingError { + pub message: String, + pub kind: String, +} + +/// A non-fatal error returned by a `CustomLogger` when it cannot enqueue an +/// event (channel full or the background worker has shut down). +#[derive(Clone, Debug)] +pub struct LogError { + pub message: String, + pub kind: String, +} + +impl LogError { + pub fn channel_full() -> Self { + Self { + message: "logging channel is full; dropping record".to_string(), + kind: "ChannelFull".to_string(), + } + } + + pub fn channel_closed() -> Self { + Self { + message: "logging channel is closed; worker has shut down".to_string(), + kind: "ChannelClosed".to_string(), + } + } +} + +impl std::fmt::Display for LogError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "{}: {}", self.kind, self.message) + } +} + +impl std::error::Error for LogError {} + +/// Batch wrapper — the top-level request body. +/// Matches Python `CallbackLogsRequest { records: list[CallbackLogRecord] }`. +#[derive(Serialize)] +pub struct CallbackLogsRequest { + pub records: Vec, +} + +/// One finished logging event. +/// Matches `CallbackLogRecord { status, standard_logging_payload, error? }`. +#[derive(Serialize)] +pub struct CallbackLogRecord { + /// "success" | "failure". On "failure", `error` (or payload.error_str) + /// becomes the replayed exception string. + pub status: String, + + pub standard_logging_payload: StandardLoggingPayload, + + /// Only meaningful when status == "failure". Omitted on success. + #[serde(skip_serializing_if = "Option::is_none")] + pub error: Option, +} + +/// The self-describing payload. Field names are the EXACT JSON keys the Python +/// replay path + spend-logs builder read. +#[derive(Clone, Debug, Serialize)] +pub struct StandardLoggingPayload { + pub id: String, + pub litellm_call_id: String, + + /// e.g. "realtime", "acompletion". Falls back to "acompletion" if absent. + pub call_type: String, + + pub model: String, + pub custom_llm_provider: String, + + /// Spend ($) written to LiteLLM_SpendLogs.spend. + pub response_cost: f64, + + pub prompt_tokens: u64, + pub completion_tokens: u64, + pub total_tokens: u64, + + /// EPOCH SECONDS as float — camelCase keys, NOT snake_case. + #[serde(rename = "startTime")] + pub start_time: f64, + #[serde(rename = "endTime")] + pub end_time: f64, + + pub stream: bool, + + pub metadata: StandardLoggingMetadata, + + /// Optional; stored as request input on the spend log row. + #[serde(skip_serializing_if = "Option::is_none")] + pub messages: Option, +} + +/// Cost-attribution keys. The replayer maps these into litellm_params.metadata, +/// which the spend-logs builder reads to set user / team_id / organization_id. +#[derive(Clone, Debug, Serialize, Default)] +pub struct StandardLoggingMetadata { + pub user_api_key_hash: Option, // -> SpendLogs.api_key + pub user_api_key_user_id: Option, // -> SpendLogs.user + pub user_api_key_team_id: Option, // -> SpendLogs.team_id + + // Optional but read by the builder; include when known: + #[serde(skip_serializing_if = "Option::is_none")] + pub user_api_key_alias: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub user_api_key_org_id: Option, // -> SpendLogs.organization_id + #[serde(skip_serializing_if = "Option::is_none")] + pub user_api_key_end_user_id: Option, // -> SpendLogs.end_user + #[serde(skip_serializing_if = "Option::is_none")] + pub spend_logs_metadata: Option>, +} + +/// The unit handed to a `CustomLogger` sink: a finished payload plus its status +/// and (on failure) the replayed error string. +#[derive(Clone, Debug)] +pub struct LogRecord { + pub status: String, + pub payload: StandardLoggingPayload, + pub error: Option, +} + +impl LogRecord { + pub fn into_callback_record(self) -> CallbackLogRecord { + CallbackLogRecord { + status: self.status, + standard_logging_payload: self.payload, + error: self.error, + } + } +} diff --git a/litellm-rust/crates/ai-gateway/src/io/realtime.rs b/litellm-rust/crates/ai-gateway/src/io/realtime.rs index 5d538d95fa47..4047de5cb266 100644 --- a/litellm-rust/crates/ai-gateway/src/io/realtime.rs +++ b/litellm-rust/crates/ai-gateway/src/io/realtime.rs @@ -126,6 +126,9 @@ pub(crate) async fn read_event(upstream_rx: &mut UpstreamRx) -> CoreResult( model: &str, @@ -133,6 +136,7 @@ pub(crate) async fn splice( mut upstream_rx: UpstreamRx, prelude: Option, idle_timeout: Option, + mut observe: impl FnMut(&RealtimeEvent) + Send, mut client_in: In, mut client_out: Out, ) -> CoreResult<()> @@ -165,6 +169,10 @@ where // client -> upstream client_event = client_in.next() => { let Some(event) = client_event else { break }; // client disconnected + // NOTE: do NOT observe client events. session.created / response.done + // (carrying usage) are server→client events; observing the client arm + // would let an authenticated client POST a fabricated response.done and + // inflate its own spend log. Logging observes upstream events only. for outbound in config.transform_realtime_request(&event, model)?.events { let payload = serde_json::to_string(&outbound) .map_err(|err| CoreError::InvalidResponse(err.to_string()))?; @@ -181,6 +189,7 @@ where Message::Text(text) => { let event: RealtimeEvent = serde_json::from_str(&text) .map_err(|err| CoreError::InvalidResponse(err.to_string()))?; + observe(&event); for outbound in config.transform_realtime_response(&event, model)?.events { client_out .send(outbound) @@ -207,11 +216,13 @@ where /// framework-agnostic; the gateway adapts its axum socket to these. This is the /// fresh-dial path: dial, then splice. The pool's warm-handoff path skips the dial /// and calls [`splice`] directly with a buffered `session.created`. +#[allow(clippy::too_many_arguments)] pub async fn realtime( model: &str, api_key: Option<&str>, api_base: Option<&str>, idle_timeout: Option, + observe: impl FnMut(&RealtimeEvent) + Send, client_in: In, client_out: Out, ) -> CoreResult<()> @@ -229,6 +240,7 @@ where upstream_rx, None, idle_timeout, + observe, client_in, client_out, ) @@ -238,10 +250,12 @@ where /// Splice a pre-warmed upstream (taken from [`crate::io::realtime_pool`]) to the /// client. Relays the buffered `session.created` first, then splices exactly like /// the fresh-dial path — so a warm session is indistinguishable from a fresh one. +#[allow(clippy::too_many_arguments)] pub async fn realtime_warm( model: &str, handoff: crate::io::realtime_pool::WarmHandoff, idle_timeout: Option, + observe: impl FnMut(&RealtimeEvent) + Send, client_in: In, client_out: Out, ) -> CoreResult<()> @@ -256,6 +270,7 @@ where handoff.rx, Some(handoff.session_created), idle_timeout, + observe, client_in, client_out, ) @@ -303,6 +318,7 @@ mod tests { Some(&key_owned), None, None, + |_| {}, client_in, client_out, ) diff --git a/litellm-rust/crates/ai-gateway/src/lib.rs b/litellm-rust/crates/ai-gateway/src/lib.rs index a2c228aeb304..6c04fbb7626e 100644 --- a/litellm-rust/crates/ai-gateway/src/lib.rs +++ b/litellm-rust/crates/ai-gateway/src/lib.rs @@ -24,5 +24,15 @@ pub mod routes; #[cfg(feature = "server")] pub mod state; +// Realtime request logging. Only the server serves realtime, so these are +// `server`-gated; `io::realtime` exposes the generic `observe` hook while the +// collector and callback fan-out live here. +#[cfg(feature = "server")] +mod constants; +#[cfg(feature = "server")] +pub mod integrations; +#[cfg(feature = "server")] +mod realtime; + #[cfg(feature = "python-config")] pub mod python; diff --git a/litellm-rust/crates/ai-gateway/src/main.rs b/litellm-rust/crates/ai-gateway/src/main.rs index a0105b3f000a..f9ce97801d39 100644 --- a/litellm-rust/crates/ai-gateway/src/main.rs +++ b/litellm-rust/crates/ai-gateway/src/main.rs @@ -16,6 +16,8 @@ use litellm_ai_gateway::routes; use litellm_ai_gateway::state::AppState; use litellm_core::router::{Deployment, LiteLLMParams, Router}; +use litellm_ai_gateway::integrations::custom_logger::CustomLogger; +use litellm_ai_gateway::integrations::litellm_python_proxy_api::LiteLLMPythonProxyAPILogger; #[cfg(feature = "python-config")] use litellm_ai_gateway::python; @@ -39,6 +41,12 @@ async fn main() { ); } + // Spawn the realtime-logging worker (drains a channel → POSTs batches to the + // Python proxy's /v1/callbacks/logs). Built here so the spawn lands on the + // tokio runtime. `from_env` reads LITELLM_PROXY_BASE_URL + LITELLM_MASTER_KEY. + let proxy_logger = LiteLLMPythonProxyAPILogger::from_env(); + let loggers: Vec> = vec![proxy_logger]; + let router = Arc::new(build_router()); // Build the pre-warmed realtime pool and register each deployment's upstream @@ -62,6 +70,7 @@ async fn main() { let state = AppState { router, master_key, + loggers: Arc::new(loggers), realtime_pool, }; diff --git a/litellm-rust/crates/ai-gateway/src/realtime/mod.rs b/litellm-rust/crates/ai-gateway/src/realtime/mod.rs new file mode 100644 index 000000000000..82be596ba86b --- /dev/null +++ b/litellm-rust/crates/ai-gateway/src/realtime/mod.rs @@ -0,0 +1,4 @@ +//! Realtime logging collector. Observes the realtime event stream and emits a +//! `StandardLoggingPayload` to the registered callbacks on session close. + +pub mod streaming; diff --git a/litellm-rust/crates/ai-gateway/src/realtime/streaming.rs b/litellm-rust/crates/ai-gateway/src/realtime/streaming.rs new file mode 100644 index 000000000000..34c82897808f --- /dev/null +++ b/litellm-rust/crates/ai-gateway/src/realtime/streaming.rs @@ -0,0 +1,352 @@ +//! `RealTimeStreaming` — the realtime logging collector. +//! +//! Mirrors Python `litellm.realtime_api.main.RealTimeStreaming`: it observes the +//! event stream in O(1) (never buffering frames), accumulating just the fields +//! the spend log needs (model, id, cumulative usage), then on session close +//! builds a `StandardLoggingPayload` and fans it out to every registered +//! `CustomLogger`. + +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +use litellm_core::realtime::types::RealtimeEvent; +use serde_json::Value; + +use crate::constants::DEFAULT_PROVIDER; +use crate::integrations::custom_logger::CustomLogger; +use crate::integrations::types::{ + RequestMetadata, StandardLoggingMetadata, StandardLoggingPayload, Usage, +}; + +/// Current wall-clock time as epoch seconds (float), matching the Python +/// `startTime`/`endTime` contract. +fn epoch_seconds() -> f64 { + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_secs_f64()) + .unwrap_or(0.0) +} + +/// Status of a finished realtime session, mapped to the callback record status. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum SessionStatus { + Success, + Failure, +} + +/// Accumulates realtime session state and emits a logging payload on close. +pub struct RealTimeStreaming { + callbacks: Vec>, + /// REQUEST-ID RULE: the SpendLogs `request_id` == the OpenAI realtime session + /// id (`sess_…`), captured from `session.created`. Both `id` and + /// `litellm_call_id` are set to that value so the Python writer logs the same + /// id regardless of which field it reads. The gateway-generated `rt-…` id + /// (the constructor seed) is only a fallback for sessions that fail before + /// `session.created` arrives. + litellm_call_id: String, + /// See the request-id rule above — mirrors `litellm_call_id`. + id: String, + model: String, + custom_llm_provider: String, + usage: Usage, + response_cost: f64, + start_time: f64, + end_time: f64, + metadata: RequestMetadata, + /// Count of logging callbacks that failed to enqueue (non-fatal). + dropped: u64, +} + +impl RealTimeStreaming { + /// Create a collector for one session. `litellm_call_id` is the gateway's + /// per-connection id; `model` is the requested model (a sane default until + /// `session.created` reports the upstream model). + pub fn new( + callbacks: Vec>, + litellm_call_id: String, + model: String, + metadata: RequestMetadata, + ) -> Self { + let now = epoch_seconds(); + Self { + callbacks, + id: litellm_call_id.clone(), + litellm_call_id, + model, + custom_llm_provider: DEFAULT_PROVIDER.to_string(), + usage: Usage::default(), + response_cost: 0.0, + start_time: now, + end_time: now, + metadata, + dropped: 0, + } + } + + /// Number of logging callbacks that failed to enqueue so far (test/observ.). + #[allow(dead_code)] + pub fn dropped(&self) -> u64 { + self.dropped + } + + /// Observe one realtime event. O(1): updates accumulated state only; never + /// buffers frames. Safe to call on every event in either direction. + pub fn observe(&mut self, event: &RealtimeEvent) { + match event.event_type.as_str() { + "session.created" | "session.updated" => self.on_session(event), + "response.done" => self.on_response_done(event), + _ => {} + } + } + + /// `session.created` / `session.updated` → capture upstream id + model. + /// Per the request-id rule, the OpenAI session id becomes BOTH `id` and + /// `litellm_call_id`, replacing the gateway-generated fallback. + fn on_session(&mut self, event: &RealtimeEvent) { + let session = event.data.get("session").and_then(Value::as_object); + if let Some(id) = session.and_then(|s| s.get("id")).and_then(Value::as_str) { + if !id.is_empty() { + self.id = id.to_string(); + self.litellm_call_id = id.to_string(); + } + } + if let Some(model) = session.and_then(|s| s.get("model")).and_then(Value::as_str) { + if !model.is_empty() { + self.model = model.to_string(); + } + } + } + + /// `response.done` → add this response's usage to the cumulative totals. + fn on_response_done(&mut self, event: &RealtimeEvent) { + let usage = event + .data + .get("response") + .and_then(Value::as_object) + .and_then(|r| r.get("usage")) + .and_then(Value::as_object); + let Some(usage) = usage else { return }; + + let input = usage.get("input_tokens").and_then(Value::as_u64); + let output = usage.get("output_tokens").and_then(Value::as_u64); + let total = usage.get("total_tokens").and_then(Value::as_u64); + + if let Some(input) = input { + self.usage.prompt_tokens += input; + } + if let Some(output) = output { + self.usage.completion_tokens += output; + } + // Prefer the upstream-reported total; otherwise derive it. + match total { + Some(total) => self.usage.total_tokens += total, + None => { + self.usage.total_tokens += input.unwrap_or(0) + output.unwrap_or(0); + } + } + } + + /// Set the per-session response cost ($). Cost computation is Python-side in + /// the proxy; the gateway forwards 0.0 by default and lets the proxy price. + /// Public API (exercised in tests) for the future path where the gateway + /// prices realtime sessions itself. + #[allow(dead_code)] + pub fn set_response_cost(&mut self, cost: f64) { + self.response_cost = cost; + } + + /// Build the `StandardLoggingPayload` from accumulated state. + pub fn build_payload(&self) -> StandardLoggingPayload { + StandardLoggingPayload { + id: self.id.clone(), + litellm_call_id: self.litellm_call_id.clone(), + call_type: "realtime".to_string(), + model: self.model.clone(), + custom_llm_provider: self.custom_llm_provider.clone(), + response_cost: self.response_cost, + prompt_tokens: self.usage.prompt_tokens, + completion_tokens: self.usage.completion_tokens, + total_tokens: self.usage.total_tokens, + start_time: self.start_time, + end_time: self.end_time, + stream: true, + metadata: StandardLoggingMetadata { + user_api_key_hash: self.metadata.user_api_key_hash.clone(), + user_api_key_user_id: self.metadata.user_api_key_user_id.clone(), + user_api_key_team_id: self.metadata.user_api_key_team_id.clone(), + ..Default::default() + }, + messages: None, + } + } + + /// Finish the session: stamp the end time and fan the payload out to every + /// callback. On a logger enqueue error we bump a non-fatal counter (the + /// realtime session has already ended; a dropped log must never propagate). + pub fn log_messages(&mut self, status: SessionStatus) { + self.end_time = epoch_seconds(); + let payload = self.build_payload(); + + match status { + SessionStatus::Success => { + for callback in &self.callbacks { + if let Err(err) = callback.log_success_event(&payload) { + self.dropped += 1; + eprintln!("litellm-ai-gateway: log_success_event dropped: {err}"); + } + } + } + SessionStatus::Failure => { + let error = crate::integrations::types::LoggingError { + message: "realtime session ended in failure".to_string(), + kind: "RealtimeSessionError".to_string(), + }; + for callback in &self.callbacks { + if let Err(err) = callback.log_failure_event(&payload, &error) { + self.dropped += 1; + eprintln!("litellm-ai-gateway: log_failure_event dropped: {err}"); + } + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::integrations::types::{LogError, LoggingError}; + use std::sync::atomic::{AtomicU64, Ordering}; + + fn event(raw: &str) -> RealtimeEvent { + serde_json::from_str(raw).expect("valid event json") + } + + /// A test logger that records the last payload it saw. + #[derive(Default)] + struct CapturingLogger { + calls: AtomicU64, + last_model: std::sync::Mutex>, + last_total_tokens: AtomicU64, + } + + impl CustomLogger for CapturingLogger { + fn log_success_event(&self, payload: &StandardLoggingPayload) -> Result<(), LogError> { + self.calls.fetch_add(1, Ordering::SeqCst); + *self.last_model.lock().unwrap() = Some(payload.model.clone()); + self.last_total_tokens + .store(payload.total_tokens, Ordering::SeqCst); + Ok(()) + } + } + + #[test] + fn observe_accumulates_model_and_tokens_then_logs() { + let logger = Arc::new(CapturingLogger::default()); + let callbacks: Vec> = vec![logger.clone()]; + let mut streaming = RealTimeStreaming::new( + callbacks, + "call_abc".to_string(), + "gpt-realtime".to_string(), + RequestMetadata { + user_api_key_hash: Some("hash123".to_string()), + user_api_key_user_id: Some("user-1".to_string()), + user_api_key_team_id: Some("team-1".to_string()), + }, + ); + + streaming.observe(&event( + r#"{"type":"session.created","session":{"id":"sess_001","model":"gpt-realtime-2025"}}"#, + )); + streaming.observe(&event( + r#"{"type":"response.done","response":{"usage":{"input_tokens":10,"output_tokens":5,"total_tokens":15}}}"#, + )); + // A second response.done accumulates. + streaming.observe(&event( + r#"{"type":"response.done","response":{"usage":{"input_tokens":3,"output_tokens":2,"total_tokens":5}}}"#, + )); + + let payload = streaming.build_payload(); + assert_eq!(payload.model, "gpt-realtime-2025"); + // Request-id rule: session.created's id becomes BOTH id and + // litellm_call_id (replacing the "call_abc" gateway fallback), so the + // SpendLogs request_id is always the OpenAI session id. + assert_eq!(payload.id, "sess_001"); + assert_eq!(payload.litellm_call_id, "sess_001"); + assert_eq!(payload.prompt_tokens, 13); + assert_eq!(payload.completion_tokens, 7); + assert_eq!(payload.total_tokens, 20); + assert_eq!(payload.response_cost, 0.0); + assert_eq!(payload.call_type, "realtime"); + assert_eq!(payload.custom_llm_provider, "openai"); + assert_eq!( + payload.metadata.user_api_key_hash.as_deref(), + Some("hash123") + ); + + streaming.log_messages(SessionStatus::Success); + assert_eq!(logger.calls.load(Ordering::SeqCst), 1); + assert_eq!( + logger.last_model.lock().unwrap().as_deref(), + Some("gpt-realtime-2025") + ); + assert_eq!(logger.last_total_tokens.load(Ordering::SeqCst), 20); + assert_eq!(streaming.dropped(), 0); + } + + #[test] + fn payload_serializes_with_camelcase_times_and_realtime_call_type() { + let mut streaming = RealTimeStreaming::new( + Vec::new(), + "call_xyz".to_string(), + "gpt-realtime".to_string(), + RequestMetadata::default(), + ); + streaming.observe(&event( + r#"{"type":"response.done","response":{"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}"#, + )); + streaming.set_response_cost(0.0042); + let payload = streaming.build_payload(); + let json = serde_json::to_string(&payload).expect("serialize payload"); + + assert!(json.contains("\"startTime\""), "missing startTime: {json}"); + assert!(json.contains("\"endTime\""), "missing endTime: {json}"); + assert!( + json.contains("\"call_type\":\"realtime\""), + "missing call_type realtime: {json}" + ); + assert!( + json.contains("\"response_cost\""), + "missing response_cost: {json}" + ); + assert_eq!(payload.response_cost, 0.0042); + } + + /// A logger whose enqueue always fails should bump the dropped counter, not + /// panic or propagate. + #[test] + fn failing_logger_bumps_dropped_counter() { + struct FailingLogger; + impl CustomLogger for FailingLogger { + fn log_success_event(&self, _p: &StandardLoggingPayload) -> Result<(), LogError> { + Err(LogError::channel_full()) + } + fn log_failure_event( + &self, + _p: &StandardLoggingPayload, + _e: &LoggingError, + ) -> Result<(), LogError> { + Err(LogError::channel_closed()) + } + } + let callbacks: Vec> = vec![Arc::new(FailingLogger)]; + let mut streaming = RealTimeStreaming::new( + callbacks, + "call_1".to_string(), + "gpt-realtime".to_string(), + RequestMetadata::default(), + ); + streaming.log_messages(SessionStatus::Success); + assert_eq!(streaming.dropped(), 1); + } +} diff --git a/litellm-rust/crates/ai-gateway/src/routes/realtime/mod.rs b/litellm-rust/crates/ai-gateway/src/routes/realtime/mod.rs index 658fa8d2dbee..899ad73829f2 100644 --- a/litellm-rust/crates/ai-gateway/src/routes/realtime/mod.rs +++ b/litellm-rust/crates/ai-gateway/src/routes/realtime/mod.rs @@ -6,7 +6,9 @@ mod service; +use std::sync::atomic::{AtomicU64, Ordering}; use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; use crate::io::realtime_pool::RealtimePool; use axum::extract::ws::{Message, WebSocket, WebSocketUpgrade}; @@ -21,8 +23,26 @@ use litellm_core::router::Router as ModelRouter; use serde::Deserialize; use crate::auth::RequireMasterKey; +use crate::integrations::custom_logger::CustomLogger; +use crate::integrations::types::RequestMetadata; +use crate::realtime::streaming::{RealTimeStreaming, SessionStatus}; use crate::state::AppState; +/// Process-local monotonic counter, mixed into the per-session call id so two +/// sessions opened in the same nanosecond still get distinct ids. +static CALL_SEQ: AtomicU64 = AtomicU64::new(0); + +/// Generate a per-connection `litellm_call_id`. No external uuid dep: epoch +/// nanos + a process-local sequence is unique enough for log correlation. +fn new_call_id() -> String { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + let seq = CALL_SEQ.fetch_add(1, Ordering::Relaxed); + format!("rt-{nanos:x}-{seq:x}") +} + /// This route's contribution to the app router. pub fn router() -> Router { Router::new().route("/v1/realtime", get(handle)) @@ -57,26 +77,63 @@ async fn handle( let router = state.router.clone(); let pool = state.realtime_pool.clone(); + let loggers = state.loggers.clone(); + let master_key = state.master_key.clone(); let model = query.model; - Ok(ws.on_upgrade(move |socket| bridge(socket, router, pool, model))) + Ok(ws.on_upgrade(move |socket| bridge(socket, router, pool, loggers, master_key, model))) } /// Adapt the axum socket (text frames) to the typed-event `Stream`/`Sink` the /// service wants, keeping axum types out of `service`. +/// +/// This is also the realtime-logging seam: every upstream→client event (the +/// direction carrying `session.created` and `response.done` with usage) is fed +/// to a [`RealTimeStreaming`] collector via the splice's `observe` callback. The +/// observe is O(1) and never buffers frames. When the splice returns (any of the +/// three break paths — client disconnect, upstream close, idle timeout), we flush +/// one logging payload to the registered callbacks. async fn bridge( socket: WebSocket, router: Arc, pool: Arc, + loggers: Arc>>, + master_key: Option>, model: String, ) { let (ws_sink, ws_stream) = socket.split(); + // Attribute the spend log to the key that authenticated this session (the + // master key — the gateway is master-key auth). A non-null user_api_key_hash + // is required for the Python spend logger to write a SpendLogs row. + // + // SECURITY: hash the key — never send the raw credential. This field fans out + // to spend logs and every callback integration; the SHA-256 (matching the + // proxy's hash_token) keeps the plaintext master key out of all of them while + // still matching the key's hash in LiteLLM_SpendLogs. + let metadata = RequestMetadata { + user_api_key_hash: master_key.as_deref().map(crate::auth::hash_token), + ..RequestMetadata::default() + }; + + // Owned by THIS task only. The splice observes it via a synchronous `&mut` + // callback (below), so there is no Arc/Mutex/atomic on the per-frame hot + // path — just a monomorphized FnMut mutating stack-local fields. This is + // what lets observe scale: 10K concurrent sessions = 10K independent + // collectors, zero cross-task synchronization. + let mut collector = RealTimeStreaming::new( + loggers.as_ref().clone(), + new_call_id(), + model.clone(), + metadata, + ); + let client_in = ws_stream.filter_map(|message| async move { match message { Ok(Message::Text(text)) => serde_json::from_str::(&text).ok(), _ => None, } }); + // Plain forwarding sink — no observe here anymore. let client_out = ws_sink.with(|event: RealtimeEvent| async move { Ok::(Message::Text( serde_json::to_string(&event).unwrap_or_default(), @@ -84,5 +141,26 @@ async fn bridge( }); futures_util::pin_mut!(client_in, client_out); - let _ = service::run(&router, &pool, &model, None, client_in, client_out).await; + + // The observe closure borrows `&mut collector` for the duration of the + // splice; the borrow ends when `run` returns, freeing the collector for the + // single post-session `log_messages` flush. `run` picks a pooled (warm) or + // fresh upstream — observe fires on the upstream arm either way. + let result = service::run( + &router, + &pool, + &model, + None, + |event: &RealtimeEvent| collector.observe(event), + client_in, + client_out, + ) + .await; + + let status = if result.is_ok() { + SessionStatus::Success + } else { + SessionStatus::Failure + }; + collector.log_messages(status); } diff --git a/litellm-rust/crates/ai-gateway/src/routes/realtime/service.rs b/litellm-rust/crates/ai-gateway/src/routes/realtime/service.rs index c78ca8df446d..d6c31edd4548 100644 --- a/litellm-rust/crates/ai-gateway/src/routes/realtime/service.rs +++ b/litellm-rust/crates/ai-gateway/src/routes/realtime/service.rs @@ -26,6 +26,7 @@ pub async fn run( pool: &RealtimePool, model: &str, idle_timeout: Option, + observe: impl FnMut(&RealtimeEvent) + Send, client_in: In, client_out: Out, ) -> CoreResult<()> @@ -56,6 +57,7 @@ where provider_model, handoff, idle_timeout, + observe, client_in, client_out, ) @@ -69,6 +71,7 @@ where params.api_key.as_deref(), params.api_base.as_deref(), idle_timeout, + observe, client_in, client_out, ) diff --git a/litellm-rust/crates/ai-gateway/src/state.rs b/litellm-rust/crates/ai-gateway/src/state.rs index c7ba92d9cbce..3b61d8309ea9 100644 --- a/litellm-rust/crates/ai-gateway/src/state.rs +++ b/litellm-rust/crates/ai-gateway/src/state.rs @@ -3,6 +3,8 @@ use std::sync::Arc; use crate::io::realtime_pool::RealtimePool; use litellm_core::router::Router; +use crate::integrations::custom_logger::CustomLogger; + /// Shared application state handed to every route handler. #[derive(Clone)] pub struct AppState { @@ -10,6 +12,8 @@ pub struct AppState { /// The gateway master key. Any caller presenting it as a bearer token may /// invoke the gateway. `None` → auth not configured (routes fail closed). pub master_key: Option>, + /// Logging callbacks fanned out at the end of each realtime session. + pub loggers: Arc>>, /// Pre-warmed upstream realtime connection pool. Disabled /// (`RealtimePool::disabled()`) when `REALTIME_POOL_SIZE=0`, in which case /// every realtime connect fresh-dials exactly as before. diff --git a/litellm/constants.py b/litellm/constants.py index 212d34357f81..09235106c63b 100644 --- a/litellm/constants.py +++ b/litellm/constants.py @@ -30,6 +30,9 @@ SQS_SEND_MESSAGE_ACTION = "SendMessage" SQS_API_VERSION = "2012-11-05" DEFAULT_MAX_RETRIES = int(os.getenv("DEFAULT_MAX_RETRIES", 2)) +# Max records accepted in one POST /v1/callbacks/logs batch. Bounds the blast +# radius: each record fans out to spend logs + every callback integration. +MAX_CALLBACK_LOG_RECORDS = 1000 DEFAULT_MAX_RECURSE_DEPTH = int(os.getenv("DEFAULT_MAX_RECURSE_DEPTH", 100)) DEFAULT_MAX_RECURSE_DEPTH_SENSITIVE_DATA_MASKER = int( os.getenv("DEFAULT_MAX_RECURSE_DEPTH_SENSITIVE_DATA_MASKER", 10) diff --git a/litellm/proxy/logging_endpoints/__init__.py b/litellm/proxy/logging_endpoints/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/litellm/proxy/logging_endpoints/callback_logs_endpoints.py b/litellm/proxy/logging_endpoints/callback_logs_endpoints.py new file mode 100644 index 000000000000..a96a5431294c --- /dev/null +++ b/litellm/proxy/logging_endpoints/callback_logs_endpoints.py @@ -0,0 +1,208 @@ +""" +Ingest pre-built logging payloads from external producers and replay them +through LiteLLM's standard success/failure callback fan-out. + +This exists for hosts that own a request outside the Python process — e.g. the +`litellm-rust` gateway proxying realtime websockets. Those hosts can't use the +in-process logging object, so they POST a finished `StandardLoggingPayload` here +and Python replays it through the exact same path a normal completion uses: +`Logging.async_success_handler` / `async_failure_handler`. Every registered +callback (spend logs, Langfuse, Datadog, ...) fires unchanged — there is no +spend-logs-specific or callback-specific code here, only the replay. + +The endpoint is generic: realtime is the first producer, but the contract is the +self-describing `StandardLoggingPayload`, so completions/responses can use it too. +""" + +import uuid +from datetime import datetime, timezone +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException + +from litellm._logging import verbose_proxy_logger +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging +from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth +from litellm.proxy.auth.user_api_key_auth import user_api_key_auth +from litellm.types.proxy.callback_logs_endpoints import ( + CallbackLogFailure, + CallbackLogRecord, + CallbackLogsRequest, + CallbackLogsResponse, +) + +# Routes the Python proxy exposes for the Rust data-plane gateway to call into +# (logging today; auth/budgets later). Namespaced under /v1/rust_control_plane so +# they're clearly distinct from the proxy's own control-plane/management routes. +rust_control_plane_router = APIRouter( + prefix="/v1/rust_control_plane", tags=["rust control plane"] +) + + +class CallbackLogsReplayer: + """ + Replays finished logging payloads through LiteLLM's callback fan-out. + + Each helper is small and pure so the replay path is easy to read and test: + rebuild a `Logging` object from the payload, seed `model_call_details` with + exactly what the callbacks read, then dispatch to the success/failure + handler. No spend/callback logic lives here — only the replay. + """ + + @staticmethod + def _epoch_to_datetime(value: Any) -> datetime: + """`StandardLoggingPayload` stores startTime/endTime as float epoch seconds.""" + if isinstance(value, (int, float)): + return datetime.fromtimestamp(float(value), tz=timezone.utc) + if isinstance(value, datetime): + return value + return datetime.now(tz=timezone.utc) + + @staticmethod + def _build_logging_obj(payload: dict[str, Any]) -> LiteLLMLogging: + """ + Reconstruct a `Logging` object from a finished payload and seed + `model_call_details` with exactly what the success/failure callbacks + read: the prebuilt `standard_logging_object`, the resolved + `response_cost`, and the `litellm_params.metadata` keys used for cost + attribution. Setting `standard_logging_object` up front makes the handler + skip rebuilding it. + """ + model = payload.get("model") or "" + call_type = payload.get("call_type") or "acompletion" + start_time = CallbackLogsReplayer._epoch_to_datetime(payload.get("startTime")) + call_id = ( + payload.get("litellm_call_id") or payload.get("id") or str(uuid.uuid4()) + ) + + logging_obj = LiteLLMLogging( + model=model, + messages=payload.get("messages") or [], + # A replayed payload is always a *terminal*, fully-aggregated event — + # the producer (e.g. the rust gateway) already collected the whole + # session before POSTing. Never mark it streaming: a streaming + # Logging object makes async_success_handler wait for a + # complete_streaming_response that will never arrive, so the spend + # log is never written. + stream=False, + call_type=call_type, + start_time=start_time, + litellm_call_id=call_id, + function_id="", + ) + + metadata: dict[str, Any] = payload.get("metadata") or {} + litellm_metadata: dict[str, Any] = { + "user_api_key": metadata.get("user_api_key_hash"), + "user_api_key_alias": metadata.get("user_api_key_alias"), + "user_api_key_user_id": metadata.get("user_api_key_user_id"), + "user_api_key_team_id": metadata.get("user_api_key_team_id"), + "user_api_key_org_id": metadata.get("user_api_key_org_id"), + "user_api_key_end_user_id": metadata.get("user_api_key_end_user_id"), + "spend_logs_metadata": metadata.get("spend_logs_metadata"), + } + + logging_obj.model_call_details.update( + { + "model": model, + "call_type": call_type, + "custom_llm_provider": payload.get("custom_llm_provider"), + "response_cost": payload.get("response_cost") or 0.0, + "standard_logging_object": payload, + "litellm_params": {"metadata": litellm_metadata}, + "cache_hit": payload.get("cache_hit") or False, + } + ) + return logging_obj + + @staticmethod + def _response_obj_from_payload(payload: dict[str, Any]) -> dict[str, Any]: + """Minimal response object so usage-derived spend-log fields resolve.""" + return { + "id": payload.get("id"), + "usage": { + "prompt_tokens": payload.get("prompt_tokens", 0), + "completion_tokens": payload.get("completion_tokens", 0), + "total_tokens": payload.get("total_tokens", 0), + }, + } + + async def replay(self, record: CallbackLogRecord) -> None: + """Replay one record through the matching success/failure handler.""" + payload = record.standard_logging_payload + verbose_proxy_logger.debug( + "CallbackLogsReplayer: replaying %s record id=%s model=%s call_type=%s", + record.status, + payload.get("id"), + payload.get("model"), + payload.get("call_type"), + ) + + logging_obj = self._build_logging_obj(payload) + start_time = self._epoch_to_datetime(payload.get("startTime")) + end_time = self._epoch_to_datetime(payload.get("endTime")) + + if record.status == "success": + await logging_obj.async_success_handler( + result=self._response_obj_from_payload(payload), + start_time=start_time, + end_time=end_time, + ) + else: + error_str = record.error or payload.get("error_str") or "replayed failure" + await logging_obj.async_failure_handler( + Exception(error_str), + traceback_exception="", + start_time=start_time, + end_time=end_time, + ) + + async def replay_batch( + self, records: list[CallbackLogRecord] + ) -> CallbackLogsResponse: + """Replay a batch; a single bad record never sinks the rest. Each failure + is reported back with its batch index so the caller can retry/triage it.""" + processed = 0 + failures: list[CallbackLogFailure] = [] + for index, record in enumerate(records): + try: + await self.replay(record) + processed += 1 + except Exception as e: + failures.append(CallbackLogFailure(index=index, error=str(e))) + verbose_proxy_logger.exception( + "CallbackLogsReplayer: failed to replay record %s: %s", + index, + str(e), + ) + verbose_proxy_logger.debug( + "CallbackLogsReplayer: batch done processed=%s failed=%s", + processed, + len(failures), + ) + return CallbackLogsResponse( + processed=processed, failed=len(failures), failures=failures + ) + + +@rust_control_plane_router.post( + "/logs", + dependencies=[Depends(user_api_key_auth)], +) +async def ingest_callback_logs( + body: CallbackLogsRequest, + user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth), +) -> CallbackLogsResponse: + """ + Replay a batch of finished logging payloads through the callback fan-out. + + Admin-only: the payloads write spend logs and trigger every callback, so this + is a trusted internal route, not a public surface. + """ + if user_api_key_dict.user_role != LitellmUserRoles.PROXY_ADMIN: + raise HTTPException( + status_code=403, + detail="/v1/rust_control_plane/logs is admin-only (proxy admin key required).", + ) + + return await CallbackLogsReplayer().replay_batch(body.records) diff --git a/litellm/proxy/proxy_server.py b/litellm/proxy/proxy_server.py index 4c36b42615e7..50db86c59ef7 100644 --- a/litellm/proxy/proxy_server.py +++ b/litellm/proxy/proxy_server.py @@ -346,6 +346,9 @@ def generate_feedback_box(): from litellm.proxy.hooks.proxy_track_cost_callback import _ProxyDBLogger from litellm.proxy.image_endpoints.endpoints import router as image_router from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request +from litellm.proxy.logging_endpoints.callback_logs_endpoints import ( + rust_control_plane_router, +) from litellm.proxy.management_endpoints.budget_management_endpoints import ( router as budget_management_router, ) @@ -16638,6 +16641,7 @@ async def get_routes(): app.include_router(analytics_router) app.include_router(callback_management_endpoints_router) app.include_router(debugging_endpoints_router) +app.include_router(rust_control_plane_router) app.include_router(ui_crud_endpoints_router) app.include_router(openai_files_router) app.include_router(team_callback_router) diff --git a/litellm/types/proxy/callback_logs_endpoints.py b/litellm/types/proxy/callback_logs_endpoints.py new file mode 100644 index 000000000000..ef148274ca7c --- /dev/null +++ b/litellm/types/proxy/callback_logs_endpoints.py @@ -0,0 +1,44 @@ +""" +Types for the callback-logs ingest endpoint (POST /v1/callbacks/logs). + +External producers (e.g. the litellm-rust gateway) POST finished logging +payloads here; the proxy replays them through the standard callback fan-out. +""" + +from typing import Any, Literal, Optional + +from pydantic import BaseModel, Field + +from litellm.constants import MAX_CALLBACK_LOG_RECORDS + + +class CallbackLogRecord(BaseModel): + """A single finished logging event to replay through the callbacks.""" + + status: Literal["success", "failure"] + standard_logging_payload: dict[str, Any] + error: Optional[str] = None + + +class CallbackLogsRequest(BaseModel): + """A batch of logging events posted by an external producer.""" + + # Bounded so one POST can't trigger an unbounded callback/DB fan-out (each + # record fires every registered integration). Over the cap → 422. + records: list[CallbackLogRecord] = Field(..., max_length=MAX_CALLBACK_LOG_RECORDS) + + +class CallbackLogFailure(BaseModel): + """A record that failed to replay, identified by its index in the batch.""" + + index: int + error: str + + +class CallbackLogsResponse(BaseModel): + """Per-batch result: counts plus per-record failure detail so the caller can + distinguish a transient callback error from a structurally bad payload.""" + + processed: int + failed: int + failures: list[CallbackLogFailure] = Field(default_factory=list) diff --git a/tests/test_litellm/proxy/logging_endpoints/__init__.py b/tests/test_litellm/proxy/logging_endpoints/__init__.py new file mode 100644 index 000000000000..e69de29bb2d1 diff --git a/tests/test_litellm/proxy/logging_endpoints/test_callback_logs_endpoints.py b/tests/test_litellm/proxy/logging_endpoints/test_callback_logs_endpoints.py new file mode 100644 index 000000000000..40e89329b8db --- /dev/null +++ b/tests/test_litellm/proxy/logging_endpoints/test_callback_logs_endpoints.py @@ -0,0 +1,195 @@ +"""Unit tests for POST /v1/callbacks/logs (replay logging payloads → callbacks).""" + +import time + +import pytest +from fastapi import HTTPException + +from litellm.litellm_core_utils.litellm_logging import Logging as LiteLLMLogging +from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth +from litellm.proxy.logging_endpoints.callback_logs_endpoints import ( + CallbackLogsReplayer, + ingest_callback_logs, +) +from litellm.types.proxy.callback_logs_endpoints import ( + CallbackLogRecord, + CallbackLogsRequest, +) + +REQ_ID = "cb-logs-unit-test-1" + + +def _sample_payload(**overrides): + payload = { + "id": REQ_ID, + "litellm_call_id": REQ_ID, + "call_type": "acompletion", + "stream": False, + "response_cost": 0.0123, + "custom_llm_provider": "openai", + "total_tokens": 42, + "prompt_tokens": 30, + "completion_tokens": 12, + "startTime": time.time() - 2, + "endTime": time.time(), + "model": "gpt-4o-mini", + "metadata": { + "user_api_key_hash": "rust-gateway-test-key", + "user_api_key_user_id": "user-cb-logs-test", + "user_api_key_team_id": "team-cb-logs-test", + }, + "messages": [{"role": "user", "content": "hi"}], + } + payload.update(overrides) + return payload + + +def test_epoch_to_datetime_handles_float_and_fallback(): + dt = CallbackLogsReplayer._epoch_to_datetime(1_700_000_000.5) + assert dt.year == 2023 + # Non-numeric input must not raise — falls back to "now". + assert CallbackLogsReplayer._epoch_to_datetime(None) is not None + + +def test_build_logging_obj_seeds_model_call_details(): + obj = CallbackLogsReplayer._build_logging_obj(_sample_payload()) + details = obj.model_call_details + # Prebuilt payload is set so the handler skips rebuilding it. + assert details["standard_logging_object"]["id"] == REQ_ID + assert details["response_cost"] == 0.0123 + assert details["call_type"] == "acompletion" + # Metadata is mapped to the keys the cost-tracking callback reads. + md = details["litellm_params"]["metadata"] + assert md["user_api_key"] == "rust-gateway-test-key" + assert md["user_api_key_user_id"] == "user-cb-logs-test" + assert md["user_api_key_team_id"] == "team-cb-logs-test" + + +def test_response_obj_carries_usage(): + obj = CallbackLogsReplayer._response_obj_from_payload(_sample_payload()) + assert obj["usage"]["total_tokens"] == 42 + assert obj["usage"]["prompt_tokens"] == 30 + assert obj["usage"]["completion_tokens"] == 12 + + +@pytest.mark.asyncio +async def test_success_record_invokes_success_handler(monkeypatch): + captured = {} + + async def fake_success(self, result=None, start_time=None, end_time=None, **kwargs): + captured["standard_logging_object"] = self.model_call_details.get( + "standard_logging_object" + ) + captured["result"] = result + + monkeypatch.setattr(LiteLLMLogging, "async_success_handler", fake_success) + + body = CallbackLogsRequest( + records=[ + CallbackLogRecord( + status="success", standard_logging_payload=_sample_payload() + ) + ] + ) + resp = await ingest_callback_logs( + body, user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN) + ) + assert resp.processed == 1 and resp.failed == 0 + assert captured["standard_logging_object"]["id"] == REQ_ID + assert captured["result"]["usage"]["total_tokens"] == 42 + + +@pytest.mark.asyncio +async def test_failure_record_invokes_failure_handler(monkeypatch): + captured = {} + + async def fake_failure( + self, exception, traceback_exception, start_time=None, end_time=None + ): + captured["exception"] = str(exception) + + monkeypatch.setattr(LiteLLMLogging, "async_failure_handler", fake_failure) + + body = CallbackLogsRequest( + records=[ + CallbackLogRecord( + status="failure", + standard_logging_payload=_sample_payload(), + error="upstream exploded", + ) + ] + ) + resp = await ingest_callback_logs( + body, user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN) + ) + assert resp.processed == 1 and resp.failed == 0 + assert captured["exception"] == "upstream exploded" + + +@pytest.mark.asyncio +async def test_non_admin_is_rejected(monkeypatch): + async def fake_success(self, **kwargs): + return None + + monkeypatch.setattr(LiteLLMLogging, "async_success_handler", fake_success) + + body = CallbackLogsRequest( + records=[ + CallbackLogRecord( + status="success", standard_logging_payload=_sample_payload() + ) + ] + ) + with pytest.raises(HTTPException) as exc_info: + await ingest_callback_logs( + body, + user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER), + ) + assert exc_info.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_one_bad_record_does_not_sink_the_batch(monkeypatch): + calls = {"n": 0} + + async def flaky_success( + self, result=None, start_time=None, end_time=None, **kwargs + ): + calls["n"] += 1 + if calls["n"] == 1: + raise ValueError("boom on first record") + + monkeypatch.setattr(LiteLLMLogging, "async_success_handler", flaky_success) + + body = CallbackLogsRequest( + records=[ + CallbackLogRecord( + status="success", standard_logging_payload=_sample_payload() + ), + CallbackLogRecord( + status="success", standard_logging_payload=_sample_payload() + ), + ] + ) + resp = await ingest_callback_logs( + body, user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN) + ) + assert resp.processed == 1 and resp.failed == 1 + # The failed record is reported back by index + error, not silently dropped. + assert len(resp.failures) == 1 + assert resp.failures[0].index == 0 + assert "boom on first record" in resp.failures[0].error + + +def test_batch_over_limit_is_rejected(): + from litellm.constants import MAX_CALLBACK_LOG_RECORDS + from pydantic import ValidationError + + # One over the cap must fail validation (422 at the API boundary), bounding + # the callback/DB fan-out a single POST can trigger. + too_many = [ + CallbackLogRecord(status="success", standard_logging_payload=_sample_payload()) + for _ in range(MAX_CALLBACK_LOG_RECORDS + 1) + ] + with pytest.raises(ValidationError): + CallbackLogsRequest(records=too_many) diff --git a/ui/litellm-dashboard/src/lib/http/schema.d.ts b/ui/litellm-dashboard/src/lib/http/schema.d.ts index 6fae14ee6ec5..3669acae67ed 100644 --- a/ui/litellm-dashboard/src/lib/http/schema.d.ts +++ b/ui/litellm-dashboard/src/lib/http/schema.d.ts @@ -17236,6 +17236,29 @@ export interface paths { patch?: never; trace?: never; }; + "/v1/rust_control_plane/logs": { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + get?: never; + put?: never; + /** + * Ingest Callback Logs + * @description Replay a batch of finished logging payloads through the callback fan-out. + * + * Admin-only: the payloads write spend logs and trigger every callback, so this + * is a trusted internal route, not a public surface. + */ + post: operations["ingest_callback_logs_v1_rust_control_plane_logs_post"]; + delete?: never; + options?: never; + head?: never; + patch?: never; + trace?: never; + }; "/v1/search": { parameters: { query?: never; @@ -21512,6 +21535,54 @@ export interface components { /** Callback Name */ callback_name: string; }; + /** + * CallbackLogFailure + * @description A record that failed to replay, identified by its index in the batch. + */ + CallbackLogFailure: { + /** Error */ + error: string; + /** Index */ + index: number; + }; + /** + * CallbackLogRecord + * @description A single finished logging event to replay through the callbacks. + */ + CallbackLogRecord: { + /** Error */ + error?: string | null; + /** Standard Logging Payload */ + standard_logging_payload: { + [key: string]: unknown; + }; + /** + * Status + * @enum {string} + */ + status: "success" | "failure"; + }; + /** + * CallbackLogsRequest + * @description A batch of logging events posted by an external producer. + */ + CallbackLogsRequest: { + /** Records */ + records: components["schemas"]["CallbackLogRecord"][]; + }; + /** + * CallbackLogsResponse + * @description Per-batch result: counts plus per-record failure detail so the caller can + * distinguish a transient callback error from a structurally bad payload. + */ + CallbackLogsResponse: { + /** Failed */ + failed: number; + /** Failures */ + failures?: components["schemas"]["CallbackLogFailure"][]; + /** Processed */ + processed: number; + }; /** CallbacksByType */ CallbacksByType: { /** Failure */ @@ -54686,6 +54757,39 @@ export interface operations { }; }; }; + ingest_callback_logs_v1_rust_control_plane_logs_post: { + parameters: { + query?: never; + header?: never; + path?: never; + cookie?: never; + }; + requestBody: { + content: { + "application/json": components["schemas"]["CallbackLogsRequest"]; + }; + }; + responses: { + /** @description Successful Response */ + 200: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["CallbackLogsResponse"]; + }; + }; + /** @description Validation Error */ + 422: { + headers: { + [name: string]: unknown; + }; + content: { + "application/json": components["schemas"]["HTTPValidationError"]; + }; + }; + }; + }; search_v1_search_post: { parameters: { query?: {