feat(tracing): add OTLP trace ingestion and reads

.github/workflows/test-unit.yml

@@ -79,7 +79,9 @@ jobs:
 
           - shard: integrations
             artifact-name: integrations
-            test-path: ""
+            test-path: >-
+              tests/test_litellm/integrations
+              tests/test_litellm/tracing
             unit-flag: integrations
             workers: 2
             reruns: 3

litellm-rust/Cargo.lock

@@ -1274,6 +1274,18 @@ version = "0.4.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6e8ccc4ea9f6acc32d102c0f6d471d11d913ad15f20c04de743374861fa1d414"
 
+[[package]]
+name = "const-hex"
+version = "1.19.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0e59eef12462b0f9b0a3620219be5d639afd79fe39dff0a42c3997061f9298b4"
+dependencies = [
+ "cfg-if",
+ "cpufeatures 0.2.17",
+ "proptest",
+ "serde_core",
+]
+
 [[package]]
 name = "const-oid"
 version = "0.9.6"
@@ -2372,9 +2384,9 @@ dependencies = [
  "http-body-util",
  "hyper 1.10.1",
  "lazy_static",
- "opentelemetry",
+ "opentelemetry 0.32.0",
  "opentelemetry-semantic-conventions",
- "opentelemetry_sdk",
+ "opentelemetry_sdk 0.32.1",
  "percent-encoding",
  "pin-project",
  "prost",
@@ -4356,7 +4368,11 @@ dependencies = [
 name = "litellm-traces"
 version = "0.1.0"
 dependencies = [
+ "base64 0.22.1",
+ "flate2",
  "litellm-http",
+ "opentelemetry-proto",
+ "prost",
  "rstest",
  "serde",
  "serde_json",
@@ -4776,6 +4792,33 @@ dependencies = [
  "tracing",
 ]
 
+[[package]]
+name = "opentelemetry"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "6cdb0b1b267eb9db3331b434ed9ddab10d50e280a9adf9d13e5233e2002b61b5"
+dependencies = [
+ "futures-core",
+ "futures-sink",
+ "js-sys",
+ "pin-project-lite",
+ "thiserror 2.0.19",
+]
+
+[[package]]
+name = "opentelemetry-proto"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "25da1ac11a0aeccf38d7f77ee0348715adaf8340f65ad46c94a02c6b20e2f65d"
+dependencies = [
+ "base64 0.22.1",
+ "const-hex",
+ "opentelemetry 0.33.0",
+ "opentelemetry_sdk 0.33.0",
+ "prost",
+ "serde",
+]
+
 [[package]]
 name = "opentelemetry-semantic-conventions"
 version = "0.32.1"
@@ -4791,7 +4834,23 @@ dependencies = [
  "futures-channel",
  "futures-executor",
  "futures-util",
- "opentelemetry",
+ "opentelemetry 0.32.0",
+ "percent-encoding",
+ "portable-atomic",
+ "rand 0.9.5",
+ "thiserror 2.0.19",
+]
+
+[[package]]
+name = "opentelemetry_sdk"
+version = "0.33.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cb39533d9d1c912123efd7d41d7e0c29d16917b60ce15b4c8d87cb1af7f67520"
+dependencies = [
+ "futures-channel",
+ "futures-executor",
+ "futures-util",
+ "opentelemetry 0.33.0",
  "percent-encoding",
  "portable-atomic",
  "rand 0.9.5",
@@ -7524,7 +7583,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26"
 dependencies = [
  "js-sys",
- "opentelemetry",
+ "opentelemetry 0.32.0",
  "tracing",
  "tracing-core",
  "tracing-subscriber",

litellm-rust/crates/python-bridge/src/lib.rs

@@ -44,7 +44,7 @@ mod _native {
     #[pymodule_export]
     use crate::routes::token_counter::TokenCounter;
     #[pymodule_export]
-    use crate::routes::traces::{trace_encode_rows, trace_ensure_schema, trace_query};
+    use crate::routes::traces::{NativeTraceStorage, trace_decode_otlp};
     #[cfg(feature = "huggingface")]
     #[pymodule_export]
     use crate::tokenizer::HuggingFaceEncoding;
@@ -109,9 +109,8 @@ mod tests {
                 "aresponses",
                 "ResponsesWebSocketConnection",
                 "NativeDiagnosticProcessor",
-                "trace_encode_rows",
-                "trace_ensure_schema",
-                "trace_query",
+                "NativeTraceStorage",
+                "trace_decode_otlp",
                 "TokenCounter",
                 "Tokenizer",
                 "gil_stats",

litellm-rust/crates/python-bridge/src/routes/traces.rs

@@ -1,19 +1,21 @@
 use std::collections::BTreeMap;
 
 use litellm_http::ClientVariant;
-use litellm_traces::{Connection, Error, Parameter};
+use litellm_traces::{Connection, Error, InsertTable, Parameter};
 use pyo3::{
-    exceptions::{PyRuntimeError, PyValueError},
+    exceptions::{PyOverflowError, PyRuntimeError, PyValueError},
     prelude::*,
 };
 
 fn map_error(error: Error) -> PyErr {
     match error {
-        Error::InvalidRow | Error::InvalidSchema | Error::EmptySql => {
+        Error::InvalidRow | Error::InvalidTable | Error::InvalidSchema | Error::EmptySql => {
             PyValueError::new_err(error.to_string())
         }
+        Error::InsertTooLarge => PyOverflowError::new_err(error.to_string()),
         Error::InvalidUrl
         | Error::QueryFailed(_)
+        | Error::InsertFailed(_)
         | Error::SchemaFailed(_)
         | Error::ResponseTooLarge
         | Error::InvalidResponse
@@ -21,63 +23,112 @@ fn map_error(error: Error) -> PyErr {
     }
 }
 
-#[pyfunction]
-pub fn trace_ensure_schema<'py>(
-    py: Python<'py>,
-    url: &str,
+#[pyclass]
+pub struct NativeTraceStorage {
     database: String,
-    user: &str,
-    password: &str,
-    trace_retention_days: u32,
-    spend_log_retention_days: u32,
-) -> PyResult<Bound<'py, PyAny>> {
-    let connection = Connection::writer(url, user, password).map_err(map_error)?;
-    let client = crate::http::host_client(py, ClientVariant::NoRedirect)?;
-    crate::execution::run_async(
-        py,
-        async move {
-            litellm_traces::ensure_schema(
-                &client,
-                &connection,
-                &database,
-                trace_retention_days,
-                spend_log_retention_days,
-            )
-            .await
-        },
-        map_error,
-    )
+    writer: Connection,
+    reader: Option<Connection>,
+}
+
+#[pymethods]
+impl NativeTraceStorage {
+    #[new]
+    fn new(database: String, url: &str, reader_url: Option<&str>) -> PyResult<Self> {
+        litellm_traces::schema_statements(&database, 1, 1).map_err(map_error)?;
+        Ok(Self {
+            writer: Connection::writer(url).map_err(map_error)?,
+            reader: reader_url
+                .map(|value| Connection::reader(value, &database))
+                .transpose()
+                .map_err(map_error)?,
+            database,
+        })
+    }
+
+    fn ensure_schema<'py>(
+        &self,
+        py: Python<'py>,
+        trace_retention_days: u32,
+        spend_log_retention_days: u32,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let client = crate::http::host_client(py, ClientVariant::NoRedirect)?;
+        let connection = self.writer.clone();
+        let database = self.database.clone();
+        crate::execution::run_async(
+            py,
+            async move {
+                litellm_traces::ensure_schema(
+                    &client,
+                    &connection,
+                    &database,
+                    trace_retention_days,
+                    spend_log_retention_days,
+                )
+                .await
+            },
+            map_error,
+        )
+    }
+
+    fn insert_rows<'py>(
+        &self,
+        py: Python<'py>,
+        table: &str,
+        #[pyo3(from_py_with = litellm_host_python::from_py_argument)] rows: Vec<
+            BTreeMap<String, serde_json::Value>,
+        >,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let table = InsertTable::parse(table).map_err(map_error)?;
+        let client = crate::http::host_client(py, ClientVariant::NoRedirect)?;
+        let connection = self.writer.clone();
+        let database = self.database.clone();
+        crate::execution::run_async(
+            py,
+            async move {
+                litellm_traces::insert_rows(&client, &connection, &database, table, rows).await
+            },
+            map_error,
+        )
+    }
+
+    fn query<'py>(
+        &self,
+        py: Python<'py>,
+        sql: String,
+        #[pyo3(from_py_with = litellm_host_python::from_py_argument)] parameters: BTreeMap<
+            String,
+            Parameter,
+        >,
+    ) -> PyResult<Bound<'py, PyAny>> {
+        let connection = self.reader.clone().ok_or_else(|| {
+            PyRuntimeError::new_err("Trace reads require a separate ClickHouse reader URL")
+        })?;
+        let client = crate::http::host_client(py, ClientVariant::NoRedirect)?;
+        crate::execution::run_async(
+            py,
+            async move { litellm_traces::execute_read(&client, &connection, &sql, &parameters).await },
+            map_error,
+        )
+    }
 }
 
 #[pyfunction]
-pub fn trace_query<'py>(
+pub fn trace_decode_otlp<'py>(
     py: Python<'py>,
-    url: &str,
-    database: &str,
-    user: &str,
-    password: &str,
-    sql: String,
-    #[pyo3(from_py_with = litellm_host_python::from_py_argument)] parameters: BTreeMap<
-        String,
-        Parameter,
-    >,
+    body: &[u8],
+    content_type: Option<&str>,
+    content_encoding: Option<&str>,
+    max_decompressed_bytes: usize,
 ) -> PyResult<Bound<'py, PyAny>> {
-    let connection = Connection::configured(url, database, user, password).map_err(map_error)?;
-    let client = crate::http::host_client(py, ClientVariant::NoRedirect)?;
-    crate::execution::run_async(
-        py,
-        async move { litellm_traces::execute_read(&client, &connection, &sql, &parameters).await },
-        map_error,
-    )
-}
-
-#[pyfunction]
-pub fn trace_encode_rows(
-    py: Python<'_>,
-    #[pyo3(from_py_with = litellm_host_python::from_py_argument)] rows: Vec<
-        BTreeMap<String, serde_json::Value>,
-    >,
-) -> PyResult<String> {
-    py.detach(|| litellm_traces::encode_rows(rows))
-        .map_err(map_error)
+    let spans = py
+        .detach(|| {
+            litellm_traces::decode_otlp(
+                body,
+                content_type,
+                content_encoding,
+                max_decompressed_bytes,
+            )
+        })
+        .map_err(|error| PyValueError::new_err(error.to_string()))?;
+    litellm_host_python::Pythonized(spans).into_pyobject(py)
 }

litellm-rust/crates/traces/AGENTS.md

@@ -2,5 +2,6 @@
 - Keep this crate independent of Python; PyO3 conversion and public Python exceptions belong in `python-bridge`
 - Keep the SQL migrations here as the only ClickHouse schema definition
 - Use typed query parameters and a dedicated SELECT-only reader with server-side limits
+- Keep `config/reader.xml` grants on the database the schema is created in (CLICKHOUSE_DATABASE, default `litellm`)
 - Bound insert time and encoded bytes; make retry deduplication behavior explicit for supported ClickHouse versions
 - Test storage behavior through the crate's public API against ClickHouse

litellm-rust/crates/traces/Cargo.toml

@@ -6,6 +6,10 @@ license.workspace = true
 repository.workspace = true
 
 [dependencies]
+base64.workspace = true
+flate2.workspace = true
+opentelemetry-proto = { version = "0.33.0", default-features = false, features = ["gen-tonic-messages", "trace", "with-serde"] }
+prost = "0.14.4"
 time = { workspace = true, features = ["formatting"] }
 litellm-http.workspace = true
 serde.workspace = true

litellm-rust/crates/traces/migrations/0008_agent_traces_by_key.sql

@@ -0,0 +1,24 @@
+CREATE TABLE IF NOT EXISTS {database}.agent_traces_by_key
+(
+    TeamId        LowCardinality(String),
+    ApiKeyHash    String,
+    TraceId       String,
+    StartTs       SimpleAggregateFunction(min, DateTime64(9)),
+    EndTs         SimpleAggregateFunction(max, DateTime64(9)),
+    ServiceName   SimpleAggregateFunction(any, LowCardinality(String)),
+    RootName      SimpleAggregateFunction(anyLast, Nullable(String)),
+    RootInput     SimpleAggregateFunction(anyLast, Nullable(String)),
+    RootStatus    SimpleAggregateFunction(anyLast, Nullable(String)),
+    SpanCount     SimpleAggregateFunction(sum, UInt64),
+    AgentCount    SimpleAggregateFunction(sum, UInt64),
+    LlmCount      SimpleAggregateFunction(sum, UInt64),
+    ToolCount     SimpleAggregateFunction(sum, UInt64),
+    ErrorCount    SimpleAggregateFunction(sum, UInt64),
+    InputTokens   SimpleAggregateFunction(sum, UInt64),
+    OutputTokens  SimpleAggregateFunction(sum, UInt64),
+    Models        SimpleAggregateFunction(groupUniqArrayArray, Array(String)),
+    AgentNames    SimpleAggregateFunction(groupUniqArrayArray, Array(String)),
+    RequestIds    SimpleAggregateFunction(groupArrayArray, Array(String))
+)
+ENGINE = AggregatingMergeTree
+ORDER BY (TeamId, ApiKeyHash, TraceId)

litellm-rust/crates/traces/migrations/0009_agent_traces_by_key_mv.sql

@@ -0,0 +1,22 @@
+CREATE MATERIALIZED VIEW IF NOT EXISTS {database}.agent_traces_by_key_mv
+TO {database}.agent_traces_by_key AS
+SELECT
+    TeamId, ApiKeyHash, TraceId,
+    min(Timestamp)                                         AS StartTs,
+    max(Timestamp + toIntervalNanosecond(Duration))        AS EndTs,
+    any(ServiceName)                                       AS ServiceName,
+    anyLastIf(toNullable(SpanName), ParentSpanId = '')     AS RootName,
+    anyLastIf(toNullable(InputPreview), ParentSpanId = '') AS RootInput,
+    anyLastIf(toNullable(StatusCode), ParentSpanId = '')   AS RootStatus,
+    count()                                                AS SpanCount,
+    countIf(ObservationType = 'agent')                     AS AgentCount,
+    countIf(ObservationType = 'llm')                       AS LlmCount,
+    countIf(ObservationType = 'tool')                      AS ToolCount,
+    countIf(StatusCode = 'STATUS_CODE_ERROR')              AS ErrorCount,
+    sum(InputTokens)                                       AS InputTokens,
+    sum(OutputTokens)                                      AS OutputTokens,
+    groupUniqArrayIf(toString(Model), Model != '')         AS Models,
+    groupUniqArrayIf(SpanName, ObservationType = 'agent')  AS AgentNames,
+    groupArrayIf(LiteLLMRequestId, LiteLLMRequestId != '') AS RequestIds
+FROM {database}.otel_traces
+GROUP BY TeamId, ApiKeyHash, TraceId

litellm-rust/crates/traces/src/error.rs

@@ -2,6 +2,8 @@
 pub enum Error {
     #[error("invalid ClickHouse insert row")]
     InvalidRow,
+    #[error("invalid ClickHouse insert table")]
+    InvalidTable,
     #[error("invalid ClickHouse HTTP URL")]
     InvalidUrl,
     #[error("database must be a nonempty SQL identifier and retention must be positive")]
@@ -10,6 +12,10 @@ pub enum Error {
     EmptySql,
     #[error("ClickHouse query failed with HTTP status {0}")]
     QueryFailed(u16),
+    #[error("ClickHouse insert failed with HTTP status {0}")]
+    InsertFailed(u16),
+    #[error("ClickHouse insert exceeds the encoded size limit")]
+    InsertTooLarge,
     #[error("ClickHouse schema setup failed with HTTP status {0}")]
     SchemaFailed(u16),
     #[error("ClickHouse query exceeded the response size limit")]
@@ -19,3 +25,11 @@ pub enum Error {
     #[error("ClickHouse query transport failed")]
     Transport,
 }
+
+#[derive(Debug, thiserror::Error)]
+pub enum DecodeError {
+    #[error("invalid OTLP trace payload")]
+    InvalidPayload,
+    #[error("OTLP trace payload exceeds the decompressed size limit")]
+    TooLarge,
+}

litellm-rust/crates/traces/src/insert.rs

@@ -1,21 +1,108 @@
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, io::Write, time::Duration};
 
+use flate2::{Compression, write::GzEncoder};
+use litellm_http::Client;
 use serde_json::Value;
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 
-use crate::Error;
+use crate::{Connection, Error};
+
+const MAX_INSERT_BYTES: usize = 64 * 1024 * 1024;
+const INSERT_TIMEOUT: Duration = Duration::from_secs(30);
+
+pub enum InsertTable {
+    OtelTraces,
+    SpendLogs,
+}
+
+impl InsertTable {
+    pub fn parse(value: &str) -> Result<Self, Error> {
+        match value {
+            "otel_traces" => Ok(Self::OtelTraces),
+            "spend_logs" => Ok(Self::SpendLogs),
+            _ => Err(Error::InvalidTable),
+        }
+    }
+
+    fn name(&self) -> &'static str {
+        match self {
+            Self::OtelTraces => "otel_traces",
+            Self::SpendLogs => "spend_logs",
+        }
+    }
+}
+
+pub async fn insert_rows(
+    client: &Client,
+    connection: &Connection,
+    database: &str,
+    table: InsertTable,
+    rows: Vec<BTreeMap<String, Value>>,
+) -> Result<(), Error> {
+    if rows.is_empty() {
+        return Ok(());
+    }
+    let encoded = encode_rows_with_limit(rows, MAX_INSERT_BYTES)?;
+    let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+    encoder
+        .write_all(encoded.as_bytes())
+        .map_err(|_| Error::InvalidRow)?;
+    let body = encoder.finish().map_err(|_| Error::InvalidRow)?;
+    let mut url = connection.url().clone();
+    url.query_pairs_mut()
+        .append_pair(
+            "query",
+            &format!(
+                "INSERT INTO `{database}`.{} FORMAT JSONEachRow",
+                table.name()
+            ),
+        )
+        .append_pair("async_insert", "1")
+        .append_pair("wait_for_async_insert", "1")
+        .append_pair("date_time_input_format", "best_effort");
+    let response = client
+        .post(url)
+        .timeout(INSERT_TIMEOUT)
+        .header("Content-Encoding", "gzip")
+        .body(body)
+        .send()
+        .await
+        .map_err(|_| Error::Transport)?;
+    if !response.status().is_success() {
+        return Err(Error::InsertFailed(response.status().as_u16()));
+    }
+    Ok(())
+}
 
 pub fn encode_rows(rows: Vec<BTreeMap<String, Value>>) -> Result<String, Error> {
-    rows.into_iter()
-        .map(|row| {
-            let encoded = row
-                .into_iter()
-                .map(|(name, value)| insert_value(&name, value).map(|value| (name, value)))
-                .collect::<Result<BTreeMap<_, _>, _>>()?;
-            serde_json::to_string(&encoded).map_err(|_| Error::InvalidRow)
-        })
-        .collect::<Result<Vec<_>, _>>()
-        .map(|rows| rows.join("\n"))
+    encode_rows_with_limit(rows, usize::MAX)
+}
+
+fn encode_rows_with_limit(
+    rows: Vec<BTreeMap<String, Value>>,
+    limit: usize,
+) -> Result<String, Error> {
+    let mut body = Vec::new();
+    for row in rows {
+        let encoded = row
+            .into_iter()
+            .map(|(name, value)| insert_value(&name, value).map(|value| (name, value)))
+            .collect::<Result<BTreeMap<_, _>, _>>()?;
+        let record = serde_json::to_vec(&encoded).map_err(|_| Error::InvalidRow)?;
+        let size = body
+            .len()
+            .checked_add(record.len())
+            .and_then(|size| size.checked_add(usize::from(!body.is_empty())))
+            .ok_or(Error::InsertTooLarge)?;
+        if size > limit {
+            return Err(Error::InsertTooLarge);
+        }
+        if !body.is_empty() {
+            body.push(b'\n');
+        }
+        body.extend_from_slice(&record);
+    }
+    String::from_utf8(body).map_err(|_| Error::InvalidRow)
 }
 
 fn insert_value(name: &str, value: Value) -> Result<Value, Error> {
@@ -35,3 +122,29 @@ fn insert_value(name: &str, value: Value) -> Result<Value, Error> {
         .map(Value::String)
         .map_err(|_| Error::InvalidRow)
 }
+
+#[cfg(test)]
+mod tests {
+    use std::collections::BTreeMap;
+
+    use rstest::rstest;
+    use serde_json::json;
+
+    use super::encode_rows_with_limit;
+    use crate::Error;
+
+    #[rstest]
+    fn encoded_limit_counts_utf8_bytes_across_rows() {
+        let rows = vec![
+            BTreeMap::from([("Input".to_owned(), json!("雪"))]),
+            BTreeMap::from([("Input".to_owned(), json!("雪"))]),
+        ];
+        let encoded = encode_rows_with_limit(rows.clone(), usize::MAX).expect("valid rows");
+
+        assert!(encode_rows_with_limit(rows.clone(), encoded.len()).is_ok());
+        assert!(matches!(
+            encode_rows_with_limit(rows, encoded.len() - 1),
+            Err(Error::InsertTooLarge)
+        ));
+    }
+}

litellm-rust/crates/traces/src/lib.rs

@@ -1,10 +1,12 @@
 mod error;
 mod insert;
+mod otlp;
 mod schema;
 mod sql;
 
-pub use error::Error;
-pub use insert::encode_rows;
+pub use error::{DecodeError, Error};
+pub use insert::{InsertTable, encode_rows, insert_rows};
+pub use otlp::{DecodedSpan, decode_otlp};
 pub use schema::{ensure_schema, schema_statements};
 pub use sql::{Parameter, execute_read};
 use url::Url;
@@ -53,17 +55,32 @@ impl Connection {
         Ok(connection)
     }
 
-    pub fn writer(url: &str, user: &str, password: &str) -> Result<Self, Error> {
+    pub fn writer(url: &str) -> Result<Self, Error> {
         let mut connection = Self::parse(url)?;
+        let pairs: Vec<_> = connection
+            .url
+            .query_pairs()
+            .filter(|(key, _)| !matches!(key.as_ref(), "database" | "readonly" | "query"))
+            .map(|(key, value)| (key.into_owned(), value.into_owned()))
+            .collect();
+        connection.url.query_pairs_mut().clear().extend_pairs(pairs);
+        Ok(connection)
+    }
+
+    pub fn reader(url: &str, database: &str) -> Result<Self, Error> {
+        let mut connection = Self::parse(url)?;
+        let pairs: Vec<_> = connection
+            .url
+            .query_pairs()
+            .filter(|(key, _)| key != "database")
+            .map(|(key, value)| (key.into_owned(), value.into_owned()))
+            .collect();
         connection
             .url
-            .set_username(user)
-            .map_err(|_| Error::InvalidUrl)?;
-        connection
-            .url
-            .set_password(Some(password))
-            .map_err(|_| Error::InvalidUrl)?;
-        connection.url.set_query(None);
+            .query_pairs_mut()
+            .clear()
+            .extend_pairs(pairs)
+            .append_pair("database", database);
         Ok(connection)
     }
 

litellm-rust/crates/traces/src/otlp.rs

@@ -0,0 +1,221 @@
+use std::{collections::BTreeMap, io::Read};
+
+use base64::Engine;
+use flate2::read::GzDecoder;
+use opentelemetry_proto::tonic::{
+    collector::trace::v1::ExportTraceServiceRequest,
+    common::v1::{AnyValue, KeyValue, any_value::Value as AttributeValue},
+    trace::v1::{Span, span::SpanKind, status::StatusCode},
+};
+use prost::Message;
+use serde::Serialize;
+use serde_json::Value;
+
+use crate::DecodeError;
+
+#[derive(Serialize)]
+pub struct DecodedEvent {
+    pub name: String,
+    pub attributes: BTreeMap<String, String>,
+}
+
+#[derive(Serialize)]
+pub struct DecodedSpan {
+    pub trace_id: String,
+    pub span_id: String,
+    pub parent_span_id: String,
+    pub trace_state: String,
+    pub name: String,
+    pub kind: String,
+    pub resource_attributes: BTreeMap<String, String>,
+    pub scope_name: String,
+    pub scope_version: String,
+    pub attributes: BTreeMap<String, String>,
+    pub start_ns: u64,
+    pub end_ns: u64,
+    pub status_code: String,
+    pub status_message: String,
+    pub events: Vec<DecodedEvent>,
+}
+
+pub fn decode_otlp(
+    body: &[u8],
+    content_type: Option<&str>,
+    content_encoding: Option<&str>,
+    max_decompressed_bytes: usize,
+) -> Result<Vec<DecodedSpan>, DecodeError> {
+    let payload = if content_encoding == Some("gzip") || body.starts_with(&[0x1f, 0x8b]) {
+        let limit = u64::try_from(max_decompressed_bytes).map_err(|_| DecodeError::TooLarge)?;
+        let mut decoded = Vec::new();
+        GzDecoder::new(body)
+            .take(limit + 1)
+            .read_to_end(&mut decoded)
+            .map_err(|_| DecodeError::InvalidPayload)?;
+        decoded
+    } else {
+        body.to_vec()
+    };
+    if payload.len() > max_decompressed_bytes {
+        return Err(DecodeError::TooLarge);
+    }
+    let request = if content_type.is_some_and(|value| value.contains("json")) {
+        let value: Value =
+            serde_json::from_slice(&payload).map_err(|_| DecodeError::InvalidPayload)?;
+        serde_json::from_value(normalize_json_ids(value)?)
+            .map_err(|_| DecodeError::InvalidPayload)?
+    } else {
+        ExportTraceServiceRequest::decode(payload.as_slice())
+            .map_err(|_| DecodeError::InvalidPayload)?
+    };
+    Ok(request
+        .resource_spans
+        .into_iter()
+        .flat_map(|resource_spans| {
+            let resource_attributes = attributes(
+                resource_spans
+                    .resource
+                    .map(|resource| resource.attributes)
+                    .unwrap_or_default(),
+            );
+            resource_spans
+                .scope_spans
+                .into_iter()
+                .flat_map(move |scope_spans| {
+                    let scope = scope_spans.scope.unwrap_or_default();
+                    let resource_attributes = resource_attributes.clone();
+                    scope_spans.spans.into_iter().map(move |span| {
+                        decoded_span(span, &resource_attributes, &scope.name, &scope.version)
+                    })
+                })
+        })
+        .collect())
+}
+
+fn normalize_json_ids(value: Value) -> Result<Value, DecodeError> {
+    match value {
+        Value::Object(fields) => fields
+            .into_iter()
+            .map(|(name, value)| {
+                let normalized = if matches!(name.as_str(), "traceId" | "spanId" | "parentSpanId") {
+                    let encoded = value.as_str().ok_or(DecodeError::InvalidPayload)?;
+                    let bytes = base64::engine::general_purpose::STANDARD
+                        .decode(encoded)
+                        .map_err(|_| DecodeError::InvalidPayload)?;
+                    Value::String(hex_bytes(&bytes))
+                } else if name == "kind" && value.is_string() {
+                    let kind = SpanKind::from_str_name(value.as_str().unwrap_or_default())
+                        .ok_or(DecodeError::InvalidPayload)?;
+                    Value::from(kind as i32)
+                } else if name == "code" && value.is_string() {
+                    let code = StatusCode::from_str_name(value.as_str().unwrap_or_default())
+                        .ok_or(DecodeError::InvalidPayload)?;
+                    Value::from(code as i32)
+                } else {
+                    normalize_json_ids(value)?
+                };
+                Ok((name, normalized))
+            })
+            .collect::<Result<serde_json::Map<_, _>, _>>()
+            .map(Value::Object),
+        Value::Array(values) => values
+            .into_iter()
+            .map(normalize_json_ids)
+            .collect::<Result<Vec<_>, _>>()
+            .map(Value::Array),
+        value => Ok(value),
+    }
+}
+
+fn hex_bytes(bytes: &[u8]) -> String {
+    bytes.iter().map(|byte| format!("{byte:02x}")).collect()
+}
+
+fn decoded_span(
+    span: Span,
+    resource_attributes: &BTreeMap<String, String>,
+    scope_name: &str,
+    scope_version: &str,
+) -> DecodedSpan {
+    let status = span.status.unwrap_or_default();
+    DecodedSpan {
+        trace_id: hex_bytes(&span.trace_id),
+        span_id: hex_bytes(&span.span_id),
+        parent_span_id: hex_bytes(&span.parent_span_id),
+        trace_state: span.trace_state,
+        name: span.name,
+        kind: SpanKind::try_from(span.kind)
+            .unwrap_or(SpanKind::Unspecified)
+            .as_str_name()
+            .to_owned(),
+        resource_attributes: resource_attributes.clone(),
+        scope_name: scope_name.to_owned(),
+        scope_version: scope_version.to_owned(),
+        attributes: attributes(span.attributes),
+        start_ns: span.start_time_unix_nano,
+        end_ns: span.end_time_unix_nano,
+        status_code: StatusCode::try_from(status.code)
+            .unwrap_or(StatusCode::Unset)
+            .as_str_name()
+            .to_owned(),
+        status_message: status.message,
+        events: span
+            .events
+            .into_iter()
+            .map(|event| DecodedEvent {
+                name: event.name,
+                attributes: attributes(event.attributes),
+            })
+            .collect(),
+    }
+}
+
+fn attributes(values: Vec<KeyValue>) -> BTreeMap<String, String> {
+    values
+        .into_iter()
+        .map(|entry| {
+            (
+                entry.key,
+                entry.value.as_ref().map(attribute_text).unwrap_or_default(),
+            )
+        })
+        .collect()
+}
+
+fn attribute_text(value: &AnyValue) -> String {
+    match value.value.as_ref() {
+        Some(AttributeValue::StringValue(value)) => value.clone(),
+        Some(AttributeValue::BoolValue(value)) => value.to_string(),
+        Some(AttributeValue::IntValue(value)) => value.to_string(),
+        Some(AttributeValue::DoubleValue(value)) => {
+            serde_json::to_string(value).unwrap_or_default()
+        }
+        Some(AttributeValue::BytesValue(value)) => String::from_utf8_lossy(value).into_owned(),
+        Some(AttributeValue::ArrayValue(value)) => format!(
+            "[{}]",
+            value
+                .values
+                .iter()
+                .map(|value| serde_json::to_string(&attribute_text(value)).unwrap_or_default())
+                .collect::<Vec<_>>()
+                .join(", ")
+        ),
+        Some(AttributeValue::KvlistValue(value)) => format!(
+            "{{{}}}",
+            value
+                .values
+                .iter()
+                .map(|entry| format!(
+                    "{}: {}",
+                    serde_json::to_string(&entry.key).unwrap_or_default(),
+                    serde_json::to_string(
+                        &entry.value.as_ref().map(attribute_text).unwrap_or_default()
+                    )
+                    .unwrap_or_default()
+                ))
+                .collect::<Vec<_>>()
+                .join(", ")
+        ),
+        Some(AttributeValue::StringValueStrindex(value)) => value.to_string(),
+        None => String::new(),
+    }
+}

litellm-rust/crates/traces/src/schema.rs

@@ -4,7 +4,9 @@ use std::time::Duration;
 use crate::Connection;
 use crate::Error;
 
-const MIGRATIONS: [&str; 7] = [
+const SCHEMA_REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
+const MIGRATIONS: [&str; 9] = [
     include_str!("../migrations/0001_otel_traces.sql"),
     include_str!("../migrations/0002_agent_traces.sql"),
     include_str!("../migrations/0003_agent_traces_mv.sql"),
@@ -12,6 +14,8 @@ const MIGRATIONS: [&str; 7] = [
     include_str!("../migrations/0005_otel_traces_ttl.sql"),
     include_str!("../migrations/0006_agent_traces_ttl.sql"),
     include_str!("../migrations/0007_spend_logs_ttl.sql"),
+    include_str!("../migrations/0008_agent_traces_by_key.sql"),
+    include_str!("../migrations/0009_agent_traces_by_key_mv.sql"),
 ];
 
 pub fn schema_statements(
@@ -49,11 +53,30 @@ pub async fn ensure_schema(
     database: &str,
     trace_retention_days: u32,
     spend_log_retention_days: u32,
+) -> Result<(), Error> {
+    ensure_schema_with_timeout(
+        client,
+        connection,
+        database,
+        trace_retention_days,
+        spend_log_retention_days,
+        SCHEMA_REQUEST_TIMEOUT,
+    )
+    .await
+}
+
+async fn ensure_schema_with_timeout(
+    client: &Client,
+    connection: &Connection,
+    database: &str,
+    trace_retention_days: u32,
+    spend_log_retention_days: u32,
+    request_timeout: Duration,
 ) -> Result<(), Error> {
     for statement in schema_statements(database, trace_retention_days, spend_log_retention_days)? {
         let response = client
             .post(connection.url().clone())
-            .timeout(Duration::from_secs(10))
+            .timeout(request_timeout)
             .body(statement)
             .send()
             .await

litellm-rust/crates/traces/tests/admin_sql.rs

@@ -40,7 +40,12 @@ async fn database() -> Result<Database, Box<dyn std::error::Error>> {
         "CREATE DATABASE litellm",
         "CREATE TABLE litellm.otel_traces (n UInt8) ENGINE = Memory",
         "INSERT INTO litellm.otel_traces VALUES (1)",
+        "CREATE TABLE litellm.agent_traces (n UInt8) ENGINE = Memory",
+        "INSERT INTO litellm.agent_traces VALUES (2)",
+        "CREATE TABLE litellm.spend_logs (n UInt8) ENGINE = Memory",
+        "INSERT INTO litellm.spend_logs VALUES (3)",
         "CREATE TABLE litellm.private_traces (n UInt8) ENGINE = Memory",
+        "CREATE TABLE private_traces (n UInt8) ENGINE = Memory",
     ] {
         client
             .post(&admin_url)

litellm-rust/crates/traces/tests/migrations.rs

@@ -154,6 +154,50 @@ async fn schema_supports_span_rollups_and_spend_joins(
     Ok(())
 }
 
+#[rstest]
+#[tokio::test]
+async fn keyed_rollup_keeps_same_trace_ids_separate_by_api_key(
+    #[future(awt)] database: TestResult<ClickHouseDatabase>,
+) -> TestResult {
+    let database = database?;
+    let writer = Connection::writer(&database.url, "default", "")?;
+    ensure_schema(&database.client, &writer, "trace_test", 7, 14).await?;
+    let timestamp = time::OffsetDateTime::now_utc().unix_timestamp_nanos() as i64;
+    let rows = vec![
+        serde_json::from_value(serde_json::json!({
+            "Timestamp": timestamp, "TraceId": "shared-id", "SpanId": "root-one",
+            "ParentSpanId": "", "SpanName": "root-one", "Input": "private-one",
+            "ResourceAttributes": {"litellm.api_key_hash": "key-one"}
+        }))?,
+        serde_json::from_value(serde_json::json!({
+            "Timestamp": timestamp, "TraceId": "shared-id", "SpanId": "root-two",
+            "ParentSpanId": "", "SpanName": "root-two", "Input": "private-two",
+            "ResourceAttributes": {"litellm.api_key_hash": "key-two"}
+        }))?,
+    ];
+    insert_rows(&database, "otel_traces", rows).await?;
+    execute_write(
+        &database,
+        "OPTIMIZE TABLE trace_test.agent_traces_by_key FINAL",
+    )
+    .await?;
+    let rows = read_json(
+        &database,
+        "SELECT ApiKeyHash, any(RootInput) AS RootInput \
+         FROM trace_test.agent_traces_by_key WHERE TraceId = 'shared-id' \
+         GROUP BY ApiKeyHash ORDER BY ApiKeyHash",
+    )
+    .await?;
+    assert_eq!(
+        rows["data"],
+        serde_json::json!([
+            {"ApiKeyHash": "key-one", "RootInput": "private-one"},
+            {"ApiKeyHash": "key-two", "RootInput": "private-two"}
+        ])
+    );
+    Ok(())
+}
+
 #[rstest]
 #[tokio::test]
 async fn rollup_merges_spans_across_days_without_losing_root_fields(

litellm-rust/crates/traces/tests/otlp.rs

@@ -0,0 +1,47 @@
+use flate2::{Compression, write::GzEncoder};
+use litellm_traces::decode_otlp;
+use rstest::rstest;
+use std::io::Write;
+
+const FIXTURE: &[u8] = include_bytes!(
+    "../../../../tests/test_litellm/tracing/fixtures/langsmith_deep_agent_export.json"
+);
+
+#[rstest]
+#[case::json(FIXTURE, Some("application/json"), None)]
+#[case::gzip_json(FIXTURE, Some("application/json"), Some("gzip"))]
+fn decodes_neutral_spans(
+    #[case] body: &[u8],
+    #[case] content_type: Option<&str>,
+    #[case] content_encoding: Option<&str>,
+) {
+    let payload = if content_encoding == Some("gzip") {
+        let mut encoder = GzEncoder::new(Vec::new(), Compression::default());
+        encoder.write_all(body).expect("gzip input");
+        encoder.finish().expect("gzip payload")
+    } else {
+        body.to_vec()
+    };
+    let spans = decode_otlp(&payload, content_type, content_encoding, 8 * 1024 * 1024)
+        .expect("valid OTLP export");
+    assert_eq!(spans.len(), 6);
+    assert_eq!(spans[0].trace_id, "4bad42b84e9de3ba46fc870185f8f023");
+    assert_eq!(spans[0].resource_attributes["service.name"], "agent-demo");
+    assert_eq!(spans[0].scope_name, "langsmith");
+    assert!(
+        spans
+            .iter()
+            .any(|span| span.attributes.contains_key("gen_ai.prompt"))
+    );
+}
+
+#[rstest]
+#[case::invalid(b"not protobuf", None, 8 * 1024 * 1024)]
+#[case::too_large(FIXTURE, Some("application/json"), 1)]
+fn rejects_invalid_or_oversized_payload(
+    #[case] body: &[u8],
+    #[case] content_type: Option<&str>,
+    #[case] limit: usize,
+) {
+    assert!(decode_otlp(body, content_type, None, limit).is_err());
+}

litellm/constants.py

@@ -46,6 +46,19 @@ ROUTER_SETTINGS_MANAGED_OUTSIDE_CONFIG: Final[frozenset[str]] = frozenset(
 )
 DEFAULT_BATCH_SIZE: Final = int(os.getenv("DEFAULT_BATCH_SIZE", 512))
 DEFAULT_FLUSH_INTERVAL_SECONDS: Final = int(os.getenv("DEFAULT_FLUSH_INTERVAL_SECONDS", 5))
+# Agent tracing / ClickHouse
+CLICKHOUSE_BATCH_SIZE: Final = int(os.getenv("CLICKHOUSE_BATCH_SIZE", 10_000))
+CLICKHOUSE_FLUSH_INTERVAL_SECONDS: Final = float(os.getenv("CLICKHOUSE_FLUSH_INTERVAL_SECONDS", 1.0))
+CLICKHOUSE_MAX_BUFFERED_ROWS: Final = int(os.getenv("CLICKHOUSE_MAX_BUFFERED_ROWS", 200_000))
+CLICKHOUSE_MAX_RETRIES: Final = int(os.getenv("CLICKHOUSE_MAX_RETRIES", 3))
+AGENT_TRACING_RETENTION_DAYS: Final = int(os.getenv("AGENT_TRACING_RETENTION_DAYS", 30))
+AGENT_TRACING_SPEND_LOG_RETENTION_DAYS: Final = int(os.getenv("AGENT_TRACING_SPEND_LOG_RETENTION_DAYS", 90))
+OTLP_MAX_BODY_BYTES: Final = int(os.getenv("OTLP_MAX_BODY_BYTES", 8 * 1024 * 1024))
+OTLP_MAX_ATTRIBUTE_VALUE_BYTES: Final = int(os.getenv("OTLP_MAX_ATTRIBUTE_VALUE_BYTES", 64 * 1024))
+OTLP_RETRY_AFTER_SECONDS: Final = int(os.getenv("OTLP_RETRY_AFTER_SECONDS", 2))
+OTLP_OFFLOAD_DECODE_BYTES: Final = int(os.getenv("OTLP_OFFLOAD_DECODE_BYTES", 256 * 1024))
+AGENT_TRACING_INPUT_PREVIEW_CHARS: Final = int(os.getenv("AGENT_TRACING_INPUT_PREVIEW_CHARS", 240))
+AGENT_TRACING_LIST_PAGE_SIZE: Final = int(os.getenv("AGENT_TRACING_LIST_PAGE_SIZE", 50))
 DEFAULT_S3_FLUSH_INTERVAL_SECONDS: Final = int(os.getenv("DEFAULT_S3_FLUSH_INTERVAL_SECONDS", 10))
 DEFAULT_S3_BATCH_SIZE: Final = int(os.getenv("DEFAULT_S3_BATCH_SIZE", 512))
 DEFAULT_S3_MAX_CONCURRENT_UPLOADS: Final = int(os.getenv("DEFAULT_S3_MAX_CONCURRENT_UPLOADS", "16"))

litellm/integrations/clickhouse/clickhouse_batch_logger.py

@@ -0,0 +1,101 @@
+"""
+Shared base for everything LiteLLM writes to ClickHouse.
+
+Built on `CustomBatchLogger`: rows accumulate in `log_queue` and are flushed as one
+gzip JSONEachRow insert, either every `CLICKHOUSE_FLUSH_INTERVAL_SECONDS` or as soon as
+`batch_size` rows are queued. Subclasses only pick the table and build rows:
+
+- `ClickHouseSpendLogger`  -> spend_logs   (LiteLLM requests, via the `clickhouse` callback)
+"""
+
+import asyncio
+import os
+from typing import Any, ClassVar
+
+from litellm._logging import verbose_logger
+from litellm.constants import (
+    CLICKHOUSE_BATCH_SIZE,
+    CLICKHOUSE_FLUSH_INTERVAL_SECONDS,
+    CLICKHOUSE_MAX_BUFFERED_ROWS,
+    CLICKHOUSE_MAX_RETRIES,
+)
+from litellm.integrations.custom_batch_logger import CustomBatchLogger
+from litellm.rust_bridge.traces import TraceStorage
+
+
+def clickhouse_storage_from_env() -> TraceStorage:
+    return TraceStorage(
+        database=os.getenv("CLICKHOUSE_DATABASE", "litellm"),
+        url=os.getenv("CLICKHOUSE_URL", ""),
+    )
+
+
+class ClickHouseBatchLogger(CustomBatchLogger):
+    table: ClassVar[str]
+
+    def __init__(self, storage: TraceStorage | None = None, **kwargs: Any) -> None:
+        self.storage = storage or clickhouse_storage_from_env()
+        self.rows_written = 0
+        self.rows_dropped = 0
+        self._failed_attempts = 0
+        super().__init__(
+            flush_lock=asyncio.Lock(),
+            batch_size=CLICKHOUSE_BATCH_SIZE,
+            flush_interval=CLICKHOUSE_FLUSH_INTERVAL_SECONDS,
+            **kwargs,
+        )
+        try:
+            asyncio.get_running_loop().create_task(self.periodic_flush())
+        except RuntimeError:  # no loop yet (e.g. sync config load); proxy startup calls start()
+            pass
+
+    def start(self) -> None:
+        asyncio.get_running_loop().create_task(self.periodic_flush())
+
+    def is_full(self) -> bool:
+        """Backpressure signal: producers should reject (429) instead of enqueueing."""
+        return len(self.log_queue) >= CLICKHOUSE_MAX_BUFFERED_ROWS
+
+    def enqueue(self, rows: list[dict[str, Any]]) -> None:
+        """Never awaits ClickHouse. Kicks off an early flush once a full batch is queued."""
+        self.log_queue.extend(rows)
+        if len(self.log_queue) >= self.batch_size:
+            asyncio.get_running_loop().create_task(self.flush_queue())
+
+    async def flush_queue(self) -> None:
+        # Swap the queue under the lock so rows enqueued during the insert are kept.
+        if self.flush_lock is None:
+            return
+        async with self.flush_lock:
+            while self.log_queue:
+                batch = self.log_queue[: self.batch_size]
+                self.log_queue = self.log_queue[len(batch) :]
+                if not await self._insert(batch):
+                    break
+
+    async def async_send_batch(self, *args: Any, **kwargs: Any) -> None:
+        await self.flush_queue()
+
+    async def _insert(self, batch: list[dict[str, Any]]) -> bool:
+        try:
+            await self.storage.insert_rows(self.table, batch)
+            self.rows_written += len(batch)
+            self._failed_attempts = 0
+            return True
+        except Exception as e:
+            self._failed_attempts += 1
+            if self._failed_attempts >= CLICKHOUSE_MAX_RETRIES:
+                self.rows_dropped += len(batch)
+                self._failed_attempts = 0
+                verbose_logger.error(
+                    "ClickHouse: dropped %s rows for %s after %s attempts: %s",
+                    len(batch),
+                    self.table,
+                    CLICKHOUSE_MAX_RETRIES,
+                    e,
+                )
+            else:
+                # put it back; the next periodic flush retries it
+                self.log_queue = batch + self.log_queue
+                verbose_logger.warning("ClickHouse: insert into %s failed, will retry: %s", self.table, e)
+            return False

litellm/integrations/clickhouse/schema.py

@@ -0,0 +1,12 @@
+from typing import Final
+
+from litellm.rust_bridge.traces import TraceStorage
+
+OTEL_TRACES_TABLE: Final = "otel_traces"
+AGENT_TRACES_TABLE: Final = "agent_traces"
+AGENT_TRACES_BY_KEY_TABLE: Final = "agent_traces_by_key"
+SPEND_LOGS_TABLE: Final = "spend_logs"
+
+
+async def ensure_schema(storage: TraceStorage, trace_retention_days: int, spend_log_retention_days: int) -> None:
+    await storage.ensure_schema(trace_retention_days, spend_log_retention_days)

litellm/integrations/custom_batch_logger.py

@@ -27,7 +27,7 @@ class CustomBatchLogger(CustomLogger):
         self,
         flush_lock: asyncio.Lock | None = None,
         batch_size: int | None = None,
-        flush_interval: int | None = None,
+        flush_interval: float | None = None,
         max_queue_size: int | None = None,
         **kwargs,
     ) -> None:

litellm/proxy/_types.py

@@ -520,6 +520,10 @@ class LiteLLMRoutes(enum.Enum):
         "/v1/rag/ingest",
         "/rag/query",
         "/v1/rag/query",
+        # agent tracing: OTLP ingest + reads (scoped to the caller's team in the handler)
+        "/v1/traces",
+        "/v1/traces/{trace_id}",
+        "/v1/traces/{trace_id}/spans/{span_id}",
     ]
 
     anthropic_routes = [

litellm/proxy/common_utils/http_parsing_utils.py

@@ -21,6 +21,9 @@ from litellm.proxy.common_utils.callback_utils import (
 from litellm.types.router import Deployment
 
 _FORM_CONTENT_TYPES: Final[frozenset[str]] = frozenset({"application/x-www-form-urlencoded", "multipart/form-data"})
+# Binary bodies (e.g. OTLP trace exports on POST /v1/traces) are not JSON: arbitrary bytes used to
+# hit the JSON surrogate-repair path and fail auth with a 400. JSON under these types still parses.
+_BINARY_CONTENT_TYPES: Final[frozenset[str]] = frozenset({"application/x-protobuf", "application/protobuf"})
 
 _ANNOTATION_QUALIFIERS: Final[frozenset[object]] = frozenset({Annotated, NotRequired, ReadOnly, Required})
 
@@ -119,6 +122,17 @@ def coerce_numeric_form_fields(
     }
 
 
+def _parse_binary_body(body: bytes) -> dict:
+    """JSON sent under a binary content type still parses; real binary (protobuf) carries no params -> {}."""
+    if not body:
+        return {}
+    try:
+        parsed: Final = orjson.loads(body)
+    except orjson.JSONDecodeError:
+        return {}
+    return parsed if isinstance(parsed, dict) else {}
+
+
 async def _read_request_body(request: Request | None) -> dict:
     """
     Safely read the request body and parse it as JSON.
@@ -141,7 +155,9 @@ async def _read_request_body(request: Request | None) -> dict:
         _request_headers: Final[dict] = _safe_get_request_headers(request=request)
         content_type: Final = _request_headers.get("content-type", "")
 
-        if _is_form_content_type(content_type):
+        if _normalize_media_type(content_type) in _BINARY_CONTENT_TYPES:
+            parsed_body = _parse_binary_body(await request.body())
+        elif _is_form_content_type(content_type):
             try:
                 form_data: Final = await request.form()
             except Exception as e:

litellm/proxy/proxy_server.py

@@ -713,6 +713,7 @@ try:
 except ImportError:
     build_billing_metrics_recorder = None
     shutdown_billing_metrics_recorder = None
+from litellm.proxy import tracing_endpoints
 from litellm.proxy.middleware.admission_control_middleware import (
     AdmissionControlMiddleware,
     admission_control_state,
@@ -844,6 +845,7 @@ from litellm.secret_managers.main import (
     secret_manager_would_be_consulted,
     str_to_bool,
 )
+from litellm.tracing import TraceReceiver
 from litellm.types.integrations.slack_alerting import AlertType, SlackAlertingArgs
 from litellm.types.llms.anthropic import (
     AnthropicMessagesRequest,
@@ -1520,6 +1522,9 @@ async def proxy_startup_event(app: FastAPI) -> AsyncGenerator[None, None]:
                 _tagged.strategy._state_loaded = True
     asyncio.create_task(_adaptive_router_flusher_loop())
 
+    ## [Optional] Initialize agent tracing
+    await ProxyStartupEvent._init_tracing(general_settings)
+
     ## [Optional] Initialize dd tracer
     ProxyStartupEvent._init_dd_tracer()
 
@@ -11309,6 +11314,23 @@ class ProxyStartupEvent:
                 )
             return connected_client
 
+    @classmethod
+    async def _init_tracing(cls, general_settings: dict) -> None:
+        """
+        Enable agent tracing (`POST/GET /v1/traces`) when configured:
+
+            general_settings:
+              tracing:
+                store: clickhouse       # CLICKHOUSE_URL / _USER / _PASSWORD / _DATABASE
+        """
+        settings = general_settings.get("tracing") or {}
+        if settings.get("store") != "clickhouse":
+            return
+        tracing = TraceReceiver.from_env()
+        await tracing.start()
+        tracing_endpoints.receiver = tracing
+        verbose_proxy_logger.info("Agent tracing enabled (store=clickhouse)")
+
     @classmethod
     def _init_dd_tracer(cls):
         """
@@ -19860,6 +19882,7 @@ app.include_router(rag_router)
 app.include_router(video_router)
 app.include_router(container_router)
 app.include_router(search_router)
+app.include_router(tracing_endpoints.router)
 app.include_router(image_router)
 app.include_router(fine_tuning_router)
 app.include_router(credential_router)

litellm/proxy/tracing_endpoints.py

@@ -0,0 +1,131 @@
+"""
+Agent tracing endpoints. Thin wrappers over `TraceReceiver`: auth -> tenant/scope -> one call.
+
+POST /v1/traces                              OTLP/HTTP trace export (protobuf or JSON)
+GET  /v1/traces                              TracePage
+GET  /v1/traces/{trace_id}                   Trace
+GET  /v1/traces/{trace_id}/spans/{span_id}   SpanDetail
+"""
+
+import time
+from typing import Final
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, Response
+
+from litellm.constants import OTLP_MAX_BODY_BYTES, OTLP_RETRY_AFTER_SECONDS
+from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+from litellm.tracing import (
+    Tenant,
+    TraceReceiver,
+    TracingPayloadTooLargeError,
+)
+from litellm.tracing.decode import encode_otlp_response
+from litellm.tracing.types import SpanDetail, Trace, TracePage, TraceScope
+
+router = APIRouter(tags=["agent tracing"])
+
+MS_PER_DAY: Final = 24 * 60 * 60 * 1000
+_ADMIN_ROLES: Final = (LitellmUserRoles.PROXY_ADMIN, LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY)
+
+receiver: TraceReceiver | None = None
+
+
+def get_receiver() -> TraceReceiver:
+    if receiver is None:
+        raise HTTPException(
+            status_code=501,
+            detail="Agent tracing is not enabled. Set `tracing:` in general_settings and CLICKHOUSE_URL.",
+        )
+    return receiver
+
+
+def tenant_for(user_api_key_dict: UserAPIKeyAuth) -> Tenant:
+    return Tenant(
+        team_id=user_api_key_dict.team_id or "",
+        api_key_hash=user_api_key_dict.token or "",
+        org_id=user_api_key_dict.org_id or "",
+    )
+
+
+def scope_for(user_api_key_dict: UserAPIKeyAuth) -> TraceScope:
+    """Admins see everything; team members see their team; team-less keys see their own traces."""
+    if user_api_key_dict.user_role in _ADMIN_ROLES:
+        return TraceScope(team_ids=[], api_key_hash="")
+    if user_api_key_dict.team_id:
+        return TraceScope(team_ids=[user_api_key_dict.team_id], api_key_hash="")
+    if not user_api_key_dict.token:
+        raise HTTPException(status_code=403, detail="Not allowed to view agent traces")
+    return TraceScope(team_ids=[""], api_key_hash=user_api_key_dict.token)
+
+
+async def _read_otlp_body(request: Request) -> bytes:
+    body: Final = bytearray()  # mutable-ok: accumulate request chunks without exceeding the configured body limit
+    async for chunk in request.stream():
+        if len(body) + len(chunk) > OTLP_MAX_BODY_BYTES:
+            raise TracingPayloadTooLargeError(f"OTLP body exceeds {OTLP_MAX_BODY_BYTES} bytes")
+        body.extend(chunk)
+    return bytes(body)
+
+
+@router.post("/v1/traces", include_in_schema=False)
+async def ingest_otlp_traces(
+    request: Request,
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+) -> Response:
+    if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY:
+        raise HTTPException(status_code=403, detail="Not allowed to ingest agent traces")
+    tracing: Final = get_receiver()
+    content_type: Final = request.headers.get("content-type")
+    try:
+        await tracing.ingest(
+            body=await _read_otlp_body(request),
+            content_type=content_type,
+            content_encoding=request.headers.get("content-encoding"),
+            tenant=tenant_for(user_api_key_dict),
+        )
+    except RuntimeError:
+        raise HTTPException(status_code=503, headers={"Retry-After": str(OTLP_RETRY_AFTER_SECONDS)})
+    except TracingPayloadTooLargeError as e:
+        raise HTTPException(status_code=413, detail=str(e))
+    body, media_type = encode_otlp_response(content_type)
+    return Response(content=body, media_type=media_type)
+
+
+@router.get("/v1/traces", response_model=None)
+async def list_agent_traces(
+    start_ms: int | None = Query(None, description="Window start, unix ms. Default: 24h ago"),
+    end_ms: int | None = Query(None, description="Window end, unix ms. Default: now"),
+    cursor: str | None = Query(None),
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+) -> TracePage:
+    now_ms: Final = int(time.time() * 1000)
+    return await get_receiver().list_traces(
+        scope=scope_for(user_api_key_dict),
+        start_ms=start_ms if start_ms is not None else now_ms - MS_PER_DAY,
+        end_ms=end_ms if end_ms is not None else now_ms,
+        cursor=cursor,
+    )
+
+
+@router.get("/v1/traces/{trace_id}", response_model=None)
+async def get_agent_trace(
+    trace_id: str,
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+) -> Trace:
+    trace: Final = await get_receiver().get_trace(trace_id, scope_for(user_api_key_dict))
+    if trace is None:
+        raise HTTPException(status_code=404, detail=f"Trace {trace_id} not found")
+    return trace
+
+
+@router.get("/v1/traces/{trace_id}/spans/{span_id}", response_model=None)
+async def get_agent_trace_span(
+    trace_id: str,
+    span_id: str,
+    user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
+) -> SpanDetail:
+    span: Final = await get_receiver().get_span(trace_id, span_id, scope_for(user_api_key_dict))
+    if span is None:
+        raise HTTPException(status_code=404, detail=f"Span {span_id} not found")
+    return span

litellm/rust_bridge/_native.pyi

@@ -11,6 +11,7 @@ from litellm.rust_bridge.embeddings.entrypoints import LiteLLMEmbeddingRequest
 from litellm.rust_bridge.messages.entrypoints import LiteLLMMessagesRequest
 from litellm.rust_bridge.ocr.entrypoints import LiteLLMOcrRequest
 from litellm.rust_bridge.responses.entrypoints import LiteLLMResponsesRequest
+from litellm.rust_bridge.traces import DecodedSpan
 from litellm.types.llms.anthropic_messages.anthropic_response import AnthropicMessagesResponse
 from litellm.types.llms.openai import ResponsesAPIResponse
 from litellm.types.utils import EmbeddingResponse, ModelResponse
@@ -20,18 +21,16 @@ class RustUpstreamError(Exception): ...
 class ForkedAfterNativeRuntimeStarted(RuntimeError): ...
 class ProcessReservedForForking(RuntimeError): ...
 
-def trace_encode_rows(rows: Sequence[Mapping[str, JsonValue]]) -> str: ...
-def trace_ensure_schema(
-    url: str, database: str, user: str, password: str, trace_retention_days: int, spend_log_retention_days: int
-) -> Future[None]: ...
-def trace_query(
-    url: str,
-    database: str,
-    user: str,
-    password: str,
-    sql: str,
-    parameters: Mapping[str, str | int | Sequence[str]],
-) -> Future[str]: ...
+def trace_decode_otlp(
+    body: bytes, content_type: str | None, content_encoding: str | None, max_decompressed_bytes: int
+) -> list[DecodedSpan]: ...
+
+@final
+class NativeTraceStorage:
+    def __new__(cls, database: str, url: str, reader_url: str | None = None) -> NativeTraceStorage: ...
+    def ensure_schema(self, trace_retention_days: int, spend_log_retention_days: int) -> Future[None]: ...
+    def insert_rows(self, table: str, rows: Sequence[Mapping[str, JsonValue]]) -> Future[None]: ...
+    def query(self, sql: str, parameters: Mapping[str, str | int | Sequence[str]]) -> Future[str]: ...
 
 @final
 class NativeDiagnosticProcessor:
@@ -327,6 +326,7 @@ __all__ = [
     "ForkedAfterNativeRuntimeStarted",
     "HuggingFaceEncoding",
     "NativeDiagnosticProcessor",
+    "NativeTraceStorage",
     "ProcessReservedForForking",
     "ResponsesWebSocketConnection",
     "RustBridgeDeclined",
@@ -351,9 +351,7 @@ __all__ = [
     "process_state_started",
     "reserve_process_for_forking",
     "responses",
-    "trace_encode_rows",
-    "trace_ensure_schema",
-    "trace_query",
+    "trace_decode_otlp",
     "transcription",
 ]
 

litellm/rust_bridge/traces.py

@@ -1,33 +1,55 @@
 from collections.abc import Awaitable, Mapping, Sequence
-from typing import Final, Protocol, cast
+from typing import Final, Protocol, TypedDict, cast
 
-from pydantic import BaseModel, ConfigDict, JsonValue
+from pydantic import BaseModel, ConfigDict, JsonValue, TypeAdapter
+from typing_extensions import ReadOnly
 
 from litellm.rust_bridge.loader import get_native_bridge
 
 
+class DecodedEvent(TypedDict):
+    name: ReadOnly[str]
+    attributes: ReadOnly[dict[str, str]]
+
+
+class DecodedSpan(TypedDict):
+    trace_id: ReadOnly[str]
+    span_id: ReadOnly[str]
+    parent_span_id: ReadOnly[str]
+    trace_state: ReadOnly[str]
+    name: ReadOnly[str]
+    kind: ReadOnly[str]
+    resource_attributes: ReadOnly[dict[str, str]]
+    scope_name: ReadOnly[str]
+    scope_version: ReadOnly[str]
+    attributes: ReadOnly[dict[str, str]]
+    start_ns: ReadOnly[int]
+    end_ns: ReadOnly[int]
+    status_code: ReadOnly[str]
+    status_message: ReadOnly[str]
+    events: ReadOnly[list[DecodedEvent]]
+
+
+class NativeStore(Protocol):
+    def __init__(self, database: str, url: str, reader_url: str | None = None) -> None: ...
+
+    def ensure_schema(self, trace_retention_days: int, spend_log_retention_days: int) -> Awaitable[None]: ...
+
+    def insert_rows(self, table: str, rows: Sequence[Mapping[str, JsonValue]]) -> Awaitable[None]: ...
+
+    def query(self, sql: str, parameters: Mapping[str, str | int | Sequence[str]]) -> Awaitable[str]: ...
+
+
 class NativeTraces(Protocol):
-    def trace_encode_rows(self, rows: Sequence[Mapping[str, JsonValue]]) -> str: ...
+    NativeTraceStorage: type[NativeStore]
 
-    def trace_ensure_schema(
+    def trace_decode_otlp(
         self,
-        url: str,
-        database: str,
-        user: str,
-        password: str,
-        trace_retention_days: int,
-        spend_log_retention_days: int,
-    ) -> Awaitable[None]: ...
-
-    def trace_query(
-        self,
-        url: str,
-        database: str,
-        user: str,
-        password: str,
-        sql: str,
-        parameters: Mapping[str, str | int | Sequence[str]],
-    ) -> Awaitable[str]: ...
+        body: bytes,
+        content_type: str | None,
+        content_encoding: str | None,
+        max_decompressed_bytes: int,
+    ) -> list[DecodedSpan]: ...
 
 
 class QueryResponse(BaseModel):
@@ -35,6 +57,10 @@ class QueryResponse(BaseModel):
     data: list[dict[str, JsonValue]]
 
 
+INSERT_ROWS: Final = TypeAdapter(list[dict[str, JsonValue]])
+QUERY_PARAMETERS: Final = TypeAdapter(dict[str, str | int | list[str]])
+
+
 def _native() -> NativeTraces:
     native: Final = get_native_bridge()
     if native is None:
@@ -42,28 +68,22 @@ def _native() -> NativeTraces:
     return cast(NativeTraces, native)  # cast-ok: the native extension is validated against this protocol at call sites
 
 
-async def ensure_schema(
-    url: str,
-    database: str,
-    user: str,
-    password: str,
-    trace_retention_days: int,
-    spend_log_retention_days: int,
-) -> None:
-    await _native().trace_ensure_schema(url, database, user, password, trace_retention_days, spend_log_retention_days)
+def decode_otlp(
+    body: bytes, content_type: str | None, content_encoding: str | None, max_decompressed_bytes: int
+) -> list[DecodedSpan]:
+    return _native().trace_decode_otlp(body, content_type, content_encoding, max_decompressed_bytes)
 
 
-async def query(
-    url: str,
-    database: str,
-    user: str,
-    password: str,
-    sql: str,
-    parameters: Mapping[str, str | int | Sequence[str]],
-) -> list[dict[str, JsonValue]]:
-    result: Final = await _native().trace_query(url, database, user, password, sql, parameters)
-    return QueryResponse.model_validate_json(result).data
+class TraceStorage:
+    def __init__(self, database: str, url: str, reader_url: str | None = None) -> None:
+        self._native: Final = _native().NativeTraceStorage(database, url, reader_url)
 
+    async def ensure_schema(self, trace_retention_days: int, spend_log_retention_days: int) -> None:
+        await self._native.ensure_schema(trace_retention_days, spend_log_retention_days)
 
-def encode_rows(rows: Sequence[Mapping[str, JsonValue]]) -> bytes:
-    return _native().trace_encode_rows(rows).encode("utf-8")
+    async def insert_rows(self, table: str, rows: Sequence[Mapping[str, object]]) -> None:
+        await self._native.insert_rows(table, INSERT_ROWS.validate_python(rows))
+
+    async def query(self, sql: str, parameters: Mapping[str, object] | None = None) -> list[dict[str, JsonValue]]:
+        result: Final = await self._native.query(sql, QUERY_PARAMETERS.validate_python(parameters or {}))
+        return QueryResponse.model_validate_json(result).data

litellm/tracing/AGENTS.md

@@ -0,0 +1,6 @@
+- Python owns tracing endpoints, authenticated tenant scope, framework normalization and API response shaping
+- Trace ingestion awaits `TraceStorage.insert_rows` before returning success; propagate storage failures so OTLP exporters can retry
+- Spend logging keeps its separate batch queue in `litellm/integrations/clickhouse`
+- Use `litellm.rust_bridge.traces.TraceStorage` for ClickHouse; keep schema, SQL, encoding and transport in `litellm-traces`
+- Derive tenant fields from authentication and overwrite matching fields supplied by the exporter
+- Test confirmed writes, failures, tenant isolation and read behavior through public functions

litellm/tracing/__init__.py

@@ -0,0 +1,16 @@
+"""
+LiteLLM agent tracing: OTLP traces from agents, joined to LiteLLM spend logs, in ClickHouse.
+
+"""
+
+from litellm.tracing.receiver import (
+    Tenant,
+    TraceReceiver,
+    TracingPayloadTooLargeError,
+)
+
+__all__ = [
+    "Tenant",
+    "TraceReceiver",
+    "TracingPayloadTooLargeError",
+]

litellm/tracing/decode.py

@@ -0,0 +1,245 @@
+"""
+OTLP/HTTP trace export -> `SpanRow`s.
+
+Pure functions, no I/O. Two steps:
+1. `decode_otlp()`   protobuf / JSON / gzip `ExportTraceServiceRequest` -> flat spans
+2. `normalize()`     framework conventions -> LiteLLM columns (type, agent, input/output,
+                     LiteLLM request id). Supported: LangSmith (LangChain, LangGraph,
+                     Deep Agents), OTEL GenAI semconv, OpenInference.
+"""
+
+import json
+from collections.abc import Callable, Mapping
+from typing import Any, Final
+
+from litellm.constants import OTLP_MAX_ATTRIBUTE_VALUE_BYTES, OTLP_MAX_BODY_BYTES
+from litellm.rust_bridge.traces import DecodedSpan
+from litellm.rust_bridge.traces import decode_otlp as native_decode_otlp
+from litellm.tracing.types import SpanRow, SpanType
+
+# attributes whose content we lift into Input/Output and drop from SpanAttributes
+_HEAVY_ATTRIBUTES: Final = frozenset(
+    {
+        "gen_ai.prompt",
+        "gen_ai.completion",
+        "gen_ai.tool.definitions",
+        "gen_ai.input.messages",
+        "gen_ai.output.messages",
+        "input.value",
+        "output.value",
+    }
+)
+# LangChain / Deep Agents middleware wrappers: real spans, but noise in the UI
+_FRAMEWORK_SUFFIXES: Final = (
+    ".wrap_model_call",
+    ".wrap_tool_call",
+    ".before_agent",
+    ".after_agent",
+    ".before_model",
+    ".after_model",
+)
+_LLM_OPERATIONS: Final = frozenset({"chat", "text_completion", "generate_content"})
+_LC_ROLES: Final = {"human": "user", "ai": "assistant", "system": "system", "tool": "tool"}
+_OPENINFERENCE_TYPES: Final[dict[str, SpanType]] = {"AGENT": "agent", "LLM": "llm", "TOOL": "tool"}
+
+
+# ---------------------------------------------------------------- decode
+
+
+def _truncate(value: str) -> str:
+    size = len(value.encode("utf-8"))
+    if size <= OTLP_MAX_ATTRIBUTE_VALUE_BYTES:
+        return value
+    kept = value.encode("utf-8")[:OTLP_MAX_ATTRIBUTE_VALUE_BYTES].decode("utf-8", "ignore")
+    return f"{kept}…[truncated {size - OTLP_MAX_ATTRIBUTE_VALUE_BYTES} bytes]"
+
+
+def decode_otlp(body: bytes, content_type: str | None = None, content_encoding: str | None = None) -> list[SpanRow]:
+    """Decode an OTLP trace export and normalize every span."""
+    return [_span_row(span) for span in native_decode_otlp(body, content_type, content_encoding, OTLP_MAX_BODY_BYTES)]
+
+
+def _exception_message(span: DecodedSpan) -> str:
+    """`span.record_exception()` writes an `exception` event; surface it when status.message is empty."""
+    for event in span["events"]:
+        if event["name"] == "exception":
+            attributes = event["attributes"]
+            return attributes.get("exception.message") or attributes.get("exception.type", "")
+    return ""
+
+
+def _span_row(span: DecodedSpan) -> SpanRow:
+    attributes = span["attributes"]
+    resource = span["resource_attributes"]
+    row = SpanRow(
+        Timestamp=span["start_ns"],
+        TraceId=span["trace_id"],
+        SpanId=span["span_id"],
+        ParentSpanId=span["parent_span_id"],
+        TraceState=span["trace_state"],
+        SpanName=span["name"],
+        SpanKind=span["kind"],
+        ServiceName=resource.get("service.name", ""),
+        ResourceAttributes=resource,
+        ScopeName=span["scope_name"],
+        ScopeVersion=span["scope_version"],
+        SpanAttributes={},
+        Duration=max(span["end_ns"] - span["start_ns"], 0),
+        StatusCode=span["status_code"],
+        StatusMessage=span["status_message"] or _exception_message(span),
+        TeamId="",
+        ApiKeyHash="",
+        ObservationType="chain",
+        AgentName="",
+        LiteLLMRequestId="",
+        Model="",
+        InputTokens=0,
+        OutputTokens=0,
+        Input="",
+        Output="",
+    )
+    normalize(row, attributes)
+    row["SpanAttributes"] = {k: _truncate(v) for k, v in attributes.items() if k not in _HEAVY_ATTRIBUTES}
+    row["Input"], row["Output"] = _truncate(row["Input"]), _truncate(row["Output"])
+    return row
+
+
+# ---------------------------------------------------------------- normalize
+
+
+def _loads(value: str) -> Any:
+    try:
+        return json.loads(value)
+    except (ValueError, TypeError):
+        return None
+
+
+def _lc_message(message: Mapping[str, Any]) -> dict[str, Any]:
+    """LangChain serialized message (or plain {role, content}) -> {role, content, tool_calls?}."""
+    kwargs = message.get("kwargs", message)
+    role = _LC_ROLES.get(kwargs.get("type") or kwargs.get("role"), kwargs.get("role") or kwargs.get("type") or "")
+    content = kwargs.get("content", "")
+    out: dict[str, Any] = {"role": role, "content": content if isinstance(content, str) else json.dumps(content)}
+    if kwargs.get("tool_calls"):
+        out["tool_calls"] = [{"name": t.get("name"), "args": t.get("args")} for t in kwargs["tool_calls"]]
+    if role == "tool" and kwargs.get("name"):
+        out["name"] = kwargs["name"]
+    return out
+
+
+def _langsmith_type(row: SpanRow, attributes: Mapping[str, str]) -> SpanType:
+    kind = attributes.get("langsmith.span.kind", "chain")
+    name = row["SpanName"]
+    if not row["ParentSpanId"] or name == attributes.get("langsmith.metadata.lc_agent_name"):
+        return "agent"
+    if kind == "llm" or kind == "tool":
+        return kind
+    if name.endswith(_FRAMEWORK_SUFFIXES):
+        return "framework"
+    return "chain"
+
+
+def _langsmith_io(row: SpanRow, attributes: Mapping[str, str]) -> None:
+    prompt = _loads(attributes.get("gen_ai.prompt", ""))
+    completion = _loads(attributes.get("gen_ai.completion", ""))
+    if row["ObservationType"] == "llm" and isinstance(completion, dict):
+        messages = (prompt or {}).get("messages") or [[]]
+        batch = messages[0] if messages and isinstance(messages[0], list) else messages
+        row["Input"] = json.dumps([_lc_message(m) for m in batch])
+        generation = completion["generations"][0][0]["message"]["kwargs"]
+        row["Output"] = json.dumps(_lc_message({"kwargs": generation}))
+        row["LiteLLMRequestId"] = (generation.get("response_metadata") or {}).get("id") or ""
+        return
+    if row["ObservationType"] == "tool":
+        output = (completion or {}).get("output", completion) if isinstance(completion, dict) else completion
+        if isinstance(output, dict) and "update" in output:  # LangGraph Command, e.g. Deep Agents `task`
+            update_messages = (output.get("update") or {}).get("messages") or []
+            output = update_messages[-1] if update_messages else output
+        if isinstance(output, dict):
+            output = output.get("content", output)
+        row["Input"] = attributes.get("gen_ai.prompt", "")
+        row["Output"] = output if isinstance(output, str) else json.dumps(output)
+        return
+    if row["ObservationType"] == "agent":
+        input_messages = (prompt or {}).get("messages") if isinstance(prompt, dict) else None
+        output_messages = (completion or {}).get("messages") if isinstance(completion, dict) else None
+        # agents built with @traceable take arbitrary args, not a message list: keep the raw payload then
+        row["Input"] = (
+            json.dumps([_lc_message(m) for m in input_messages if isinstance(m, dict)])
+            if input_messages
+            else attributes.get("gen_ai.prompt", "")
+        )
+        row["Output"] = (
+            json.dumps(_lc_message(output_messages[-1]))
+            if output_messages and isinstance(output_messages[-1], dict)
+            else attributes.get("gen_ai.completion", "")
+        )
+        return
+    row["Input"] = attributes.get("gen_ai.prompt", "")
+    row["Output"] = attributes.get("gen_ai.completion", "")
+
+
+def normalize_langsmith(row: SpanRow, attributes: Mapping[str, str]) -> None:
+    row["ObservationType"] = _langsmith_type(row, attributes)
+    row["AgentName"] = attributes.get("langsmith.metadata.lc_agent_name", "")
+    row["Model"] = attributes.get("gen_ai.request.model", "")
+    _langsmith_io(row, attributes)
+
+
+def normalize_genai(row: SpanRow, attributes: Mapping[str, str]) -> None:
+    operation = attributes.get("gen_ai.operation.name", "")
+    if operation == "invoke_agent" or not row["ParentSpanId"]:
+        row["ObservationType"] = "agent"
+    elif operation in _LLM_OPERATIONS:
+        row["ObservationType"] = "llm"
+    elif operation == "execute_tool":
+        row["ObservationType"] = "tool"
+    row["AgentName"] = attributes.get("gen_ai.agent.name", "")
+    row["Model"] = attributes.get("gen_ai.request.model") or attributes.get("gen_ai.response.model", "")
+    row["LiteLLMRequestId"] = attributes.get("gen_ai.response.id", "")
+    row["Input"] = attributes.get("gen_ai.input.messages") or attributes.get("gen_ai.tool.call.arguments", "")
+    row["Output"] = attributes.get("gen_ai.output.messages") or attributes.get("gen_ai.tool.call.result", "")
+
+
+def normalize_openinference(row: SpanRow, attributes: Mapping[str, str]) -> None:
+    kind = attributes.get("openinference.span.kind", "").upper()
+    row["ObservationType"] = _OPENINFERENCE_TYPES.get(kind, "agent" if not row["ParentSpanId"] else "chain")
+    row["AgentName"] = attributes.get("agent.name", "")
+    row["Model"] = attributes.get("llm.model_name", "")
+    row["Input"] = attributes.get("input.value", "")
+    row["Output"] = attributes.get("output.value", "")
+    row["InputTokens"] = _to_int(attributes.get("llm.token_count.prompt"))
+    row["OutputTokens"] = _to_int(attributes.get("llm.token_count.completion"))
+
+
+def _set_tokens(row: SpanRow, attributes: Mapping[str, str]) -> None:
+    row["InputTokens"] = _to_int(attributes.get("gen_ai.usage.input_tokens"))
+    row["OutputTokens"] = _to_int(attributes.get("gen_ai.usage.output_tokens"))
+
+
+def _to_int(value: str | None) -> int:
+    try:
+        return int(value) if value else 0
+    except ValueError:
+        return 0
+
+
+def select_normalizer(scope_name: str, attributes: Mapping[str, str]) -> Callable[[SpanRow, Mapping[str, str]], None]:
+    if scope_name == "langsmith" or "langsmith.span.kind" in attributes:
+        return normalize_langsmith
+    if "openinference.span.kind" in attributes:
+        return normalize_openinference
+    return normalize_genai
+
+
+def normalize(row: SpanRow, attributes: Mapping[str, str]) -> None:
+    select_normalizer(row["ScopeName"], attributes)(row, attributes)
+    if not row["InputTokens"] and not row["OutputTokens"]:
+        _set_tokens(row, attributes)
+
+
+def encode_otlp_response(content_type: str | None) -> tuple[bytes, str]:
+    """Empty ExportTraceServiceResponse in the caller's encoding."""
+    if content_type and "json" in content_type:
+        return b"{}", "application/json"
+    return b"", "application/x-protobuf"

litellm/tracing/receiver.py

@@ -0,0 +1,117 @@
+"""
+`TraceReceiver`: the one entry point for agent tracing.
+
+    tracing = TraceReceiver.from_env()          # or TraceReceiver(store=...)
+    await tracing.start()                        # create tables if missing
+
+    tracing.ingest(otlp_body, content_type, content_encoding, tenant)   # POST /v1/traces
+    await tracing.list_traces(scope, start_ms, end_ms, cursor)          # GET  /v1/traces
+    await tracing.get_trace(trace_id, scope)                            # GET  /v1/traces/{id}
+    await tracing.get_span(trace_id, span_id, scope)                    # GET  /v1/traces/{id}/spans/{span_id}
+
+The proxy endpoints are thin wrappers: auth -> build tenant/scope -> call one method.
+"""
+
+import asyncio
+import os
+from typing import Final
+
+from litellm.constants import (
+    AGENT_TRACING_RETENTION_DAYS,
+    AGENT_TRACING_SPEND_LOG_RETENTION_DAYS,
+    OTLP_MAX_BODY_BYTES,
+    OTLP_OFFLOAD_DECODE_BYTES,
+)
+from litellm.integrations.clickhouse.schema import ensure_schema
+from litellm.rust_bridge.traces import TraceStorage
+from litellm.tracing.decode import decode_otlp
+from litellm.tracing.store import ClickHouseTraceStore
+from litellm.tracing.types import (
+    SpanDetail,
+    SpanRow,
+    Trace,
+    TracePage,
+    TraceScope,
+)
+
+
+class TracingPayloadTooLargeError(Exception):
+    pass
+
+
+class Tenant:
+    """Who sent the spans. Always taken from auth, never from span attributes."""
+
+    def __init__(self, team_id: str, api_key_hash: str, org_id: str = ""):
+        self.team_id = team_id
+        self.api_key_hash = api_key_hash
+        self.org_id = org_id
+
+    def stamp(self, row: SpanRow) -> SpanRow:
+        row["TeamId"] = self.team_id
+        row["ApiKeyHash"] = self.api_key_hash
+        row["ResourceAttributes"] = {
+            **row["ResourceAttributes"],
+            "litellm.team_id": self.team_id,
+            "litellm.api_key_hash": self.api_key_hash,
+            "litellm.org_id": self.org_id,
+        }
+        return row
+
+
+class TraceReceiver:
+    def __init__(self, store: ClickHouseTraceStore):
+        self.store = store
+
+    @classmethod
+    def from_env(cls) -> "TraceReceiver":
+        return cls(
+            store=ClickHouseTraceStore(
+                TraceStorage(
+                    database=os.getenv("CLICKHOUSE_DATABASE", "litellm"),
+                    url=os.environ["CLICKHOUSE_URL"],
+                    reader_url=os.environ["CLICKHOUSE_READER_URL"],
+                )
+            )
+        )
+
+    async def start(self) -> None:
+        await ensure_schema(
+            self.store.storage,
+            trace_retention_days=AGENT_TRACING_RETENTION_DAYS,
+            spend_log_retention_days=AGENT_TRACING_SPEND_LOG_RETENTION_DAYS,
+        )
+
+    # ------------------------------------------------------------ write
+
+    async def ingest(
+        self,
+        body: bytes,
+        content_type: str | None,
+        content_encoding: str | None,
+        tenant: Tenant,
+    ) -> int:
+        """Decode an OTLP trace export and store its authenticated spans."""
+        if len(body) > OTLP_MAX_BODY_BYTES:
+            raise TracingPayloadTooLargeError(f"OTLP body exceeds {OTLP_MAX_BODY_BYTES} bytes")
+        rows: Final = (
+            await asyncio.to_thread(decode_otlp, body, content_type, content_encoding)
+            if len(body) > OTLP_OFFLOAD_DECODE_BYTES
+            else decode_otlp(body, content_type, content_encoding)
+        )
+        try:
+            await self.store.insert_spans([tenant.stamp(r) for r in rows])
+        except OverflowError as error:
+            raise TracingPayloadTooLargeError(str(error)) from error
+        return len(rows)
+
+    # ------------------------------------------------------------ read
+
+    async def list_traces(self, scope: TraceScope, start_ms: int, end_ms: int, cursor: str | None = None) -> TracePage:
+        return await self.store.list_traces(scope, start_ms, end_ms, cursor)
+
+    async def get_trace(self, trace_id: str, scope: TraceScope) -> Trace | None:
+        return await self.store.get_trace(trace_id, scope)
+
+    async def get_span(self, trace_id: str, span_id: str, scope: TraceScope) -> SpanDetail | None:
+        return await self.store.get_span(trace_id, span_id, scope)

litellm/tracing/store.py

@@ -0,0 +1,267 @@
+"""ClickHouse-backed trace store: batched span writes and scoped reads."""
+
+import base64
+import json
+from datetime import datetime, timezone
+from typing import Any, Final
+
+from litellm.constants import AGENT_TRACING_LIST_PAGE_SIZE
+from litellm.integrations.clickhouse.schema import (
+    AGENT_TRACES_BY_KEY_TABLE,
+    OTEL_TRACES_TABLE,
+)
+from litellm.rust_bridge.traces import TraceStorage
+from litellm.tracing.types import (
+    AgentNode,
+    Span,
+    SpanDetail,
+    SpanRow,
+    SpanStatus,
+    Trace,
+    TracePage,
+    TraceScope,
+    TraceSummary,
+)
+
+NANOS_PER_MS: Final = 1_000_000
+_STATUS: Final[dict[str, SpanStatus]] = {"STATUS_CODE_OK": "ok", "STATUS_CODE_ERROR": "error"}
+
+_SCOPE_OTEL: Final = (
+    "(empty({team_ids:Array(String)}) OR TeamId IN {team_ids:Array(String)})"
+    " AND ({api_key_hash:String} = '' OR ApiKeyHash = {api_key_hash:String})"
+)
+LIST_TRACES_SQL: Final = f"""
+SELECT t.TraceId AS trace_id, ifNull(any(t.RootName), '') AS name, any(t.ServiceName) AS service,
+       ifNull(any(t.RootInput), '') AS input_preview, ifNull(any(t.RootStatus), '') AS status,
+       toUnixTimestamp64Milli(any(t.StartTs)) AS start_ms,
+       dateDiff('millisecond', any(t.StartTs), any(t.EndTs)) AS duration_ms,
+       any(t.SpanCount) AS span_count, length(any(t.AgentNames)) AS agent_count,
+       any(t.AgentCount) AS agent_invocations,
+       any(t.LlmCount) AS llm_calls, any(t.ToolCount) AS tool_calls,
+       any(t.InputTokens) AS input_tokens, any(t.OutputTokens) AS output_tokens,
+       any(t.Models) AS models, any(t.ErrorCount) AS error_count
+FROM (
+    SELECT TeamId, TraceId, min(StartTs) AS StartTs, max(EndTs) AS EndTs,
+           any(ServiceName) AS ServiceName, anyLast(a.RootName) AS RootName,
+           anyLast(a.RootInput) AS RootInput,
+           anyLast(a.RootStatus) AS RootStatus,
+           sum(SpanCount) AS SpanCount, sum(AgentCount) AS AgentCount, sum(LlmCount) AS LlmCount,
+           sum(ToolCount) AS ToolCount, sum(ErrorCount) AS ErrorCount, sum(InputTokens) AS InputTokens,
+           sum(OutputTokens) AS OutputTokens, groupUniqArrayArray(Models) AS Models,
+           groupUniqArrayArray(AgentNames) AS AgentNames
+    FROM {AGENT_TRACES_BY_KEY_TABLE} AS a
+    WHERE (empty({{team_ids:Array(String)}}) OR TeamId IN {{team_ids:Array(String)}})
+      AND ({{api_key_hash:String}} = '' OR ApiKeyHash = {{api_key_hash:String}})
+    GROUP BY TeamId, TraceId
+    HAVING StartTs >= fromUnixTimestamp64Milli({{start_ms:Int64}})
+       AND StartTs < fromUnixTimestamp64Milli({{end_ms:Int64}})
+       AND (({{cursor_ms:Int64}} = 0) OR (toUnixTimestamp64Milli(StartTs), TraceId)
+            < ({{cursor_ms:Int64}}, {{cursor_trace_id:String}}))
+    ORDER BY StartTs DESC, TraceId DESC
+    LIMIT {{limit:UInt32}}
+) AS t
+GROUP BY t.TraceId
+ORDER BY start_ms DESC, t.TraceId DESC
+"""
+
+TRACE_SPANS_SQL: Final = f"""
+SELECT o.SpanId AS span_id, o.ParentSpanId AS parent_span_id, o.SpanName AS name,
+       o.ObservationType AS type, o.AgentName AS agent, o.StatusCode AS status,
+       o.StatusMessage AS status_message,
+       toUnixTimestamp64Nano(o.Timestamp) AS start_ns, o.Duration AS duration_ns,
+       o.ServiceName AS service, o.InputPreview AS input_preview, o.Model AS model,
+       o.InputTokens AS input_tokens, o.OutputTokens AS output_tokens,
+       o.LiteLLMRequestId AS litellm_request_id
+FROM {OTEL_TRACES_TABLE} AS o
+WHERE o.TraceId = {{trace_id:String}} AND {_SCOPE_OTEL}
+ORDER BY o.Timestamp
+LIMIT 1 BY o.SpanId
+"""
+
+SPAN_DETAIL_SQL: Final = f"""
+SELECT SpanId AS span_id, Input AS input, Output AS output, SpanAttributes AS attributes
+FROM {OTEL_TRACES_TABLE}
+WHERE TraceId = {{trace_id:String}} AND SpanId = {{span_id:String}} AND {_SCOPE_OTEL}
+LIMIT 1
+"""
+
+
+def encode_cursor(start_ms: int, trace_id: str) -> str:
+    return base64.urlsafe_b64encode(json.dumps([start_ms, trace_id]).encode()).decode()
+
+
+def decode_cursor(cursor: str | None) -> tuple[int, str]:
+    if not cursor:
+        return 0, ""
+    start_ms, trace_id = json.loads(base64.urlsafe_b64decode(cursor.encode()))
+    return int(start_ms), str(trace_id)
+
+
+def _iso(ms: int) -> str:
+    return datetime.fromtimestamp(ms / 1000, tz=timezone.utc).isoformat()
+
+
+def _status(code: str) -> SpanStatus:
+    return _STATUS.get(code, "unset")
+
+
+def trace_summary_from_row(row: dict[str, Any]) -> TraceSummary:
+    return TraceSummary(
+        trace_id=row["trace_id"],
+        name=row["name"],
+        service=row["service"],
+        input_preview=row["input_preview"],
+        start_time=_iso(int(row["start_ms"])),
+        duration_ms=float(row["duration_ms"]),
+        status=_status(row["status"]),
+        span_count=int(row["span_count"]),
+        agent_count=int(row["agent_count"]),
+        agent_invocations=int(row.get("agent_invocations") or row["agent_count"]),
+        llm_calls=int(row["llm_calls"]),
+        tool_calls=int(row["tool_calls"]),
+        error_count=int(row.get("error_count") or 0),
+        input_tokens=int(row["input_tokens"]),
+        output_tokens=int(row["output_tokens"]),
+        models=list(row["models"]),
+    )
+
+
+def span_from_row(row: dict[str, Any], trace_start_ns: int) -> Span:
+    return Span(
+        span_id=row["span_id"],
+        parent_span_id=row["parent_span_id"] or None,
+        name=row["name"],
+        type=row["type"],
+        agent=row["agent"],
+        start_offset_ms=(int(row["start_ns"]) - trace_start_ns) / NANOS_PER_MS,
+        duration_ms=int(row["duration_ns"]) / NANOS_PER_MS,
+        status=_status(row["status"]),
+        error=row.get("status_message") or None,
+        input_preview=row["input_preview"],
+        model=row["model"] or None,
+        input_tokens=int(row["input_tokens"]),
+        output_tokens=int(row["output_tokens"]),
+        litellm_request_id=row["litellm_request_id"] or None,
+    )
+
+
+def _parent_agent_of(span: Span, by_id: dict[str, Span]) -> str | None:
+    parent_id = span["parent_span_id"]
+    while parent_id is not None and parent_id in by_id:
+        parent = by_id[parent_id]
+        if parent["type"] == "agent" and parent["name"] != span["name"]:
+            return parent["name"]
+        parent_id = parent["parent_span_id"]
+    return None
+
+
+def agent_nodes(spans: list[Span]) -> list[AgentNode]:
+    """One node per distinct agent name (200 `researcher` invocations = 1 node), with who invoked it."""
+    by_id: Final = {s["span_id"]: s for s in spans}
+    agents: dict[str, AgentNode] = {}
+    for span in spans:
+        if span["type"] != "agent":
+            continue
+        node = agents.setdefault(
+            span["name"],
+            AgentNode(
+                name=span["name"],
+                parent_agent=_parent_agent_of(span, by_id),
+                invocations=0,
+                llm_calls=0,
+                tool_calls=0,
+                duration_ms=0.0,
+            ),
+        )
+        node["invocations"] += 1
+        node["duration_ms"] += span["duration_ms"]
+    for span in spans:
+        owner = agents.get(span["agent"])
+        if owner is None:
+            continue
+        if span["type"] == "llm":
+            owner["llm_calls"] += 1
+        elif span["type"] == "tool":
+            owner["tool_calls"] += 1
+    return list(agents.values())
+
+
+def trace_from_rows(trace_id: str, rows: list[dict[str, Any]]) -> Trace | None:
+    if not rows:
+        return None
+    trace_start_ns: Final = min(int(r["start_ns"]) for r in rows)
+    trace_end_ns: Final = max(int(r["start_ns"]) + int(r["duration_ns"]) for r in rows)
+    spans: Final = [span_from_row(r, trace_start_ns) for r in rows]
+    root: Final = next((s for s in spans if s["parent_span_id"] is None), spans[0])
+    agents: Final = agent_nodes(spans)
+    llm_spans: Final = [s for s in spans if s["type"] == "llm"]
+    return Trace(
+        summary=TraceSummary(
+            trace_id=trace_id,
+            name=root["name"],
+            service=rows[0]["service"],
+            input_preview=root["input_preview"],
+            start_time=_iso(trace_start_ns // NANOS_PER_MS),
+            duration_ms=(trace_end_ns - trace_start_ns) / NANOS_PER_MS,
+            status=root["status"],
+            span_count=len(spans),
+            agent_count=len(agents),
+            agent_invocations=sum(a["invocations"] for a in agents),
+            llm_calls=len(llm_spans),
+            tool_calls=sum(1 for s in spans if s["type"] == "tool"),
+            error_count=sum(1 for s in spans if s["status"] == "error"),
+            input_tokens=sum(s["input_tokens"] for s in spans),
+            output_tokens=sum(s["output_tokens"] for s in spans),
+            models=sorted({s["model"] for s in llm_spans if s["model"]}),
+        ),
+        agents=agents,
+        spans=spans,
+    )
+
+
+class ClickHouseTraceStore:
+    """Stores spans and runs scoped trace reads."""
+
+    def __init__(self, storage: TraceStorage):
+        self.storage = storage
+
+    async def insert_spans(self, rows: list[SpanRow]) -> None:
+        await self.storage.insert_rows(OTEL_TRACES_TABLE, [dict(row) for row in rows])
+
+    async def list_traces(
+        self,
+        scope: TraceScope,
+        start_ms: int,
+        end_ms: int,
+        cursor: str | None = None,
+        limit: int = AGENT_TRACING_LIST_PAGE_SIZE,
+    ) -> TracePage:
+        cursor_ms, cursor_trace_id = decode_cursor(cursor)
+        rows = await self.storage.query(
+            LIST_TRACES_SQL,
+            {
+                **scope,
+                "start_ms": start_ms,
+                "end_ms": end_ms,
+                "cursor_ms": cursor_ms,
+                "cursor_trace_id": cursor_trace_id,
+                "limit": limit,
+            },
+        )
+        next_cursor = encode_cursor(int(rows[-1]["start_ms"]), rows[-1]["trace_id"]) if len(rows) == limit else None
+        return TracePage(data=[trace_summary_from_row(r) for r in rows], next_cursor=next_cursor)
+
+    async def get_trace(self, trace_id: str, scope: TraceScope) -> Trace | None:
+        rows = await self.storage.query(TRACE_SPANS_SQL, {**scope, "trace_id": trace_id})
+        return trace_from_rows(trace_id, rows)
+
+    async def get_span(self, trace_id: str, span_id: str, scope: TraceScope) -> SpanDetail | None:
+        rows = await self.storage.query(SPAN_DETAIL_SQL, {**scope, "trace_id": trace_id, "span_id": span_id})
+        if not rows:
+            return None
+        return SpanDetail(
+            span_id=rows[0]["span_id"],
+            input=rows[0]["input"],
+            output=rows[0]["output"],
+            attributes=dict(rows[0]["attributes"]),
+        )

litellm/tracing/types.py

@@ -0,0 +1,119 @@
+"""
+Agent tracing types.
+
+A trace is one agent run. It's made of spans (agent / llm / tool / chain / framework).
+    Trace
+    ├── summary: TraceSummary
+    ├── agents:  list[AgentNode]      one per distinct agent name (for the agent graph)
+    └── spans:   list[Span]           flat, linked by parent_span_id
+
+"""
+
+from typing import Literal
+
+from typing_extensions import TypedDict
+
+SpanType = Literal["agent", "llm", "tool", "chain", "framework"]
+SpanStatus = Literal["ok", "error", "unset"]
+
+
+class Span(TypedDict):
+    span_id: str
+    parent_span_id: str | None
+    name: str
+    type: SpanType
+    agent: str  # the agent this span runs inside, e.g. "researcher"
+    start_offset_ms: float  # relative to trace start
+    duration_ms: float
+    status: SpanStatus
+    error: str | None  # exception message when status == "error"
+    input_preview: str
+    model: str | None
+    input_tokens: int
+    output_tokens: int
+    litellm_request_id: str | None
+
+
+class AgentNode(TypedDict):
+    """One distinct agent in a trace. 200 invocations of `researcher` = one node."""
+
+    name: str
+    parent_agent: str | None
+    invocations: int
+    llm_calls: int
+    tool_calls: int
+    duration_ms: float
+
+
+class TraceSummary(TypedDict):
+    trace_id: str
+    name: str
+    service: str
+    input_preview: str
+    start_time: str  # ISO 8601
+    duration_ms: float
+    status: SpanStatus
+    span_count: int
+    agent_count: int  # distinct agent names (researcher x200 counts once)
+    agent_invocations: int  # agent spans (researcher x200 counts 200)
+    llm_calls: int
+    tool_calls: int
+    error_count: int  # spans with an error status; > 0 means the run shows as failed
+    input_tokens: int
+    output_tokens: int
+    models: list[str]
+
+
+class Trace(TypedDict):
+    summary: TraceSummary
+    agents: list[AgentNode]
+    spans: list[Span]
+
+
+class TracePage(TypedDict):
+    data: list[TraceSummary]
+    next_cursor: str | None
+
+
+class SpanDetail(TypedDict):
+    span_id: str
+    input: str
+    output: str
+    attributes: dict[str, str]
+
+
+class TraceScope(TypedDict):
+    """Who is asking. Empty team_ids = all teams (admins only)."""
+
+    team_ids: list[str]
+    api_key_hash: str
+
+
+class SpanRow(TypedDict):
+    """One stored span (ClickHouse `otel_traces` row). Produced by `litellm.tracing.decode`."""
+
+    Timestamp: int  # unix ns
+    TraceId: str
+    SpanId: str
+    ParentSpanId: str
+    TraceState: str
+    SpanName: str
+    SpanKind: str
+    ServiceName: str
+    ResourceAttributes: dict[str, str]
+    ScopeName: str
+    ScopeVersion: str
+    SpanAttributes: dict[str, str]
+    Duration: int  # ns
+    StatusCode: str
+    StatusMessage: str
+    TeamId: str
+    ApiKeyHash: str
+    ObservationType: SpanType
+    AgentName: str
+    LiteLLMRequestId: str
+    Model: str
+    InputTokens: int
+    OutputTokens: int
+    Input: str
+    Output: str

tests/test_litellm/integrations/clickhouse/test_clickhouse_batch_logger.py

@@ -0,0 +1,71 @@
+"""
+Tests for the CustomBatchLogger-based ClickHouse base logger.
+"""
+
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock, patch
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+import pytest
+
+from litellm.integrations.clickhouse import clickhouse_batch_logger as module
+from litellm.integrations.clickhouse.clickhouse_batch_logger import ClickHouseBatchLogger
+from litellm.integrations.custom_batch_logger import CustomBatchLogger
+
+
+class _TestLogger(ClickHouseBatchLogger):
+    table = "test_table"
+
+
+def _logger(insert: AsyncMock) -> _TestLogger:
+    storage = MagicMock()
+    storage.insert_rows = insert
+    return _TestLogger(storage=storage)
+
+
+def test_is_a_custom_batch_logger():
+    assert issubclass(ClickHouseBatchLogger, CustomBatchLogger)
+
+
+@pytest.mark.asyncio
+async def test_flush_splits_into_batches_and_empties_queue():
+    insert = AsyncMock()
+    logger = _logger(insert)
+    logger.batch_size = 2
+    logger.log_queue.extend([{"i": i} for i in range(5)])
+
+    await logger.flush_queue()
+
+    assert [len(c.args[1]) for c in insert.await_args_list] == [2, 2, 1]
+    assert all(c.args[0] == "test_table" for c in insert.await_args_list)
+    assert logger.log_queue == []
+    assert logger.rows_written == 5
+
+
+@pytest.mark.asyncio
+async def test_is_full_signals_backpressure():
+    logger = _logger(AsyncMock())
+    with patch.object(module, "CLICKHOUSE_MAX_BUFFERED_ROWS", 3):
+        logger.log_queue.extend([{}, {}])
+        assert logger.is_full() is False
+        logger.log_queue.append({})
+        assert logger.is_full() is True
+
+
+@pytest.mark.asyncio
+async def test_failed_insert_is_requeued_then_dropped():
+    insert = AsyncMock(side_effect=RuntimeError("clickhouse down"))
+    logger = _logger(insert)
+    logger.log_queue.extend([{"request_id": "a"}, {"request_id": "b"}])
+
+    with patch.object(module, "CLICKHOUSE_MAX_RETRIES", 2):
+        await logger.flush_queue()
+        assert len(logger.log_queue) == 2  # kept for retry
+        await logger.flush_queue()
+
+    assert insert.await_count == 2
+    assert logger.rows_dropped == 2
+    assert logger.rows_written == 0
+    assert logger.log_queue == []

tests/test_litellm/proxy/common_utils/test_http_parsing_utils.py

@@ -71,6 +71,18 @@ async def test_read_raw_json_body_is_none_for_form_bodies():
     assert await read_raw_json_body(request) is None
 
 
+@pytest.mark.asyncio
+@pytest.mark.parametrize("content_type", ["application/x-protobuf", "application/protobuf; charset=binary"])
+async def test_protobuf_body_is_not_parsed_as_json(content_type):
+    # OTLP trace exports (POST /v1/traces) are binary protobuf; arbitrary bytes like these
+    # used to hit the JSON surrogate-repair path and fail auth with a 400.
+    body = b"\n\xa2\x01\n\x1c\n\x0cservice.name\x12\x0c\n\nswarm\xed\xa0\x80\xff"
+    request = _starlette_request(body, content_type)
+
+    assert await _read_request_body(request) == {}
+    assert await request.body() == body  # body is still readable by the endpoint
+
+
 @pytest.mark.asyncio
 async def test_read_raw_json_body_is_none_for_a_request_that_only_mocks_the_parsed_body_path():
     mock_request = MagicMock()

tests/test_litellm/proxy/test_tracing_endpoints.py

@@ -0,0 +1,169 @@
+"""
+Tests for the agent tracing endpoints (litellm/proxy/tracing_endpoints.py).
+"""
+
+import os
+import sys
+from unittest.mock import AsyncMock, MagicMock
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+import pytest
+from fastapi import FastAPI, HTTPException
+from fastapi.testclient import TestClient
+
+from litellm.proxy import tracing_endpoints
+from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+from litellm.tracing import TracingPayloadTooLargeError
+
+TEAM_KEY = UserAPIKeyAuth(
+    token="hashed-key", team_id="team-research", org_id="org-1", user_role=LitellmUserRoles.INTERNAL_USER
+)
+
+
+# ---------------------------------------------------------------- scope / tenant
+
+
+def test_scope_for_admin_sees_everything():
+    for role in (LitellmUserRoles.PROXY_ADMIN, LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY):
+        auth = UserAPIKeyAuth(token="k", team_id="team-a", user_role=role)
+        assert tracing_endpoints.scope_for(auth) == {"team_ids": [], "api_key_hash": ""}
+
+
+def test_scope_for_team_key_sees_its_team():
+    assert tracing_endpoints.scope_for(TEAM_KEY) == {"team_ids": ["team-research"], "api_key_hash": ""}
+
+
+def test_scope_for_teamless_key_sees_only_its_own_traces():
+    auth = UserAPIKeyAuth(token="hashed-key", user_role=LitellmUserRoles.INTERNAL_USER)
+    assert tracing_endpoints.scope_for(auth) == {"team_ids": [""], "api_key_hash": "hashed-key"}
+
+
+def test_scope_for_no_team_no_token_is_forbidden():
+    with pytest.raises(HTTPException) as e:
+        tracing_endpoints.scope_for(UserAPIKeyAuth(user_role=LitellmUserRoles.INTERNAL_USER))
+    assert e.value.status_code == 403
+
+
+def test_tenant_for_comes_from_auth():
+    tenant = tracing_endpoints.tenant_for(TEAM_KEY)
+    assert (tenant.team_id, tenant.api_key_hash, tenant.org_id) == ("team-research", "hashed-key", "org-1")
+    blank = tracing_endpoints.tenant_for(UserAPIKeyAuth())
+    assert (blank.team_id, blank.api_key_hash, blank.org_id) == ("", "", "")
+
+
+# ---------------------------------------------------------------- endpoints
+
+
+@pytest.fixture
+def receiver(monkeypatch) -> MagicMock:
+    fake = MagicMock()
+    fake.ingest = AsyncMock(return_value=1)
+    fake.list_traces = AsyncMock(return_value={"data": [], "next_cursor": None})
+    fake.get_trace = AsyncMock(return_value=None)
+    fake.get_span = AsyncMock(return_value=None)
+    monkeypatch.setattr(tracing_endpoints, "receiver", fake)
+    return fake
+
+
+@pytest.fixture
+def client() -> TestClient:
+    app = FastAPI()
+    app.include_router(tracing_endpoints.router)
+    app.dependency_overrides[user_api_key_auth] = lambda: TEAM_KEY
+    return TestClient(app)
+
+
+def test_501_when_tracing_not_enabled(client, monkeypatch):
+    monkeypatch.setattr(tracing_endpoints, "receiver", None)
+    assert client.post("/v1/traces", content=b"").status_code == 501
+    assert client.get("/v1/traces").status_code == 501
+
+
+def test_post_protobuf_returns_empty_protobuf(client, receiver):
+    response = client.post(
+        "/v1/traces",
+        content=b"\x0a\x00",
+        headers={"content-type": "application/x-protobuf", "content-encoding": "gzip"},
+    )
+    assert response.status_code == 200
+    assert response.content == b""
+    assert response.headers["content-type"] == "application/x-protobuf"
+    kwargs = receiver.ingest.call_args.kwargs
+    assert kwargs["body"] == b"\x0a\x00"
+    assert kwargs["content_type"] == "application/x-protobuf"
+    assert kwargs["content_encoding"] == "gzip"
+    assert kwargs["tenant"].team_id == "team-research"
+
+
+def test_post_json_returns_empty_json(client, receiver):
+    response = client.post("/v1/traces", content=b"{}", headers={"content-type": "application/json"})
+    assert response.status_code == 200
+    assert response.json() == {}
+
+
+def test_post_clickhouse_failure_is_503_with_retry_after(client, receiver):
+    receiver.ingest.side_effect = RuntimeError("ClickHouse unavailable")
+    response = client.post("/v1/traces", content=b"", headers={"content-type": "application/x-protobuf"})
+    assert response.status_code == 503
+    assert response.headers["retry-after"] == str(tracing_endpoints.OTLP_RETRY_AFTER_SECONDS)
+
+
+def test_post_too_large_is_413(client, receiver):
+    receiver.ingest.side_effect = TracingPayloadTooLargeError("OTLP body exceeds 10 bytes")
+    response = client.post("/v1/traces", content=b"x" * 20)
+    assert response.status_code == 413
+    assert "exceeds" in response.json()["detail"]
+
+
+def test_list_traces_passes_scope_window_and_cursor(client, receiver):
+    response = client.get("/v1/traces", params={"start_ms": 1, "end_ms": 2, "cursor": "abc"})
+    assert response.status_code == 200
+    assert response.json() == {"data": [], "next_cursor": None}
+    receiver.list_traces.assert_awaited_once_with(
+        scope={"team_ids": ["team-research"], "api_key_hash": ""}, start_ms=1, end_ms=2, cursor="abc"
+    )
+
+
+def test_list_traces_defaults_to_last_24h(client, receiver):
+    client.get("/v1/traces")
+    kwargs = receiver.list_traces.call_args.kwargs
+    assert kwargs["end_ms"] - kwargs["start_ms"] == tracing_endpoints.MS_PER_DAY
+    assert kwargs["cursor"] is None
+
+
+def test_get_trace_404_and_200(client, receiver):
+    assert client.get("/v1/traces/missing").status_code == 404
+    trace = {"summary": {"trace_id": "t1"}, "agents": [], "spans": []}
+    receiver.get_trace.return_value = trace
+    response = client.get("/v1/traces/t1")
+    assert response.status_code == 200
+    assert response.json() == trace
+    receiver.get_trace.assert_awaited_with("t1", {"team_ids": ["team-research"], "api_key_hash": ""})
+
+
+def test_get_span_404_and_200(client, receiver):
+    assert client.get("/v1/traces/t1/spans/s1").status_code == 404
+    receiver.get_span.return_value = {"span_id": "s1", "input": "", "output": "", "attributes": {}}
+    response = client.get("/v1/traces/t1/spans/s1")
+    assert response.status_code == 200
+    assert response.json()["span_id"] == "s1"
+    receiver.get_span.assert_awaited_with("t1", "s1", {"team_ids": ["team-research"], "api_key_hash": ""})
+
+
+def test_teamless_key_without_token_gets_403_on_reads(client, receiver):
+    client.app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
+        user_role=LitellmUserRoles.INTERNAL_USER
+    )
+    assert client.get("/v1/traces").status_code == 403
+    receiver.list_traces.assert_not_called()
+
+
+def test_view_only_admin_cannot_ingest_traces(client, receiver):
+    client.app.dependency_overrides[user_api_key_auth] = lambda: UserAPIKeyAuth(
+        token="admin-key", user_role=LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY
+    )
+    response = client.post("/v1/traces", content=b"{}")
+    assert response.status_code == 403
+    receiver.ingest.assert_not_called()

tests/test_litellm/tracing/test_decode.py

@@ -0,0 +1,301 @@
+"""
+Tests for OTLP decode + normalization (litellm/tracing/decode.py).
+
+The fixture is a trimmed real export from a Deep Agents run (LangSmith OTEL mode):
+deep_research_agent -> task (tool) -> researcher (subagent) -> search_docs (tool).
+"""
+
+import gzip
+import json
+import os
+import sys
+from pathlib import Path
+from unittest.mock import patch
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+import pytest
+from google.protobuf.json_format import Parse
+from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
+from opentelemetry.proto.common.v1.common_pb2 import AnyValue, KeyValue
+from opentelemetry.proto.trace.v1.trace_pb2 import ResourceSpans, ScopeSpans, Span, Status
+
+from litellm.tracing import decode
+from litellm.tracing.decode import decode_otlp, encode_otlp_response
+
+pytestmark = pytest.mark.requires_rust_extension
+
+FIXTURE = Path(__file__).parent / "fixtures" / "langsmith_deep_agent_export.json"
+TRACE_ID = "4bad42b84e9de3ba46fc870185f8f023"
+
+
+def _fixture_json() -> bytes:
+    return FIXTURE.read_bytes()
+
+
+def _fixture_protobuf() -> bytes:
+    request = ExportTraceServiceRequest()
+    Parse(_fixture_json().decode(), request)
+    return request.SerializeToString()
+
+
+@pytest.fixture
+def rows_by_name() -> dict:
+    rows = decode_otlp(_fixture_json(), "application/json")
+    return {r["SpanName"]: r for r in rows}
+
+
+def _kv(key: str, value: str | int) -> KeyValue:
+    if isinstance(value, int):
+        return KeyValue(key=key, value=AnyValue(int_value=value))
+    return KeyValue(key=key, value=AnyValue(string_value=value))
+
+
+def _export(*spans: Span, service: str = "svc", scope: str = "test") -> bytes:
+    resource_spans = ResourceSpans(scope_spans=[ScopeSpans(spans=list(spans))])
+    resource_spans.resource.attributes.append(_kv("service.name", service))
+    resource_spans.scope_spans[0].scope.name = scope
+    return ExportTraceServiceRequest(resource_spans=[resource_spans]).SerializeToString()
+
+
+def _span(name: str, span_id: bytes, parent: bytes = b"", **attributes: str | int) -> Span:
+    return Span(
+        trace_id=bytes.fromhex(TRACE_ID),
+        span_id=span_id,
+        parent_span_id=parent,
+        name=name,
+        start_time_unix_nano=1_000,
+        end_time_unix_nano=5_000,
+        attributes=[_kv(k.replace("__", "."), v) for k, v in attributes.items()],
+    )
+
+
+# ---------------------------------------------------------------- LangSmith / Deep Agents fixture
+
+
+def test_classifies_every_langsmith_span(rows_by_name):
+    assert {name: r["ObservationType"] for name, r in rows_by_name.items()} == {
+        "deep_research_agent": "agent",
+        "ChatOpenAI": "llm",
+        "FilesystemMiddleware.wrap_model_call": "framework",
+        "task": "tool",
+        "researcher": "agent",
+        "search_docs": "tool",
+    }
+
+
+def test_agent_name_is_the_enclosing_agent(rows_by_name):
+    assert rows_by_name["task"]["AgentName"] == "deep_research_agent"
+    assert rows_by_name["ChatOpenAI"]["AgentName"] == "deep_research_agent"
+    assert rows_by_name["researcher"]["AgentName"] == "researcher"
+    assert rows_by_name["search_docs"]["AgentName"] == "researcher"
+
+
+def test_subagent_is_nested_under_task_tool(rows_by_name):
+    assert rows_by_name["researcher"]["ParentSpanId"] == rows_by_name["task"]["SpanId"]
+    assert rows_by_name["deep_research_agent"]["ParentSpanId"] == ""
+
+
+def test_llm_span_carries_litellm_request_id_model_and_tokens(rows_by_name):
+    llm = rows_by_name["ChatOpenAI"]
+    assert llm["LiteLLMRequestId"] == "chatcmpl-4077bb36-9380-4a3b-9481-245700cef09a"
+    assert llm["Model"] == "claude-sonnet-4-5"
+    assert (llm["InputTokens"], llm["OutputTokens"]) == (3332, 467)
+
+
+def test_llm_input_output_are_normalized_messages(rows_by_name):
+    llm = rows_by_name["ChatOpenAI"]
+    messages = json.loads(llm["Input"])
+    assert [m["role"] for m in messages][:2] == ["system", "user"]
+    assert "research lead" in messages[0]["content"]
+    output = json.loads(llm["Output"])
+    assert output["role"] == "assistant"
+    assert output["tool_calls"][0]["name"]
+
+
+def test_task_tool_output_is_subagent_final_message_text(rows_by_name):
+    task = rows_by_name["task"]
+    assert json.loads(task["Input"])["subagent_type"] == "researcher"
+    assert task["Output"].startswith("Based on my research")
+    assert not task["Output"].startswith("{")
+
+
+def test_agent_input_output(rows_by_name):
+    root = rows_by_name["deep_research_agent"]
+    assert json.loads(root["Input"]) == [
+        {"role": "user", "content": "Should we store OTEL agent spans in ClickHouse or Postgres at 50k spans/sec?"}
+    ]
+    assert json.loads(root["Output"])["role"] == "assistant"
+
+
+def test_plain_tool_input_output(rows_by_name):
+    tool = rows_by_name["search_docs"]
+    assert json.loads(tool["Input"]) == {"query": "ClickHouse Postgres OpenTelemetry OTEL spans performance comparison"}
+    assert tool["Output"].startswith("ClickHouse ingests")
+
+
+def test_heavy_attributes_are_lifted_out_of_span_attributes(rows_by_name):
+    for row in rows_by_name.values():
+        assert not set(row["SpanAttributes"]) & decode._HEAVY_ATTRIBUTES
+    assert rows_by_name["ChatOpenAI"]["SpanAttributes"]["langsmith.span.kind"] == "llm"
+
+
+def test_ids_are_hex_and_resource_is_kept(rows_by_name):
+    root = rows_by_name["deep_research_agent"]
+    assert root["TraceId"] == TRACE_ID
+    assert root["SpanId"] == "5e79f3b5b504985e"
+    assert root["ServiceName"] == "agent-demo"
+    assert root["ScopeName"] == "langsmith"
+    assert root["SpanKind"] == "SPAN_KIND_INTERNAL"
+    assert root["StatusCode"] == "STATUS_CODE_OK"
+    assert root["Duration"] > 0
+
+
+def test_protobuf_and_json_decode_identically():
+    from_json = decode_otlp(_fixture_json(), "application/json")
+    from_protobuf = decode_otlp(_fixture_protobuf(), "application/x-protobuf")
+    assert from_json == from_protobuf
+    assert len(from_json) == 6
+
+
+def test_content_type_defaults_to_protobuf():
+    assert len(decode_otlp(_fixture_protobuf(), None)) == 6
+
+
+@pytest.mark.parametrize("content_encoding", ["gzip", None])
+def test_gzip_body_by_header_or_magic_bytes(content_encoding):
+    rows = decode_otlp(gzip.compress(_fixture_protobuf()), "application/x-protobuf", content_encoding)
+    assert len(rows) == 6
+
+
+def test_long_values_are_truncated_with_marker():
+    with patch.object(decode, "OTLP_MAX_ATTRIBUTE_VALUE_BYTES", 100):
+        rows = {r["SpanName"]: r for r in decode_otlp(_fixture_json(), "application/json")}
+    task = rows["task"]
+    assert "…[truncated " in task["Input"]
+    assert task["Input"].encode().startswith(task["Input"].split("…")[0].encode())
+    assert len(task["Input"].split("…")[0].encode()) <= 100
+
+
+# ---------------------------------------------------------------- status / exceptions
+
+
+def test_exception_event_fills_status_message():
+    span = _span("get_customer_plan", b"\x01" * 8, b"\x02" * 8)
+    span.status.CopyFrom(Status(code=Status.STATUS_CODE_ERROR))
+    event = span.events.add()
+    event.name = "exception"
+    event.attributes.extend(
+        [_kv("exception.type", "KeyError"), _kv("exception.message", "customer acme-404 not found")]
+    )
+    (row,) = decode_otlp(_export(span))
+    assert row["StatusCode"] == "STATUS_CODE_ERROR"
+    assert row["StatusMessage"] == "customer acme-404 not found"
+
+
+def test_status_message_wins_over_exception_event():
+    span = _span("tool", b"\x01" * 8, b"\x02" * 8)
+    span.status.CopyFrom(Status(code=Status.STATUS_CODE_ERROR, message="boom"))
+    event = span.events.add()
+    event.name = "exception"
+    event.attributes.append(_kv("exception.message", "other"))
+    (row,) = decode_otlp(_export(span))
+    assert row["StatusMessage"] == "boom"
+
+
+# ---------------------------------------------------------------- GenAI semconv / OpenInference
+
+
+def test_genai_semconv_spans():
+    root = _span(
+        "invoke_agent planner", b"\x01" * 8, gen_ai__operation__name="invoke_agent", gen_ai__agent__name="planner"
+    )
+    chat = _span(
+        "chat gpt-4o",
+        b"\x02" * 8,
+        b"\x01" * 8,
+        gen_ai__operation__name="chat",
+        gen_ai__agent__name="planner",
+        gen_ai__request__model="gpt-4o",
+        gen_ai__response__id="chatcmpl-abc",
+        gen_ai__usage__input_tokens=12,
+        gen_ai__usage__output_tokens=3,
+        gen_ai__input__messages='[{"role":"user","content":"hi"}]',
+        gen_ai__output__messages='[{"role":"assistant","content":"hello"}]',
+    )
+    tool = _span(
+        "execute_tool search",
+        b"\x03" * 8,
+        b"\x01" * 8,
+        gen_ai__operation__name="execute_tool",
+        gen_ai__tool__call__arguments='{"q":"x"}',
+        gen_ai__tool__call__result="found",
+    )
+    rows = {r["SpanName"]: r for r in decode_otlp(_export(root, chat, tool))}
+    assert rows["invoke_agent planner"]["ObservationType"] == "agent"
+    assert rows["invoke_agent planner"]["AgentName"] == "planner"
+    llm = rows["chat gpt-4o"]
+    assert (llm["ObservationType"], llm["Model"], llm["LiteLLMRequestId"]) == ("llm", "gpt-4o", "chatcmpl-abc")
+    assert (llm["InputTokens"], llm["OutputTokens"]) == (12, 3)
+    assert json.loads(llm["Input"])[0]["content"] == "hi"
+    assert "gen_ai.input.messages" not in llm["SpanAttributes"]
+    assert (rows["execute_tool search"]["ObservationType"], rows["execute_tool search"]["Output"]) == ("tool", "found")
+
+
+def test_openinference_spans():
+    root = _span("agent", b"\x01" * 8, openinference__span__kind="AGENT", agent__name="writer", input__value="task")
+    llm = _span(
+        "llm",
+        b"\x02" * 8,
+        b"\x01" * 8,
+        openinference__span__kind="LLM",
+        llm__model_name="claude-sonnet-4-5",
+        llm__token_count__prompt=40,
+        llm__token_count__completion=8,
+        input__value="prompt",
+        output__value="answer",
+    )
+    chain = _span("retriever", b"\x03" * 8, b"\x01" * 8, openinference__span__kind="RETRIEVER")
+    rows = {r["SpanName"]: r for r in decode_otlp(_export(root, llm, chain))}
+    assert (rows["agent"]["ObservationType"], rows["agent"]["AgentName"], rows["agent"]["Input"]) == (
+        "agent",
+        "writer",
+        "task",
+    )
+    assert rows["llm"]["ObservationType"] == "llm"
+    assert (rows["llm"]["Model"], rows["llm"]["InputTokens"], rows["llm"]["OutputTokens"]) == (
+        "claude-sonnet-4-5",
+        40,
+        8,
+    )
+    assert (rows["llm"]["Input"], rows["llm"]["Output"]) == ("prompt", "answer")
+    assert "input.value" not in rows["llm"]["SpanAttributes"]
+    assert rows["retriever"]["ObservationType"] == "chain"
+
+
+def test_non_string_attribute_values_are_stringified():
+    span = _span("root", b"\x01" * 8)
+    span.attributes.extend(
+        [
+            KeyValue(key="flag", value=AnyValue(bool_value=True)),
+            KeyValue(key="ratio", value=AnyValue(double_value=0.5)),
+            KeyValue(key="raw", value=AnyValue(bytes_value=b"abc")),
+        ]
+    )
+    array = KeyValue(key="list")
+    array.value.array_value.values.extend([AnyValue(string_value="a"), AnyValue(int_value=1)])
+    span.attributes.append(array)
+    (row,) = decode_otlp(_export(span))
+    assert row["SpanAttributes"]["flag"] == "true"
+    assert row["SpanAttributes"]["ratio"] == "0.5"
+    assert row["SpanAttributes"]["raw"] == "abc"
+    assert json.loads(row["SpanAttributes"]["list"]) == ["a", "1"]
+
+
+# ---------------------------------------------------------------- helpers
+
+
+def test_encode_otlp_response_matches_request_encoding():
+    assert encode_otlp_response("application/json") == (b"{}", "application/json")
+    assert encode_otlp_response("application/x-protobuf") == (b"", "application/x-protobuf")
+    assert encode_otlp_response(None) == (b"", "application/x-protobuf")

tests/test_litellm/tracing/test_receiver.py

@@ -0,0 +1,121 @@
+"""
+Tests for TraceReceiver.ingest (litellm/tracing/receiver.py) with a fake store.
+"""
+
+import os
+import sys
+from pathlib import Path
+from unittest.mock import AsyncMock, MagicMock, patch
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+import pytest
+from opentelemetry.proto.collector.trace.v1.trace_service_pb2 import ExportTraceServiceRequest
+from opentelemetry.proto.common.v1.common_pb2 import AnyValue, KeyValue
+from opentelemetry.proto.trace.v1.trace_pb2 import ResourceSpans, ScopeSpans, Span
+
+from litellm.tracing import Tenant, TraceReceiver, TracingPayloadTooLargeError
+from litellm.tracing import receiver as receiver_module
+from litellm.tracing.types import TraceScope
+
+pytestmark = pytest.mark.requires_rust_extension
+
+FIXTURE = Path(__file__).parent / "fixtures" / "langsmith_deep_agent_export.json"
+TENANT = Tenant(team_id="team-research", api_key_hash="hashed-key", org_id="org-1")
+
+
+def _fake_store() -> MagicMock:
+    store = MagicMock()
+    store.insert_spans = AsyncMock()
+    store.get_trace = AsyncMock(return_value=None)
+    return store
+
+
+def _spoofed_export() -> bytes:
+    """A client that tries to claim another team via resource attributes."""
+    resource_spans = ResourceSpans(scope_spans=[ScopeSpans(spans=[Span(trace_id=b"\x01" * 16, span_id=b"\x02" * 8)])])
+    resource_spans.resource.attributes.extend(
+        [
+            KeyValue(key="service.name", value=AnyValue(string_value="svc")),
+            KeyValue(key="litellm.team_id", value=AnyValue(string_value="someone-elses-team")),
+            KeyValue(key="litellm.api_key_hash", value=AnyValue(string_value="someone-elses-key")),
+        ]
+    )
+    return ExportTraceServiceRequest(resource_spans=[resource_spans]).SerializeToString()
+
+
+@pytest.mark.asyncio
+async def test_ingest_returns_span_count_and_writes_stamped_rows():
+    store = _fake_store()
+    count = await TraceReceiver(store).ingest(FIXTURE.read_bytes(), "application/json", None, TENANT)
+    assert count == 6
+    (rows,) = store.insert_spans.await_args.args
+    assert len(rows) == 6
+    for row in rows:
+        assert (row["TeamId"], row["ApiKeyHash"]) == ("team-research", "hashed-key")
+        assert row["ResourceAttributes"]["litellm.org_id"] == "org-1"
+        assert row["ResourceAttributes"]["service.name"] == "agent-demo"
+
+
+@pytest.mark.asyncio
+async def test_ingest_overwrites_client_supplied_tenant_attributes():
+    store = _fake_store()
+    await TraceReceiver(store).ingest(_spoofed_export(), "application/x-protobuf", None, TENANT)
+    ((row,),) = store.insert_spans.await_args.args
+    assert row["TeamId"] == "team-research"
+    assert row["ResourceAttributes"]["litellm.team_id"] == "team-research"
+    assert row["ResourceAttributes"]["litellm.api_key_hash"] == "hashed-key"
+
+
+@pytest.mark.asyncio
+async def test_ingest_does_not_acknowledge_failed_clickhouse_write():
+    store = _fake_store()
+    store.insert_spans.side_effect = RuntimeError("ClickHouse unavailable")
+    with pytest.raises(RuntimeError, match="ClickHouse unavailable"):
+        await TraceReceiver(store).ingest(FIXTURE.read_bytes(), "application/json", None, TENANT)
+    store.insert_spans.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_ingest_rejects_oversized_encoded_batch():
+    store = _fake_store()
+    store.insert_spans.side_effect = OverflowError("ClickHouse insert exceeds the encoded size limit")
+    with pytest.raises(TracingPayloadTooLargeError, match="encoded size limit"):
+        await TraceReceiver(store).ingest(FIXTURE.read_bytes(), "application/json", None, TENANT)
+
+
+@pytest.mark.asyncio
+async def test_ingest_rejects_oversized_body():
+    store = _fake_store()
+    with patch.object(receiver_module, "OTLP_MAX_BODY_BYTES", 10):
+        with pytest.raises(TracingPayloadTooLargeError):
+            await TraceReceiver(store).ingest(FIXTURE.read_bytes(), "application/json", None, TENANT)
+    store.insert_spans.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_large_body_is_decoded_off_the_event_loop():
+    store = _fake_store()
+    with (
+        patch.object(receiver_module, "OTLP_OFFLOAD_DECODE_BYTES", 0),
+        patch.object(receiver_module.asyncio, "to_thread", wraps=receiver_module.asyncio.to_thread) as to_thread,
+    ):
+        count = await TraceReceiver(store).ingest(FIXTURE.read_bytes(), "application/json", None, TENANT)
+    assert count == 6
+    to_thread.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_empty_export_writes_nothing():
+    store = _fake_store()
+    assert await TraceReceiver(store).ingest(b"", "application/x-protobuf", None, TENANT) == 0
+    store.insert_spans.assert_awaited_once_with([])
+
+
+@pytest.mark.asyncio
+async def test_reads_delegate_to_store():
+    store = _fake_store()
+    tracing = TraceReceiver(store)
+    scope: TraceScope = {"team_ids": ["team-research"], "api_key_hash": ""}
+    assert await tracing.get_trace("t1", scope) is None
+    store.get_trace.assert_awaited_once_with("t1", scope)

tests/test_litellm/tracing/test_store.py

@@ -0,0 +1,298 @@
+"""
+Tests for the pure read-side helpers in litellm/tracing/store.py (no ClickHouse needed).
+"""
+
+import os
+import sys
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+sys.path.insert(0, os.path.abspath("../../.."))
+
+import pytest
+
+from litellm.tracing.types import TraceScope
+from litellm.tracing.store import (
+    ClickHouseTraceStore,
+    agent_nodes,
+    decode_cursor,
+    encode_cursor,
+    span_from_row,
+    trace_from_rows,
+    trace_summary_from_row,
+)
+
+T0 = 1_790_742_989_000_000_000  # ns
+MS = 1_000_000
+
+
+def _row(
+    span_id: str,
+    parent: str,
+    name: str,
+    type_: str,
+    agent: str,
+    start_ms: float = 0,
+    duration_ms: float = 10,
+    status: str = "STATUS_CODE_OK",
+    **extra: Any,
+) -> dict[str, Any]:
+    return {
+        "span_id": span_id,
+        "parent_span_id": parent,
+        "name": name,
+        "type": type_,
+        "agent": agent,
+        "status": status,
+        "start_ns": T0 + int(start_ms * MS),
+        "duration_ns": int(duration_ms * MS),
+        "service": "agent-demo",
+        "input_preview": f"input of {name}",
+        "model": "",
+        "input_tokens": 0,
+        "output_tokens": 0,
+        "litellm_request_id": "",
+        **extra,
+    }
+
+
+def _llm_row(span_id: str, parent: str, agent: str, request_id: str, start_ms: float = 1) -> dict:
+    return _row(
+        span_id,
+        parent,
+        "ChatOpenAI",
+        "llm",
+        agent,
+        start_ms=start_ms,
+        duration_ms=100,
+        model="claude-sonnet-4-5",
+        input_tokens=100,
+        output_tokens=20,
+        litellm_request_id=request_id,
+    )
+
+
+def _deep_agent_rows(researcher_invocations: int = 1) -> list[dict[str, Any]]:
+    """root agent -> llm, task tool -> researcher subagent (N times) -> llm + search_docs tool."""
+    rows = [
+        _row("root", "", "deep_research_agent", "agent", "deep_research_agent", duration_ms=1000),
+        _llm_row("llm-root", "root", "deep_research_agent", "chatcmpl-root"),
+        _row("task", "root", "task", "tool", "deep_research_agent", start_ms=200, duration_ms=700),
+    ]
+    for i in range(researcher_invocations):
+        rows += [
+            _row(f"res-{i}", "task", "researcher", "agent", "researcher", start_ms=201, duration_ms=5),
+            _llm_row(f"res-llm-{i}", f"res-{i}", "researcher", f"chatcmpl-res-{i}", start_ms=202),
+            _row(f"res-tool-{i}", f"res-{i}", "search_docs", "tool", "researcher", start_ms=203, duration_ms=1),
+            _row(f"res-mw-{i}", f"res-{i}", "FilesystemMiddleware.wrap_model_call", "framework", "researcher"),
+        ]
+    return rows
+
+
+# ---------------------------------------------------------------- trace_from_rows
+
+
+def test_empty_rows_is_none():
+    assert trace_from_rows("abc", []) is None
+
+
+def test_llm_response_id_is_preserved_without_spend_enrichment():
+    trace = trace_from_rows("t1", _deep_agent_rows())
+    assert trace is not None
+    spans = {span["span_id"]: span for span in trace["spans"]}
+    assert spans["llm-root"]["litellm_request_id"] == "chatcmpl-root"
+    assert spans["task"]["litellm_request_id"] is None
+    assert "spend" not in trace["summary"]
+
+
+def test_summary_totals():
+    trace = trace_from_rows("t1", _deep_agent_rows())
+    assert trace is not None
+    summary = trace["summary"]
+    assert summary["trace_id"] == "t1"
+    assert summary["name"] == "deep_research_agent"
+    assert summary["service"] == "agent-demo"
+    assert summary["input_preview"] == "input of deep_research_agent"
+    assert summary["status"] == "ok"
+    assert summary["span_count"] == 7
+    assert summary["agent_count"] == 2
+    assert summary["llm_calls"] == 2
+    assert summary["tool_calls"] == 2
+    assert summary["error_count"] == 0
+    assert (summary["input_tokens"], summary["output_tokens"]) == (200, 40)
+    assert summary["models"] == ["claude-sonnet-4-5"]
+    assert summary["duration_ms"] == 1000
+    assert summary["start_time"].startswith("2026-09-30T")
+
+
+def test_error_count_counts_error_spans():
+    rows = _deep_agent_rows()
+    rows[2]["status"] = "STATUS_CODE_ERROR"
+    trace = trace_from_rows("t1", rows)
+    assert trace is not None
+    assert trace["summary"]["error_count"] == 1
+    assert trace["summary"]["status"] == "ok"  # root span status; the UI uses error_count for "failed"
+    assert trace["spans"][2]["status"] == "error"
+
+
+def test_offsets_are_relative_to_trace_start_in_ms():
+    trace = trace_from_rows("t1", _deep_agent_rows())
+    assert trace is not None
+    spans = {s["span_id"]: s for s in trace["spans"]}
+    assert spans["root"]["start_offset_ms"] == 0
+    assert spans["task"]["start_offset_ms"] == 200
+    assert spans["task"]["duration_ms"] == 700
+    assert spans["root"]["parent_span_id"] is None
+    assert spans["task"]["parent_span_id"] == "root"
+
+
+def test_span_from_row_optional_fields():
+    span = span_from_row(_row("s", "", "x", "chain", "a", status="STATUS_CODE_UNSET"), T0)
+    assert (span["model"], span["parent_span_id"], span["status"], span["litellm_request_id"]) == (
+        None,
+        None,
+        "unset",
+        None,
+    )
+
+
+def test_agent_nodes_parent_and_per_agent_counts():
+    trace = trace_from_rows("t1", _deep_agent_rows())
+    assert trace is not None
+    assert trace["agents"] == [
+        {
+            "name": "deep_research_agent",
+            "parent_agent": None,
+            "invocations": 1,
+            "llm_calls": 1,
+            "tool_calls": 1,
+            "duration_ms": 1000,
+        },
+        {
+            "name": "researcher",
+            "parent_agent": "deep_research_agent",
+            "invocations": 1,
+            "llm_calls": 1,
+            "tool_calls": 1,
+            "duration_ms": 5,
+        },
+    ]
+
+
+def test_200_subagent_invocations_aggregate_into_one_node():
+    trace = trace_from_rows("t1", _deep_agent_rows(researcher_invocations=200))
+    assert trace is not None
+    assert [a["name"] for a in trace["agents"]] == ["deep_research_agent", "researcher"]
+    researcher = trace["agents"][1]
+    assert researcher["parent_agent"] == "deep_research_agent"
+    assert researcher["invocations"] == 200
+    assert researcher["llm_calls"] == 200
+    assert researcher["tool_calls"] == 200
+    assert researcher["duration_ms"] == pytest.approx(1000)
+    assert trace["summary"]["agent_count"] == 2
+    assert trace["summary"]["span_count"] == 3 + 4 * 200
+
+
+def test_parent_agent_skips_same_name_ancestors():
+    """A recursive agent (researcher -> researcher) still reports the nearest *different* agent."""
+    rows = [
+        _row("root", "", "lead", "agent", "lead"),
+        _row("r1", "root", "researcher", "agent", "researcher"),
+        _row("r2", "r1", "researcher", "agent", "researcher"),
+    ]
+    spans = [span_from_row(r, T0) for r in rows]
+    nodes = {n["name"]: n for n in agent_nodes(spans)}
+    assert nodes["researcher"]["parent_agent"] == "lead"
+    assert nodes["researcher"]["invocations"] == 2
+
+
+def test_agent_nodes_ignores_spans_of_unknown_agents():
+    spans = [span_from_row(_row("t", "", "tool", "tool", "ghost"), T0)]
+    assert agent_nodes(spans) == []
+
+
+# ---------------------------------------------------------------- list helpers
+
+
+def test_cursor_round_trip():
+    cursor = encode_cursor(1790742989377, "4bad42b84e9de3ba46fc870185f8f023")
+    assert decode_cursor(cursor) == (1790742989377, "4bad42b84e9de3ba46fc870185f8f023")
+    assert decode_cursor(None) == (0, "")
+    assert decode_cursor("") == (0, "")
+
+
+def test_trace_summary_from_row():
+    summary = trace_summary_from_row(
+        {
+            "trace_id": "t1",
+            "name": "deep_research_agent",
+            "service": "agent-demo",
+            "input_preview": "hi",
+            "start_ms": 1790742989377,
+            "duration_ms": 51385,
+            "status": "STATUS_CODE_OK",
+            "span_count": "126",
+            "agent_count": "2",
+            "llm_calls": "7",
+            "tool_calls": "26",
+            "error_count": "1",
+            "input_tokens": "30175",
+            "output_tokens": "2620",
+            "models": ["claude-sonnet-4-5"],
+        }
+    )
+    assert summary["status"] == "ok"
+    assert (summary["span_count"], summary["error_count"]) == (126, 1)
+    assert summary["start_time"] == "2026-09-30T04:36:29.377000+00:00"
+
+
+@pytest.mark.asyncio
+async def test_list_traces_sets_next_cursor_on_full_page():
+    client = MagicMock()
+    row = {
+        "trace_id": "t2",
+        "name": "a",
+        "service": "s",
+        "input_preview": "",
+        "start_ms": 1000,
+        "duration_ms": 1,
+        "status": "STATUS_CODE_OK",
+        "span_count": 1,
+        "agent_count": 1,
+        "llm_calls": 0,
+        "tool_calls": 0,
+        "error_count": 0,
+        "input_tokens": 0,
+        "output_tokens": 0,
+        "models": [],
+    }
+    client.query = AsyncMock(return_value=[row, {**row, "trace_id": "t1", "start_ms": 900}])
+    store = ClickHouseTraceStore(client)
+    scope: TraceScope = {"team_ids": ["team-a"], "api_key_hash": ""}
+
+    page = await store.list_traces(scope, 0, 2000, limit=2)
+    assert [t["trace_id"] for t in page["data"]] == ["t2", "t1"]
+    assert page["next_cursor"] is not None
+    assert decode_cursor(page["next_cursor"]) == (900, "t1")
+    params = client.query.call_args.args[1]
+    assert params["team_ids"] == ["team-a"] and params["limit"] == 2 and params["cursor_ms"] == 0
+
+    page = await store.list_traces(scope, 0, 2000, cursor=page["next_cursor"], limit=3)
+    assert page["next_cursor"] is None
+    assert client.query.call_args.args[1]["cursor_trace_id"] == "t1"
+
+
+@pytest.mark.asyncio
+async def test_get_span_not_found_and_found():
+    client = MagicMock()
+    client.query = AsyncMock(return_value=[])
+    store = ClickHouseTraceStore(client)
+    scope: TraceScope = {"team_ids": [], "api_key_hash": ""}
+    assert await store.get_span("t", "s", scope) is None
+    client.query = AsyncMock(return_value=[{"span_id": "s", "input": "i", "output": "o", "attributes": {"k": "v"}}])
+    assert await store.get_span("t", "s", scope) == {
+        "span_id": "s",
+        "input": "i",
+        "output": "o",
+        "attributes": {"k": "v"},
+    }

tests/test_litellm_rust/test_traces.py

@@ -1,11 +1,12 @@
 import base64
+import gzip
 import json
 from typing import Final
 from urllib.parse import parse_qs, urlsplit
 
 import pytest
 
-from litellm.rust_bridge.traces import encode_rows, ensure_schema, query
+from litellm.rust_bridge._native import NativeTraceStorage
 from tests.test_litellm_rust.support.recording_server import RecordingServer, ResponseSpec
 
 pytestmark = pytest.mark.requires_rust_extension
@@ -14,14 +15,9 @@ pytestmark = pytest.mark.requires_rust_extension
 @pytest.mark.asyncio
 async def test_trace_reader_projects_connection_and_parameters(recording_server: RecordingServer) -> None:
     recording_server.enqueue(ResponseSpec(body={"data": [{"trace_id": "trace-1"}]}))
-    rows: Final = await query(
-        recording_server.base_url + "?database=wrong&user=wrong&password=wrong",
-        "trace_test",
-        "reader",
-        "p@ss/word%",
-        "SELECT {trace_id:String} AS trace_id",
-        {"trace_id": "trace-1"},
-    )
+    reader_url: Final = recording_server.base_url.replace("http://", "http://reader:p%40ss%2Fword%25@")
+    storage: Final = NativeTraceStorage("trace_test", recording_server.base_url, reader_url + "?database=wrong")
+    rows: Final = json.loads(await storage.query("SELECT {trace_id:String} AS trace_id", {"trace_id": "trace-1"}))
     request: Final = recording_server.requests[0]
     parameters: Final = parse_qs(urlsplit(request.path).query)
     assert rows == [{"trace_id": "trace-1"}]
@@ -37,15 +33,22 @@ async def test_trace_reader_projects_connection_and_parameters(recording_server:
 @pytest.mark.asyncio
 async def test_trace_reader_rejects_success_status_with_embedded_error(recording_server: RecordingServer) -> None:
     recording_server.enqueue(ResponseSpec(body={"data": [], "exception": "query failed"}))
+    storage: Final = NativeTraceStorage("trace_test", recording_server.base_url, recording_server.base_url)
     with pytest.raises(RuntimeError, match="invalid or failed JSON"):
-        await query(recording_server.base_url, "trace_test", "reader", "password", "SELECT 1", {})
+        await storage.query("SELECT 1", {})
 
 
 @pytest.mark.asyncio
-@pytest.mark.parametrize("database,retention", [("db; DROP DATABASE default", 7), ("traces", 0)])
-async def test_schema_binding_preserves_configuration_validation(database: str, retention: int) -> None:
+async def test_schema_binding_rejects_invalid_database() -> None:
     with pytest.raises(ValueError, match=r"database.*retention"):
-        await ensure_schema("http://localhost:8123", database, "writer", "password", retention, 14)
+        NativeTraceStorage("db; DROP DATABASE default", "http://localhost:8123")
+
+
+@pytest.mark.asyncio
+async def test_schema_binding_rejects_non_positive_retention() -> None:
+    storage: Final = NativeTraceStorage("traces", "http://localhost:8123")
+    with pytest.raises(ValueError, match=r"database.*retention"):
+        await storage.ensure_schema(0, 14)
 
 
 @pytest.mark.asyncio
@@ -53,15 +56,10 @@ async def test_schema_setup_uses_writer_credentials_and_rejects_failed_statement
     recording_server.expected_requests = 2
     recording_server.enqueue(ResponseSpec(body=""))
     recording_server.enqueue(ResponseSpec(status=403, body="denied"))
+    writer_url: Final = recording_server.base_url.replace("http://", "http://writer:p%40ss%2Fword%25@")
+    storage: Final = NativeTraceStorage("trace_test", writer_url + "?database=wrong&readonly=1")
     with pytest.raises(RuntimeError, match="schema setup failed with HTTP status 403"):
-        await ensure_schema(
-            recording_server.base_url + "?database=wrong&readonly=1",
-            "trace_test",
-            "writer",
-            "p@ss/word%",
-            7,
-            14,
-        )
+        await storage.ensure_schema(7, 14)
     assert len(recording_server.requests) == 2
     assert recording_server.requests[0].raw_body.startswith(b"CREATE DATABASE IF NOT EXISTS")
     assert recording_server.requests[1].raw_body.startswith(b"CREATE TABLE IF NOT EXISTS")
@@ -71,8 +69,15 @@ async def test_schema_setup_uses_writer_credentials_and_rejects_failed_statement
     ).decode()
 
 
-def test_insert_encoding_preserves_nanoseconds_through_bridge() -> None:
-    assert json.loads(encode_rows([{"Timestamp": 1_234_567_890, "Input": "hello"}])) == {
+@pytest.mark.asyncio
+async def test_insert_encodes_and_sends_rows(recording_server: RecordingServer) -> None:
+    recording_server.enqueue(ResponseSpec(body=""))
+    storage: Final = NativeTraceStorage("trace_test", recording_server.base_url)
+    await storage.insert_rows("otel_traces", [{"Timestamp": 1_234_567_890, "Input": "hello"}])
+    request: Final = recording_server.requests[0]
+    assert json.loads(gzip.decompress(request.raw_body)) == {
         "Input": "hello",
         "Timestamp": "1970-01-01T00:00:01.23456789Z",
     }
+    assert parse_qs(urlsplit(request.path).query)["query"] == ["INSERT INTO `trace_test`.otel_traces FORMAT JSONEachRow"]
+    assert request.headers["content-encoding"] == "gzip"

ui/litellm-dashboard/src/lib/http/schema.d.ts

@@ -21262,6 +21262,58 @@ export interface paths {
         patch?: never;
         trace?: never;
     };
+    "/v1/traces": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /** List Agent Traces */
+        get: operations["list_agent_traces_v1_traces_get"];
+        put?: never;
+        /** Ingest Otlp Traces */
+        post: operations["ingest_otlp_traces_v1_traces_post"];
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
+    "/v1/traces/{trace_id}": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /** Get Agent Trace */
+        get: operations["get_agent_trace_v1_traces__trace_id__get"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
+    "/v1/traces/{trace_id}/spans/{span_id}": {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        /** Get Agent Trace Span */
+        get: operations["get_agent_trace_span_v1_traces__trace_id__spans__span_id__get"];
+        put?: never;
+        post?: never;
+        delete?: never;
+        options?: never;
+        head?: never;
+        patch?: never;
+        trace?: never;
+    };
     "/v1/unified_access_group": {
         parameters: {
             query?: never;
@@ -74649,6 +74701,124 @@ export interface operations {
             };
         };
     };
+    list_agent_traces_v1_traces_get: {
+        parameters: {
+            query?: {
+                /** @description Window start, unix ms. Default: 24h ago */
+                start_ms?: number | null;
+                /** @description Window end, unix ms. Default: now */
+                end_ms?: number | null;
+                cursor?: string | null;
+            };
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": unknown;
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    ingest_otlp_traces_v1_traces_post: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path?: never;
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": unknown;
+                };
+            };
+        };
+    };
+    get_agent_trace_v1_traces__trace_id__get: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path: {
+                trace_id: string;
+            };
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": unknown;
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
+    get_agent_trace_span_v1_traces__trace_id__spans__span_id__get: {
+        parameters: {
+            query?: never;
+            header?: never;
+            path: {
+                trace_id: string;
+                span_id: string;
+            };
+            cookie?: never;
+        };
+        requestBody?: never;
+        responses: {
+            /** @description Successful Response */
+            200: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": unknown;
+                };
+            };
+            /** @description Validation Error */
+            422: {
+                headers: {
+                    [name: string]: unknown;
+                };
+                content: {
+                    "application/json": components["schemas"]["HTTPValidationError"];
+                };
+            };
+        };
+    };
     list_access_groups_v1_unified_access_group_get: {
         parameters: {
             query?: never;