diff --git a/litellm-rust/Cargo.lock b/litellm-rust/Cargo.lock
index a52f379f98b..133769dc7ec 100644
--- a/litellm-rust/Cargo.lock
+++ b/litellm-rust/Cargo.lock
@@ -1409,14 +1409,10 @@ name = "litellm-ai-gateway"
version = "0.1.0"
dependencies = [
"axum",
- "base64",
"futures-channel",
"futures-util",
"litellm-config",
"litellm-core",
- "reqwest",
- "rustls 0.23.42",
- "rustls-native-certs",
"serde",
"serde_json",
"sha2 0.10.9",
@@ -2306,6 +2302,7 @@ version = "1.0.150"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "e8014e44b4736ed0538adeecded0fce2a272f22dc9578a7eb6b2d9993c74cfb9"
dependencies = [
+ "indexmap",
"itoa",
"memchr",
"serde",
diff --git a/litellm-rust/Cargo.toml b/litellm-rust/Cargo.toml
index b33fed58fca..0a69cb86276 100644
--- a/litellm-rust/Cargo.toml
+++ b/litellm-rust/Cargo.toml
@@ -32,7 +32,7 @@ rustls = { version = "0.23", default-features = false, features = ["ring", "std"
rustls-native-certs = "0.8"
serial_test = { version = "4.0.1", default-features = false }
serde = { version = "1.0", features = ["derive"] }
-serde_json = { version = "1.0", features = ["float_roundtrip"] }
+serde_json = { version = "1.0", features = ["float_roundtrip", "preserve_order"] }
sha2 = "0.10"
subtle = "2"
thiserror = "2.0"
diff --git a/litellm-rust/crates/ai-gateway/ARCHITECTURE.md b/litellm-rust/crates/ai-gateway/ARCHITECTURE.md
index 6d090cf4c8e..82301795687 100644
--- a/litellm-rust/crates/ai-gateway/ARCHITECTURE.md
+++ b/litellm-rust/crates/ai-gateway/ARCHITECTURE.md
@@ -1,13 +1,23 @@
# 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.
+The Rust ai-gateway is the Axum transport host for core routes. It authenticates
+clients, selects deployments, adapts HTTP or WebSocket traffic, and supplies
+terminal logging services. Provider transformation, authentication, HTTP and
+WebSocket I/O, stream instrumentation, and session completion live in
+`litellm-core`.
+
+Spend tracking is an API callback: the gateway's terminal logger POSTs each
+finished call or session to the LiteLLM proxy, which records spend and runs the
+usual callbacks. Streaming and WebSocket routes retain their core completion
+owner until the stream or session ends, so committed calls produce one terminal
+record. Realtime pool warmup is only connection preparation and produces zero
+terminal records on success or failure.
```mermaid
flowchart LR
- C[client] <--> G[Rust ai-gateway
LLM inference]
- G <--> O[OpenAI realtime]
+ C[client] <--> G[Rust ai-gateway
Axum transport]
+ G <--> K[litellm-core
route and provider I/O]
+ K <--> O[provider]
G -. spend tracking callback .-> P[litellm proxy]
F[litellm-config
load-time only] --> G
F -. Python backend .-> P
diff --git a/litellm-rust/crates/ai-gateway/Cargo.toml b/litellm-rust/crates/ai-gateway/Cargo.toml
index f1834922c84..2d317e65159 100644
--- a/litellm-rust/crates/ai-gateway/Cargo.toml
+++ b/litellm-rust/crates/ai-gateway/Cargo.toml
@@ -17,14 +17,10 @@ required-features = ["server"]
tracing.workspace = true
litellm-core = { workspace = true, features = ["bedrock-auth"] }
litellm-config.workspace = true
-# reqwest (rustls + json) is used by io/ocr and ships realtime logs to the
-# Python proxy callbacks API.
-reqwest.workspace = true
# `sync` powers the bounded mpsc channel the realtime logger drains.
tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "time", "sync"] }
futures-util.workspace = true
serde_json.workspace = true
-base64.workspace = true
axum = { workspace = true, features = ["ws"], optional = true }
serde.workspace = true
subtle = { workspace = true, optional = true }
@@ -43,4 +39,5 @@ trace-parity = ["server", "dep:tower", "litellm-core/observability"]
[dev-dependencies]
futures-channel = "0.3"
+tokio-tungstenite.workspace = true
tower = { version = "0.5.3", features = ["util"] }
diff --git a/litellm-rust/crates/ai-gateway/README.md b/litellm-rust/crates/ai-gateway/README.md
index 28d94bf16e8..48659bc571e 100644
--- a/litellm-rust/crates/ai-gateway/README.md
+++ b/litellm-rust/crates/ai-gateway/README.md
@@ -1,8 +1,8 @@
# LiteLLM Rust AI Gateway
-A minimal Axum service that fronts OpenAI's realtime API. Clients open a
-WebSocket to `GET /v1/realtime`; the gateway authenticates, selects a deployment,
-dials OpenAI upstream, and splices the two sockets frame-by-frame.
+A minimal Axum service that hosts LiteLLM core routes. For realtime, clients open
+a WebSocket to `GET /v1/realtime`; the gateway authenticates, selects a
+deployment, and adapts frames while core dials OpenAI and splices the session.
## Crates
@@ -16,7 +16,9 @@ dials OpenAI upstream, and splices the two sockets frame-by-frame.
| litellm-python-interop | Domain-neutral PyO3 foundation: typed Python/Serde conversion, retained callbacks, and sync/async Python↔Tokio execution. |
| litellm-python-bridge | PyO3 cdylib exposing LiteLLM Rust APIs to the Python SDK. |
-Dependency direction is acyclic: config depends on core, the gateway depends on config and core, and the Python bridge depends on the domain layers and Python interop.
+Dependency direction is acyclic: config depends on core, the gateway depends on
+config and core, and the Python bridge depends on core and Python interop. Its
+optional `trace-parity` diagnostics also depend on the gateway.
- **Client endpoint:** `wss:///v1/realtime?model=` (WebSocket)
- **Auth:** `Authorization: Bearer $LITELLM_MASTER_KEY` (fails closed if unset)
@@ -94,12 +96,15 @@ stand-in only for the leanest possible build.
## Request logging
-The gateway runs no spend logic. When a session ends it builds one
+The gateway runs no spend logic. When core completes a call or session, the
+gateway terminal logger 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.
+down. It sends one payload per completed call or session. Realtime connection
+warmup sends none; only a connection handed to a serving session can complete
+and emit a terminal payload. 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).
diff --git a/litellm-rust/crates/ai-gateway/src/constants.rs b/litellm-rust/crates/ai-gateway/src/constants.rs
index 1ed44cf7048..07f48f5fa80 100644
--- a/litellm-rust/crates/ai-gateway/src/constants.rs
+++ b/litellm-rust/crates/ai-gateway/src/constants.rs
@@ -5,10 +5,6 @@
//! modules. Env-overridable tunables keep their `DEFAULT_*` value here; the env
//! read + fallback happens at the host/config layer.
-/// Provider attributed to realtime sessions in the logging payload.
-#[cfg(feature = "server")]
-pub(crate) const DEFAULT_PROVIDER: &str = "openai";
-
/// HTTP path for the non-streaming Anthropic Messages route.
#[cfg(feature = "server")]
pub(crate) const MESSAGES_ROUTE_PATH: &str = "/v1/messages";
diff --git a/litellm-rust/crates/ai-gateway/src/integrations/README.md b/litellm-rust/crates/ai-gateway/src/integrations/README.md
deleted file mode 100644
index 16a162dac57..00000000000
--- a/litellm-rust/crates/ai-gateway/src/integrations/README.md
+++ /dev/null
@@ -1,127 +0,0 @@
-# LiteLLM Rust integrations
-
-This directory contains Rust-native equivalents of LiteLLM integration hooks.
-The first supported surfaces are terminal custom loggers and pre/during-call
-custom guardrails.
-
-## File layout
-
-Every integration is a folder:
-
-- `mod.rs` contains the implementation, trait, runner, or adapter
-- `types.rs` contains the integration-local request, response, error, and future
- types
-
-Do not add new flat integration files such as `custom_logger.rs`. Shared wire
-contracts that are used by multiple integrations can stay in
-`integrations/types.rs`.
-
-Call ordering and lifecycle timing live in `litellm-core/src/call_lifecycle`.
-Call-type modules, such as OCR, adapt their request and response shapes into
-that generic lifecycle runner.
-
-## CustomLogger
-
-Implement `CustomLogger` when Rust code needs to observe terminal success or
-failure events. Method names intentionally match Python `CustomLogger` names.
-
-```rust
-use litellm_ai_gateway::integrations::custom_logger::{
- CallbackTiming, CallbackValue, CustomLogger, LogFuture, ModelCallDetails,
-};
-
-struct RecordingLogger;
-
-impl CustomLogger for RecordingLogger {
- fn async_log_success_event<'a>(
- &'a self,
- model_call_details: &'a ModelCallDetails,
- response_obj: &'a CallbackValue,
- timing: CallbackTiming,
- ) -> LogFuture<'a> {
- Box::pin(async move {
- let model = &model_call_details.model;
- let provider = &model_call_details.custom_llm_provider;
- let call_type = model_call_details.call_type.to_string();
- let request_id = model_call_details.request_id.as_deref();
- let response_object = &response_obj.object;
- let duration = timing.end_time - timing.start_time;
- let standard_payload = model_call_details.standard_logging_payload.as_ref();
-
- Ok(())
- })
- }
-
- fn async_log_failure_event<'a>(
- &'a self,
- model_call_details: &'a ModelCallDetails,
- response_obj: Option<&'a CallbackValue>,
- timing: CallbackTiming,
- ) -> LogFuture<'a> {
- Box::pin(async move {
- let error = model_call_details.failure_error.as_ref();
- let response_object = response_obj.map(|value| value.object.as_str());
- let duration = timing.end_time - timing.start_time;
-
- Ok(())
- })
- }
-}
-```
-
-Use `CustomLoggerRunner` to fan out terminal events to configured loggers. The
-runner is a no-op when no loggers are configured, which is the expected fast
-path for requests without callbacks.
-
-## CustomGuardrail
-
-Implement `CustomGuardrail` when Rust code needs to run pre-call or native
-during-call checks. Method names intentionally match Python `CustomGuardrail`
-entrypoints inherited from Python `CustomLogger`.
-
-```rust
-use litellm_ai_gateway::integrations::custom_guardrail::{
- CustomGuardrail, GuardrailContext, GuardrailDecision, GuardrailEventHook,
- GuardrailFuture, GuardrailRequest,
-};
-
-struct BlocklistedPromptGuardrail;
-
-impl CustomGuardrail for BlocklistedPromptGuardrail {
- fn guardrail_name(&self) -> &str {
- "blocklisted-prompt"
- }
-
- fn supported_event_hooks(&self) -> &[GuardrailEventHook] {
- &[GuardrailEventHook::PreCall]
- }
-
- fn async_pre_call_hook<'a>(
- &'a self,
- _context: &'a GuardrailContext,
- request: GuardrailRequest,
- ) -> GuardrailFuture<'a> {
- Box::pin(async move {
- if request.data.to_string().contains("blocked phrase") {
- return Ok(GuardrailDecision::Block(
- litellm_ai_gateway::integrations::custom_guardrail::GuardrailError::blocked(
- "blocked phrase detected",
- ),
- ));
- }
- Ok(GuardrailDecision::Allow(request))
- })
- }
-}
-```
-
-Use `CustomGuardrailRunner::run_pre_call` for `pre_call` guardrails and
-`CustomGuardrailRunner::run_during_call` for `during_call` guardrails. A
-`GuardrailDecision::Mask` continues with modified request data.
-`GuardrailDecision::Block` short-circuits the provider call.
-
-## Current boundary
-
-These are Rust-only primitives. Python callback and guardrail adapters are a
-separate layer that should implement these Rust traits instead of changing the
-runner interfaces.
diff --git a/litellm-rust/crates/ai-gateway/src/io/mod.rs b/litellm-rust/crates/ai-gateway/src/io/mod.rs
index 90c7a65ddf8..b26180c495e 100644
--- a/litellm-rust/crates/ai-gateway/src/io/mod.rs
+++ b/litellm-rust/crates/ai-gateway/src/io/mod.rs
@@ -1,3 +1,2 @@
pub mod audio_transcription;
pub mod realtime_pool;
-pub(crate) mod tls;
diff --git a/litellm-rust/crates/ai-gateway/src/io/realtime_pool.rs b/litellm-rust/crates/ai-gateway/src/io/realtime_pool.rs
index e910c92bffc..29c0efac041 100644
--- a/litellm-rust/crates/ai-gateway/src/io/realtime_pool.rs
+++ b/litellm-rust/crates/ai-gateway/src/io/realtime_pool.rs
@@ -425,7 +425,7 @@ pub fn upstream_key(
#[cfg(test)]
mod tests {
use super::*;
- use futures_util::SinkExt;
+ use futures_util::{SinkExt, StreamExt};
use std::net::SocketAddr;
use tokio::net::TcpListener;
use tokio_tungstenite::tungstenite::Message;
@@ -478,15 +478,11 @@ mod tests {
}
fn key_for(base: &str) -> UpstreamKey {
- UpstreamKey {
- model: "gpt-realtime".to_string(),
- api_key: "sk-test".to_string(),
- api_base: Some(base.to_string()),
- }
+ UpstreamKey::new("gpt-realtime", Some("sk-test"), Some(base)).unwrap()
}
#[tokio::test]
- async fn warm_handoff_relays_buffered_session_created() {
+ async fn taking_a_warm_connection_consumes_one_pool_entry() {
let base = spawn_fake_openai().await;
let pool = RealtimePool::new_unspawned(test_config());
let key = key_for(&base);
@@ -494,17 +490,7 @@ mod tests {
pool.warm_now(&key).await;
assert_eq!(pool.warm_len(&key), 2);
- let handoff = pool.take(&key).expect("a warm socket should be available");
- assert_eq!(handoff.session_created.event_type, "session.created");
- assert_eq!(
- handoff
- .session_created
- .data
- .get("session")
- .and_then(|s| s.get("id"))
- .and_then(|v| v.as_str()),
- Some("sess_fake")
- );
+ let _handoff = pool.take(&key).expect("a warm socket should be available");
// Taking one leaves one.
assert_eq!(pool.warm_len(&key), 1);
}
@@ -577,29 +563,7 @@ mod tests {
test_config().target_size,
"background replenisher should warm up to target_size"
);
- let handoff = pool.take(&key).expect("a warm socket should be available");
- assert_eq!(handoff.session_created.event_type, "session.created");
- }
-
- #[tokio::test]
- async fn closed_upstream_socket_is_detected_dead() {
- // A genuinely dead socket: dial the fake, read session.created, then drop
- // the server by closing from our side and waiting for the close to land.
- let base = spawn_fake_openai().await;
- let pool = RealtimePool::new_unspawned(test_config());
- let key = key_for(&base);
- pool.register(key.clone());
-
- let mut conn = warm_one(&key).await.expect("warm one");
- // Close the upstream from the client side; the server echoes a close.
- let _ = conn.tx.send(Message::Close(None)).await;
- // Give the close a moment to arrive on rx.
- tokio::time::sleep(Duration::from_millis(50)).await;
- pool.insert_warm(key.clone(), conn);
-
- // Liveness check at take() should detect the close and discard it.
- assert!(pool.take(&key).is_none());
- assert_eq!(pool.warm_len(&key), 0);
+ let _handoff = pool.take(&key).expect("a warm socket should be available");
}
#[tokio::test]
diff --git a/litellm-rust/crates/ai-gateway/src/io/tls.rs b/litellm-rust/crates/ai-gateway/src/io/tls.rs
deleted file mode 100644
index a2562f60345..00000000000
--- a/litellm-rust/crates/ai-gateway/src/io/tls.rs
+++ /dev/null
@@ -1,80 +0,0 @@
-//! Outbound WebSocket dials over a TLS config this crate builds once and owns.
-//!
-//! `reqwest/rustls-tls` enables `rustls/ring` and `litellm-core`'s `bedrock-auth`
-//! enables `rustls/aws-lc-rs`, so the bare `ClientConfig::builder()` that
-//! `tokio-tungstenite` uses when handed no connector panics rather than guess
-//! between them. Naming ring on a connector of our own settles that for these
-//! dials without touching the process-wide default, and building the config
-//! once keeps the platform trust store, which `tokio-tungstenite` would
-//! otherwise re-read on every dial, off the dial path.
-
-use std::io;
-use std::sync::{Arc, OnceLock};
-
-use rustls::{ClientConfig, RootCertStore};
-use tokio::net::TcpStream;
-use tokio_tungstenite::tungstenite::Error;
-use tokio_tungstenite::tungstenite::client::IntoClientRequest;
-use tokio_tungstenite::tungstenite::error::TlsError;
-use tokio_tungstenite::tungstenite::handshake::client::Response;
-use tokio_tungstenite::{
- Connector, MaybeTlsStream, WebSocketStream, connect_async_tls_with_config,
-};
-
-static TLS_CONFIG: OnceLock> = OnceLock::new();
-
-fn build_config() -> Result> {
- let native = rustls_native_certs::load_native_certs();
- let roots = {
- let mut store = RootCertStore::empty();
- let (added, _ignored) = store.add_parsable_certificates(native.certs);
- if added == 0 {
- return Err(Box::new(Error::Io(io::Error::other(format!(
- "no usable native root certificates: {:?}",
- native.errors
- )))));
- }
- store
- };
-
- ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider()))
- .with_safe_default_protocol_versions()
- .map(|builder| builder.with_root_certificates(roots).with_no_client_auth())
- .map_err(|error| Box::new(Error::Tls(TlsError::Rustls(error))))
-}
-
-fn tls_config() -> Result, Box> {
- if let Some(config) = TLS_CONFIG.get() {
- return Ok(Arc::clone(config));
- }
- let built = Arc::new(build_config()?);
- Ok(Arc::clone(TLS_CONFIG.get_or_init(|| built)))
-}
-
-pub(crate) async fn connect_upstream(
- request: R,
-) -> Result<(WebSocketStream>, Response), Box>
-where
- R: IntoClientRequest + Unpin,
-{
- let request = request.into_client_request().map_err(Box::new)?;
- let connector = match request.uri().scheme_str() {
- Some("wss") => Some(Connector::Rustls(tls_config()?)),
- _ => None,
- };
- connect_async_tls_with_config(request, None, false, connector)
- .await
- .map_err(Box::new)
-}
-
-#[cfg(test)]
-mod tests {
- use super::build_config;
-
- #[test]
- fn builds_a_usable_config_with_both_provider_features_enabled() {
- let config = build_config().expect("a client config");
-
- assert!(!config.crypto_provider().cipher_suites.is_empty());
- }
-}
diff --git a/litellm-rust/crates/ai-gateway/src/main.rs b/litellm-rust/crates/ai-gateway/src/main.rs
index 4cfd784dda4..a1b1aa1c278 100644
--- a/litellm-rust/crates/ai-gateway/src/main.rs
+++ b/litellm-rust/crates/ai-gateway/src/main.rs
@@ -1,8 +1,8 @@
//! LiteLLM AI Gateway — a minimal Axum server fronting the Rust router.
//!
//! Flow: client → `POST /v1/realtime` → `router.realtime()` selects a deployment
-//! (simple-shuffle) → `io::realtime::realtime()` invokes OpenAI. The
-//! server owns transport + config; routing lives in the `router` crate.
+//! (simple-shuffle) → `core::realtime()` invokes OpenAI. The server owns config
+//! and Axum adaptation; routing and provider execution live in core.
//!
//! The binary requires the `server` feature (declared in `Cargo.toml` via
//! `required-features`), so cargo skips it unless that feature is on. Everything
@@ -135,12 +135,8 @@ where
fn register_deployments(router: &Router, pool: &RealtimePool) {
for deployment in router.deployments() {
let params = &deployment.litellm_params;
- let provider_model = params
- .model
- .strip_prefix("openai/")
- .unwrap_or(¶ms.model);
if let Some(key) = upstream_key(
- provider_model,
+ ¶ms.model,
params.api_key.as_deref(),
params.api_base.as_deref(),
) {
diff --git a/litellm-rust/crates/ai-gateway/src/routes/realtime/README.md b/litellm-rust/crates/ai-gateway/src/routes/realtime/README.md
index 3301576bb85..4e7e43ab9d4 100644
--- a/litellm-rust/crates/ai-gateway/src/routes/realtime/README.md
+++ b/litellm-rust/crates/ai-gateway/src/routes/realtime/README.md
@@ -1,9 +1,11 @@
# Realtime route (`GET /v1/realtime`)
-Proxies OpenAI's realtime WebSocket. `mod.rs` is the axum surface (handler +
-socket↔events adapter); `service.rs` is the pure logic (select a deployment, then
-splice client ↔ upstream). The pool itself lives in
-`crates/providers/src/realtime_pool.rs`.
+Proxies OpenAI's realtime WebSocket. `mod.rs` is the Axum surface and converts
+socket frames to core events. `service.rs` selects a deployment and calls
+`litellm_core::realtime::realtime`. Core owns provider resolution, upstream
+dialing, event splicing, usage observation, and session completion. The pool
+lives in `ai-gateway/src/io/realtime_pool.rs` and stores core `WarmConnection`
+values as a transport optimization.
## Connection pooling
@@ -46,6 +48,11 @@ unprompted on connect, we pre-read exactly that one frame and relay it on handof
and we send nothing else on the socket before a client exists — so the client's first
`session.update` behaves identically either way.
+Warmup has no user-visible call lifecycle. Whether it succeeds or fails, it emits
+zero terminal records. Ownership transfers only when a warm connection is handed
+to the core realtime route; that serving session emits exactly one terminal record
+when splicing completes or fails.
+
### Sizing
Each warm socket serves **exactly one** session (realtime isn't multiplexed), so the
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 bafe342dbe5..58370605954 100644
--- a/litellm-rust/crates/ai-gateway/src/routes/realtime/service.rs
+++ b/litellm-rust/crates/ai-gateway/src/routes/realtime/service.rs
@@ -7,19 +7,40 @@
//! socket we fresh-dial exactly as before — the pool is never on the critical path
//! for correctness, only latency.
-use std::time::Duration;
use std::sync::Arc;
+use std::time::Duration;
use crate::io::realtime_pool::{RealtimePool, upstream_key};
use futures_util::{Sink, Stream};
use litellm_core::error::Error;
-use litellm_core::integrations::custom_logger::{CustomLogger, CustomLoggerRunner};
+use litellm_core::integrations::custom_logger::{CustomLogger, CustomLoggerRunner, LogFuture};
use litellm_core::integrations::types::{RequestMetadata, StandardLoggingMetadata};
-use litellm_core::lifecycle::{CallLifecycleContext, ExecutedCall};
-use litellm_core::realtime::{RealtimeRequest, realtime};
+use litellm_core::lifecycle::{
+ CallLifecycleContext, Clock, ExecutedCall, TerminalDispatcher, TerminalRecord,
+};
use litellm_core::realtime::types::RealtimeEvent;
+use litellm_core::realtime::{RealtimeRequest, realtime};
use litellm_core::router::Router;
+struct GatewayRealtimeServices {
+ runner: CustomLoggerRunner,
+}
+
+impl Clock for GatewayRealtimeServices {
+ fn now(&self) -> f64 {
+ std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .map(|duration| duration.as_secs_f64())
+ .unwrap_or(0.0)
+ }
+}
+
+impl TerminalDispatcher for GatewayRealtimeServices {
+ fn dispatch<'a>(&'a self, terminal: &'a TerminalRecord) -> LogFuture<'a> {
+ self.runner.dispatch(terminal)
+ }
+}
+
/// Select a deployment for `model` and splice the client stream to the provider.
///
/// `pool` supplies a pre-warmed upstream when one is available; otherwise we
@@ -45,29 +66,28 @@ where
.get_available_deployment(model)
.ok_or_else(|| Error::Routing(format!("no deployment available for model '{model}'")))?;
let params = &deployment.litellm_params;
- // Strip a leading `openai/` so the OpenAI-only realtime fn gets the bare model.
- let provider_model = params
- .model
- .strip_prefix("openai/")
- .unwrap_or(¶ms.model);
-
let connection = upstream_key(
- provider_model,
+ ¶ms.model,
params.api_key.as_deref(),
params.api_base.as_deref(),
- ).ok_or_else(|| Error::Auth("missing realtime provider API key".to_string()))?;
- let warm = pool.take(&connection);
- let context = CallLifecycleContext::new("realtime", model, "openai", call_id)
- .with_metadata(StandardLoggingMetadata {
+ );
+ let warm = connection.as_ref().and_then(|key| pool.take(key));
+ let context = CallLifecycleContext::new("realtime", model, "openai", call_id).with_metadata(
+ StandardLoggingMetadata {
user_api_key_hash: metadata.user_api_key_hash,
user_api_key_user_id: metadata.user_api_key_user_id,
user_api_key_team_id: metadata.user_api_key_team_id,
..Default::default()
- });
+ },
+ );
Ok(realtime(
- &CustomLoggerRunner::new(loggers.as_ref().clone()),
+ &GatewayRealtimeServices {
+ runner: CustomLoggerRunner::new(loggers.as_ref().clone()),
+ },
RealtimeRequest {
- connection,
+ model: params.model.clone(),
+ api_key: params.api_key.clone(),
+ api_base: params.api_base.clone(),
warm,
idle_timeout,
},
diff --git a/litellm-rust/crates/ai-gateway/src/routes/responses/mod.rs b/litellm-rust/crates/ai-gateway/src/routes/responses/mod.rs
index 1912a51ff17..b11fb7b411d 100644
--- a/litellm-rust/crates/ai-gateway/src/routes/responses/mod.rs
+++ b/litellm-rust/crates/ai-gateway/src/routes/responses/mod.rs
@@ -212,7 +212,11 @@ async fn bridge(
};
let client_in = Box::pin(stream.filter_map(|message| async move {
match message {
- Ok(Message::Text(text)) => serde_json::from_str::(&text).ok(),
+ Ok(Message::Text(text)) => Some(
+ serde_json::from_str::(&text)
+ .map_err(|error| litellm_core::Error::InvalidRequest(error.to_string())),
+ ),
+ Err(error) => Some(Err(litellm_core::Error::Network(error.to_string()))),
_ => None,
}
}));
diff --git a/litellm-rust/crates/ai-gateway/src/routes/responses/service.rs b/litellm-rust/crates/ai-gateway/src/routes/responses/service.rs
index 9bf56d5cc02..36e64bee259 100644
--- a/litellm-rust/crates/ai-gateway/src/routes/responses/service.rs
+++ b/litellm-rust/crates/ai-gateway/src/routes/responses/service.rs
@@ -51,7 +51,7 @@ pub async fn run(
client_out: Out,
) -> Result, Error>
where
- In: Stream- + Unpin + Send,
+ In: Stream
- > + Unpin + Send,
Out: Sink + Unpin + Send,
Out::Error: std::fmt::Display,
{
@@ -76,7 +76,7 @@ where
..Default::default()
});
responses_websocket(
- &GatewayResponsesServices::new(loggers),
+ Arc::new(GatewayResponsesServices::new(loggers)),
ResponsesWebSocketRequest {
model: provider_model.to_string(),
api_key: params.api_key.clone(),
diff --git a/litellm-rust/crates/core/AGENTS.md b/litellm-rust/crates/core/AGENTS.md
index aa0a94521d4..75daa5a6912 100644
--- a/litellm-rust/crates/core/AGENTS.md
+++ b/litellm-rust/crates/core/AGENTS.md
@@ -46,8 +46,9 @@ core/src/messages/
authoritative roots into a native request after callbacks, and sends it through
`http_utils::buffered_post`. Reducto upload, Azure Document Intelligence polling,
and HTTP document URL conversion are declined at admission until they have an
-implementation on this settled-request path. `audio_transcription` and
-`realtime` remain in flight.
+implementation on this settled-request path. Audio transcription provider I/O,
+realtime WebSocket dialing and splicing, and Responses WebSocket dialing and
+splicing are also core-owned.
The invariant is one function body owns the route lifecycle. The conceptual
shape is:
@@ -75,7 +76,7 @@ requirements traits (an `OcrServices` bundle) declare exactly what a route needs
passed a catch-all gateway environment. The ordinary Rust client supplies native
defaults; callers override implementations at construction.
-`CallServices` opens one request-scoped sessions per call, owning per-call state
+`CallServices` opens one request-scoped session per call, owning per-call state
(timing, logging state, retained host objects, deferred completion, correlation
state). No-op call services are the default, and their presence must not move
provider behavior into a host or force callback payload materialization on an
@@ -108,21 +109,20 @@ service construction dependencies do not leak into service interfaces
## Not the target
-The current `CallLifecycleHooks` shape is not the final public service API: it
-folds lifecycle sequencing into a stateless generic transformation interface,
-needs `Send` futures, cannot express the full replacement and error contracts,
-and does not model request-scoped retained ownership or Python caller-task
-driving. It is a stepping stone, not the contract to build new routes against.
-
Do not introduce a dynamic `TypeId` service map, a shared gateway callback
environment reused from core or the bridge, per-callback JSON serialization, or
a full Effect layer API. Services traits plus constructors and a scoped call
owner are the minimum design; add more machinery only when concrete consumers
require it.
-## Gateway migration
+## Streaming ownership
-Existing gateway-hosted OCR, transcription and WebSocket provider execution
-predates this boundary and must move here as those routes migrate. Keeping a
-gateway callback as a `CallServices` implementation does not justify keeping
-provider orchestration beside it in the gateway.
+Core owns provider sessions through completion. Streaming HTTP calls transfer a
+`StreamingCall` whose completion registration keeps terminal dispatch alive
+until the host finishes or drops the stream. Realtime and Responses WebSocket
+entrypoints retain the provider connection while they splice events, then emit
+exactly one terminal record after the committed session completes or fails.
+
+Realtime pool warmup is not a user call and must emit zero terminal records on
+both success and failure. A warmed connection transfers into `realtime`; only
+that serving session owns completion and terminal dispatch.
diff --git a/litellm-rust/crates/core/src/audio_transcription/handler.rs b/litellm-rust/crates/core/src/audio_transcription/handler.rs
index 9a96b9d1140..bf36a6b9fb7 100644
--- a/litellm-rust/crates/core/src/audio_transcription/handler.rs
+++ b/litellm-rust/crates/core/src/audio_transcription/handler.rs
@@ -7,7 +7,7 @@ use super::client::http_client;
use super::types::ProviderAudioTranscriptionRequest;
#[tracing::instrument(target = "litellm::function_trace", level = "trace", skip_all)]
-pub async fn execute_audio_transcription_provider_call(
+pub(super) async fn execute_audio_transcription_provider_call(
request: ProviderAudioTranscriptionRequest,
) -> Result {
let body = serde_json::to_vec(&request.body)
diff --git a/litellm-rust/crates/core/src/audio_transcription/mod.rs b/litellm-rust/crates/core/src/audio_transcription/mod.rs
index 811acd4a8e5..40af2e3c86d 100644
--- a/litellm-rust/crates/core/src/audio_transcription/mod.rs
+++ b/litellm-rust/crates/core/src/audio_transcription/mod.rs
@@ -8,9 +8,7 @@ pub mod types;
use serde_json::Value;
-pub use handler::execute_audio_transcription_provider_call;
pub use lifecycle::{AudioRoute, AudioServices, DefaultAudioServices};
-pub use prepare::prepare_audio_transcription_provider_call;
pub use types::{AudioRouteRequest, AudioTranscriptionRequest, ProviderAudioTranscriptionRequest};
#[tracing::instrument(target = "litellm::function_trace", level = "trace", skip_all)]
diff --git a/litellm-rust/crates/core/src/audio_transcription/prepare.rs b/litellm-rust/crates/core/src/audio_transcription/prepare.rs
index bbef97341a9..25139da1d1c 100644
--- a/litellm-rust/crates/core/src/audio_transcription/prepare.rs
+++ b/litellm-rust/crates/core/src/audio_transcription/prepare.rs
@@ -18,7 +18,7 @@ fn provider_config(provider: &str) -> Option<&'static dyn AudioTranscriptionProv
}
#[tracing::instrument(target = "litellm::function_trace", level = "trace", skip_all)]
-pub fn prepare_audio_transcription_provider_call(
+pub(super) fn prepare_audio_transcription_provider_call(
request: AudioTranscriptionRequest<'_>,
) -> Result {
let provider_info = get_custom_llm_provider(request.model, request.custom_llm_provider)
diff --git a/litellm-rust/crates/core/src/call_lifecycle/README.md b/litellm-rust/crates/core/src/call_lifecycle/README.md
deleted file mode 100644
index 692e249ef27..00000000000
--- a/litellm-rust/crates/core/src/call_lifecycle/README.md
+++ /dev/null
@@ -1,167 +0,0 @@
-# Call lifecycle
-
-`litellm_core::call_lifecycle` is the shared execution wrapper for LiteLLM call
-types migrated to Rust. It owns lifecycle ordering, phase timing, and trace
-observer calls. It must not know about OCR, chat, messages, responses,
-completions, provider auth, request transforms, or response normalization.
-
-Call-type modules own their domain behavior. For example, OCR owns document
-payloads, OCR provider transforms, safe document fetch, guardrail payload shape,
-callback payload shape, and provider HTTP execution.
-
-## Runtime order
-
-Every wrapped call runs in this order:
-
-1. `async_pre_call_hook`
-2. `async_during_call_hook`
-3. provider call
-4. `async_log_success_event` or `async_log_failure_event`
-
-`async_pre_call_hook` receives the initial LiteLLM request shape. It is where
-pre-call custom guardrails run.
-
-`async_during_call_hook` converts the initial request into the provider-ready
-request. It is where provider config selection, parameter mapping, auth/header
-resolution, request transforms, and during-call guardrails belong.
-
-The provider call receives only the provider-ready request. It should execute
-I/O and call the provider response transform.
-
-Success and failure callbacks receive `CallLifecycleTiming`. Callback failures
-must not replace the original provider or guardrail result.
-
-## Trace contract
-
-The lifecycle runner records:
-
-- full call start and end time
-- `pre_call` phase timing
-- `during_call` phase timing
-- `provider_call` phase timing
-- `success_callback` phase timing
-- `failure_callback` phase timing
-
-`CallLifecycleObserver` receives phase start and end events. The default
-observer is a no-op. Future OTEL support should implement this observer instead
-of editing OCR, chat, messages, responses, completions, or provider modules.
-
-## Required shape
-
-Each migrated call type should use this folder shape:
-
-```text
-litellm-rust/crates/ai-gateway/src//
- mod.rs # thin public entrypoint
- types.rs # public request, prepared request, provider request, response types
- prepare.rs # model/provider/callback/guardrail setup
- hooks.rs # CallLifecycleHooks implementation
- handler.rs # provider I/O and response normalization
- tests.rs # call-type lifecycle and handler tests
-```
-
-Provider transforms can live in `litellm-rust/crates/core/src/providers/...`.
-Shared call-type helpers can live beside the call type, but generic lifecycle
-code stays in this folder.
-
-## Core API
-
-The prepared request implements `CallLifecycleRequest`:
-
-```rust
-impl CallLifecycleRequest for PreparedMessagesRequest {
- fn lifecycle_context(&self) -> CallLifecycleContext {
- CallLifecycleContext::new(
- "messages",
- self.model.clone(),
- self.custom_llm_provider.clone(),
- self.litellm_call_id.clone(),
- )
- }
-}
-```
-
-The call-type hooks implement `CallLifecycleHooks`:
-
-```rust
-impl CallLifecycleHooks<
- PreparedMessagesRequest,
- ProviderMessagesRequest,
- MessagesResponse,
-> for MessagesLifecycleHooks {
- fn async_pre_call_hook(...) {
- // run pre-call custom guardrails against the LiteLLM request shape
- }
-
- fn async_during_call_hook(...) {
- // map params, validate env, transform request, run during-call guardrails
- }
-
- fn async_log_success_event(...) {
- // call async_log_success_event on configured custom loggers
- }
-
- fn async_log_failure_event(...) {
- // call async_log_failure_event without swallowing the original error
- }
-}
-```
-
-The public entrypoint stays thin:
-
-```rust
-pub async fn messages(request: MessagesRequest<'_>) -> CoreResult {
- let PreparedMessagesCall { request, hooks } = prepare_messages_call(request)?;
-
- CallLifecycle::default()
- .run_request(request, &hooks, execute_messages_provider_call)
- .await
-}
-```
-
-Use `run_request` for new call types. Keep `run` available only for specialized
-tests or existing code that already has a `CallLifecycleContext`.
-
-## Adding a new call type
-
-1. Add `/types.rs`
-
-Define the public request accepted by the bridge, the prepared request used by
-the lifecycle runner, and the provider request consumed by the handler.
-
-2. Implement `CallLifecycleRequest`
-
-Return `call_type`, `model`, `custom_llm_provider`, and `litellm_call_id`.
-Do not put provider-specific logic here.
-
-3. Add `/prepare.rs`
-
-Resolve model/provider once, generate or preserve `litellm_call_id`, construct
-callback and guardrail runners, and return `PreparedCall`.
-
-4. Add `/hooks.rs`
-
-Implement `CallLifecycleHooks`. Put pre-call guardrail payload construction,
-provider config selection, param mapping, request transform, during-call
-guardrail payload construction, and callback payload construction here.
-
-5. Add `/handler.rs`
-
-Execute the provider request and normalize the provider response. Do not repeat
-provider-specific transforms here; call the provider config.
-
-6. Add tests
-
-Cover hook order, success callback payload, failure callback payload, pre-call
-guardrail blocking before provider I/O, during-call body mutation, and provider
-error mapping.
-
-## Review checklist
-
-- Core lifecycle has no call-type or provider-specific branches
-- Public call-type entrypoint only prepares and calls `run_request`
-- Provider behavior lives behind provider config/transformation code
-- Hook method names map to the Python custom logger and guardrail concepts
-- Phase timing is recorded once in lifecycle, not separately per call type
-- Callback failures never hide the original provider or guardrail error
-- Tests prove the provider socket is not touched when pre-call guardrails block
diff --git a/litellm-rust/crates/core/src/chat_completions/lifecycle.rs b/litellm-rust/crates/core/src/chat_completions/lifecycle.rs
index c2fba6c2b02..9abd7e26293 100644
--- a/litellm-rust/crates/core/src/chat_completions/lifecycle.rs
+++ b/litellm-rust/crates/core/src/chat_completions/lifecycle.rs
@@ -1,29 +1,12 @@
use crate::Error;
-use crate::lifecycle::{
- ActionBinding, ActionKind, Delivery, ErrorDisposition, FailurePolicy, Lifecycle,
- LifecycleRoute, Outcome, Owner, ResultPolicy,
-};
+use crate::lifecycle::program::{CallProgram, ProgramOptions, actions_for};
+use crate::lifecycle::{ActionBinding, Lifecycle, LifecycleRoute, Outcome};
use super::chat_completions_decline_reason;
use serde_json::{Map, Value};
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum Operation {
- Setup,
- DeploymentPre,
- Prepare,
- Send,
- DeploymentSuccess,
- DeploymentFailure,
- SyncSuccess,
- AsyncSuccess,
- SyncSuccessIfNeeded,
- SyncFailure,
- AsyncFailure,
- Restore,
- Complete(Outcome),
-}
+pub use crate::lifecycle::program::{Observations, Operation, Transition};
#[derive(Clone, Debug)]
pub struct Admission {
@@ -39,17 +22,6 @@ pub struct Options {
pub internal_call: bool,
}
-#[derive(Clone, Copy, Debug, Default)]
-pub struct Observations {
- pub logger_available: bool,
- pub has_fallbacks: bool,
-}
-
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub struct Transition {
- pub error: ErrorDisposition,
-}
-
#[derive(Debug, PartialEq, Eq)]
pub struct Decline(&'static str);
@@ -61,10 +33,7 @@ impl Decline {
#[derive(Debug)]
pub struct ChatCompletionsState {
- operation: Operation,
- outcome: Outcome,
- asynchronous: bool,
- internal_call: bool,
+ program: CallProgram,
}
#[derive(Debug)]
@@ -95,15 +64,16 @@ impl LifecycleRoute for ChatCompletionsRoute {
return Ok(Err(Decline(reason)));
}
Ok(Ok(ChatCompletionsState {
- operation: Operation::Setup,
- outcome: Outcome::Success,
- asynchronous: options.asynchronous,
- internal_call: options.internal_call,
+ program: CallProgram::new(ProgramOptions {
+ asynchronous: options.asynchronous,
+ internal_call: options.internal_call,
+ pre_call: false,
+ }),
}))
}
fn operation(state: &Self::State) -> Operation {
- state.operation
+ state.program.operation()
}
fn advance(
@@ -111,95 +81,25 @@ impl LifecycleRoute for ChatCompletionsRoute {
outcome: Outcome,
observations: Observations,
) -> Result {
- use Operation::*;
-
- if matches!(state.operation, Complete(_)) {
- return Err(Error::InvalidRequest(
- "chat completions lifecycle is already complete".into(),
- ));
- }
- let failure =
- if observations.logger_available && !(state.asynchronous && state.internal_call) {
- SyncFailure
- } else {
- Restore
- };
- let error = if outcome != Outcome::Success && state.operation != DeploymentFailure {
- state.outcome = outcome;
- ErrorDisposition::Replace
- } else {
- ErrorDisposition::Preserve
- };
- state.operation = match (state.operation, outcome) {
- (Restore, _) => Complete(state.outcome),
- (DeploymentFailure, _) => failure,
- (_, Outcome::Abort) => Restore,
- (SyncFailure | AsyncFailure, Outcome::Failure) => Restore,
- (Prepare | Send, Outcome::Failure) if state.asynchronous => DeploymentFailure,
- (_, Outcome::Failure) => failure,
- (Setup, Outcome::Success) if state.asynchronous => DeploymentPre,
- (Setup | DeploymentPre, Outcome::Success) => Prepare,
- (Prepare, Outcome::Success) => Send,
- (Send, Outcome::Success) if state.asynchronous => DeploymentSuccess,
- (Send, Outcome::Success) => SyncSuccess,
- (DeploymentSuccess, Outcome::Success) => {
- if state.internal_call || observations.has_fallbacks {
- SyncSuccessIfNeeded
- } else {
- AsyncSuccess
- }
- }
- (AsyncSuccess, Outcome::Success) => SyncSuccessIfNeeded,
- (SyncFailure, Outcome::Success) if state.asynchronous => AsyncFailure,
- (SyncSuccess | SyncSuccessIfNeeded | SyncFailure | AsyncFailure, Outcome::Success) => {
- Restore
- }
- (Complete(_), _) => unreachable!(),
- };
- Ok(Transition { error })
+ state.program.advance(outcome, observations).ok_or_else(|| {
+ Error::InvalidRequest("chat completions lifecycle is already complete".into())
+ })
}
fn actions_for(operation: Operation, _: &Observations) -> &'static [ActionBinding] {
- match operation {
- Operation::Prepare | Operation::Send => &PROVIDER_ACTION,
- Operation::SyncFailure | Operation::AsyncFailure | Operation::DeploymentFailure => {
- &FAILURE_ACTION
- }
- Operation::Restore => &RESTORE_ACTION,
- Operation::Complete(_) => &[],
- _ => &CALLBACK_ACTION,
- }
+ actions_for(operation)
}
}
-const PROVIDER_ACTION: [ActionBinding; 1] = [ActionBinding {
- kind: ActionKind::ProviderCall,
- delivery: Delivery::InlineAwaited,
- on_result: ResultPolicy::Replace,
- on_error: FailurePolicy::Propagate,
- owner: Owner::Core,
-}];
-const CALLBACK_ACTION: [ActionBinding; 1] = [ActionBinding {
- kind: ActionKind::TerminalSuccess,
- delivery: Delivery::InlineAwaited,
- on_result: ResultPolicy::Continue,
- on_error: FailurePolicy::RecordAndContinue,
- owner: Owner::Route,
-}];
-const FAILURE_ACTION: [ActionBinding; 1] = [ActionBinding {
- kind: ActionKind::TerminalFailure,
- delivery: Delivery::InlineAwaited,
- on_result: ResultPolicy::Continue,
- on_error: FailurePolicy::PreserveOriginalFailure,
- owner: Owner::Route,
-}];
-const RESTORE_ACTION: [ActionBinding; 1] = [ActionBinding {
- kind: ActionKind::Restore,
- delivery: Delivery::InlineDirect,
- on_result: ResultPolicy::Continue,
- on_error: FailurePolicy::Propagate,
- owner: Owner::Core,
-}];
+impl Lifecycle {
+ pub fn commitment(&self) -> crate::lifecycle::Commitment {
+ self.state.program.commitment()
+ }
+
+ pub fn failure_stage(&self) -> Option {
+ self.state.program.failure_stage()
+ }
+}
pub fn machine(
admission: &Admission,
diff --git a/litellm-rust/crates/core/src/lifecycle/execution.rs b/litellm-rust/crates/core/src/lifecycle/execution.rs
index fa88b134536..53dbc6f292b 100644
--- a/litellm-rust/crates/core/src/lifecycle/execution.rs
+++ b/litellm-rust/crates/core/src/lifecycle/execution.rs
@@ -188,6 +188,47 @@ impl CallLifecycle {
ClockImpl: Clock,
ProviderCall: FnOnce(ProviderReq) -> ProviderFuture,
ProviderFuture: Future