feat(obs): content capture for embeddings/rerank/images/audio (LiteLLM parity)

crates/aisix-proxy/Cargo.toml

@@ -52,6 +52,7 @@ dashmap.workspace = true
 # a constant that aliases against the modulus).
 rand = { workspace = true }
 chrono.workspace = true
+sha2.workspace = true
 thiserror.workspace = true
 tracing.workspace = true
 uuid.workspace = true

crates/aisix-proxy/src/audio.rs

@@ -19,7 +19,7 @@
 
 use aisix_core::AppliedGuardrail;
 use aisix_gateway::{ChatMessage, ChatResponse, FinishReason, UsageStats};
-use aisix_obs::{AccessLog, RequestOutcome, UsageEvent};
+use aisix_obs::{content_capture_cap, AccessLog, CapturedContent, RequestOutcome, UsageEvent};
 use axum::body::Bytes;
 use axum::extract::{Multipart, State};
 use axum::http::{header, HeaderMap};
@@ -65,6 +65,12 @@ struct AudioDispatchSuccess {
     /// `guardrail_blocked`) doesn't under-report spend — same convention as
     /// completions #911 [23] / responses #543.
     guardrail_blocked: bool,
+    /// Captured request/response content for content-capturing exporters
+    /// (#700, LiteLLM parity: the audio bytes are represented by their
+    /// sha256, text form fields verbatim; the response is the post-redaction
+    /// transcript). `Some` only when an exporter opted into
+    /// `content_mode = full`.
+    captured_content: Option<CapturedContent>,
 }
 
 // ─────────────────────────────────────────────────────────────────────────────
@@ -280,7 +286,15 @@ pub async fn speech(
         .to_string();
 
     match speech_dispatch(&state, &auth, body, &request_id, &client.source_ip).await {
-        Ok((resp, provider, model_id, provider_key_id, applied_guardrails, redactions)) => {
+        Ok((
+            resp,
+            provider,
+            model_id,
+            provider_key_id,
+            applied_guardrails,
+            redactions,
+            captured,
+        )) => {
             let elapsed = started.elapsed();
             emit_access_log(
                 "POST",
@@ -319,6 +333,7 @@ pub async fn speech(
                 &client,
                 redactions,
                 /* guardrail_blocked */ false,
+                captured.as_ref(),
             );
             resp
         }
@@ -473,6 +488,51 @@ async fn multipart_dispatch(
         }
     }
 
+    // Content capture (#700, LiteLLM parity): the audio bytes are NOT
+    // captured — the file is represented by its sha256 (exactly what LiteLLM
+    // logs for transcription input); the text form fields (model, prompt,
+    // language, …) are captured verbatim, POST-redaction so a masked
+    // `prompt` field stays masked in the exported content.
+    let content_cap = content_capture_cap(
+        snapshot
+            .observability_exporters
+            .entries()
+            .iter()
+            .map(|e| &e.value),
+    );
+    let captured_prompt = content_cap.map(|_| {
+        use sha2::Digest;
+        let mut obj = serde_json::Map::new();
+        // Appends on a repeated name (multipart allows repeats and all are
+        // forwarded) so no field disappears from the export. The filename is
+        // deliberately NOT captured — it is user-controlled text that skips
+        // the redaction path; the checksum alone represents the file,
+        // matching LiteLLM.
+        let mut push = |key: String, value: String| match obj.get_mut(&key) {
+            Some(Value::String(existing)) => {
+                existing.push('\n');
+                existing.push_str(&value);
+            }
+            _ => {
+                obj.insert(key, Value::String(value));
+            }
+        };
+        for (name, _, _, data) in &fields {
+            match std::str::from_utf8(data) {
+                Ok(text) if name != "file" => {
+                    push(name.clone(), text.to_string());
+                }
+                _ => {
+                    push(
+                        format!("{name}_sha256"),
+                        format!("{:x}", sha2::Sha256::digest(data)),
+                    );
+                }
+            }
+        }
+        serde_json::to_string(&Value::Object(obj)).unwrap_or_default()
+    });
+
     let model_rl =
         crate::quota::ModelRateLimit::from_model(&model_name, &model_entry.id, &model_entry.value);
     let reservation = crate::quota::enforce(state, auth, Some(&model_rl)).await?;
@@ -657,6 +717,9 @@ async fn multipart_dispatch(
                     applied_guardrails,
                     redactions,
                     guardrail_blocked: true,
+                    // The blocked transcript never reached the client — no
+                    // content capture, matching the chat surface.
+                    captured_content: None,
                 });
             }
         }
@@ -673,6 +736,18 @@ async fn multipart_dispatch(
             None => body_bytes,
         };
 
+    // Content capture (#700): the transcript the caller sees — read from
+    // the POST-redaction body so masked PII stays masked in the exported
+    // content.
+    let captured_content = match (&captured_prompt, content_cap) {
+        (Some(prompt), Some(cap)) => Some(CapturedContent::new(
+            prompt,
+            &String::from_utf8_lossy(&body_bytes),
+            cap as usize,
+        )),
+        _ => None,
+    };
+
     let mut out = axum::response::Response::new(axum::body::Body::from(body_bytes));
     copy_response_header(&upstream_headers, &mut out, header::CONTENT_TYPE);
     Ok(AudioDispatchSuccess {
@@ -685,6 +760,7 @@ async fn multipart_dispatch(
         applied_guardrails,
         redactions,
         guardrail_blocked: false,
+        captured_content,
     })
 }
 
@@ -739,6 +815,7 @@ async fn speech_dispatch(
         String,
         Vec<AppliedGuardrail>,
         crate::redact::RedactionCounts,
+        Option<CapturedContent>,
     ),
     ProxyError,
 > {
@@ -799,6 +876,24 @@ async fn speech_dispatch(
     // Pre-#696 a mask-action detector was a silent no-op here.
     let redactions = crate::redact::redact_speech_request(&resolved_chain, &mut body);
 
+    // Content capture (#700, LiteLLM parity): the post-redaction request
+    // body (the `input` text to synthesize) is the prompt; the binary audio
+    // response is NOT captured — LiteLLM logs no TTS response either.
+    let captured_content = content_capture_cap(
+        snapshot
+            .observability_exporters
+            .entries()
+            .iter()
+            .map(|e| &e.value),
+    )
+    .map(|cap| {
+        CapturedContent::new(
+            &serde_json::to_string(&body).unwrap_or_default(),
+            "",
+            cap as usize,
+        )
+    });
+
     let model_rl =
         crate::quota::ModelRateLimit::from_model(&model_name, &model_entry.id, &model_entry.value);
     let reservation = crate::quota::enforce(state, auth, Some(&model_rl)).await?;
@@ -929,6 +1024,7 @@ async fn speech_dispatch(
         pk_entry.id.to_string(),
         applied_guardrails,
         redactions,
+        captured_content,
     ))
 }
 
@@ -976,6 +1072,7 @@ fn emit_audio_usage(
         client,
         success.redactions.clone(),
         success.guardrail_blocked,
+        success.captured_content.as_ref(),
     );
 }
 
@@ -1004,6 +1101,9 @@ fn emit_usage_event(
     redacted_entity_counts: crate::redact::RedactionCounts,
     // #696: transcript blocked by an output guardrail after upstream billing.
     guardrail_blocked: bool,
+    // Captured request/response content (#700). Forwarded only to `fan_out`,
+    // never to the CP sink.
+    content: Option<&CapturedContent>,
 ) {
     let snap = state.snapshot.load();
     let mut event = UsageEvent {
@@ -1032,7 +1132,7 @@ fn emit_usage_event(
     let exporters = snap.observability_exporters.entries();
     state
         .otlp_fan_out
-        .fan_out(&event, None, exporters.iter().map(|e| &e.value));
+        .fan_out(&event, content, exporters.iter().map(|e| &e.value));
 }
 
 fn copy_response_header(src: &HeaderMap, dst: &mut Response, name: header::HeaderName) {

crates/aisix-proxy/src/embeddings.rs

@@ -16,7 +16,7 @@
 
 use aisix_core::AppliedGuardrail;
 use aisix_gateway::{BridgeContext, BridgeError, ChatFormat, ChatMessage, EmbeddingRequest};
-use aisix_obs::{AccessLog, RequestOutcome, UsageEvent};
+use aisix_obs::{content_capture_cap, AccessLog, CapturedContent, RequestOutcome, UsageEvent};
 use axum::extract::State;
 use axum::http::StatusCode;
 use axum::response::{IntoResponse, Response};
@@ -35,7 +35,7 @@ use crate::state::ProxyState;
 ///
 /// `input` may be a single string **or** an array of strings; both are
 /// handled by the `InputField` helper so callers don't need to know.
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, serde::Serialize)]
 pub struct EmbeddingRequestBody {
     pub model: String,
     pub input: InputField,
@@ -47,7 +47,7 @@ pub struct EmbeddingRequestBody {
 
 /// Deserialises both `"text"` and `["text", ...]` forms of the
 /// OpenAI embeddings `input` field.
-#[derive(Debug, Deserialize)]
+#[derive(Debug, Deserialize, serde::Serialize)]
 #[serde(untagged)]
 pub enum InputField {
     Single(String),
@@ -173,6 +173,7 @@ pub async fn embeddings(
                     success.prompt_tokens,
                     &client,
                     success.redactions.clone(),
+                    success.captured_content.as_ref(),
                 );
             }
             success.response
@@ -232,6 +233,11 @@ struct EmbedDispatchSuccess {
     applied_guardrails: Vec<AppliedGuardrail>,
     /// Per-detector PII mask counts (#932) applied to the input.
     redactions: crate::redact::RedactionCounts,
+    /// Captured request/response content for content-capturing exporters
+    /// (#700, LiteLLM parity: the full response JSON — vectors included —
+    /// truncated at the capture cap). `Some` only when an exporter opted
+    /// into `content_mode = full`.
+    captured_content: Option<CapturedContent>,
     prompt_tokens: u32,
     /// `true` when the dispatch produced a real 200 from the upstream
     /// (we have authoritative usage data to attribute). `false` for the
@@ -339,6 +345,19 @@ async fn dispatch(
         }
     }
 
+    // Content capture (#700): the client-facing request body
+    // (post-#932-redaction) is the prompt; the full response JSON — vectors
+    // included, matching LiteLLM's logging payload — is the response, both
+    // truncated at the largest cap an enabled exporter wants.
+    let content_cap = content_capture_cap(
+        snapshot
+            .observability_exporters
+            .entries()
+            .iter()
+            .map(|e| &e.value),
+    );
+    let captured_prompt = content_cap.map(|_| serde_json::to_string(&body).unwrap_or_default());
+
     let model_rl =
         crate::quota::ModelRateLimit::from_model(&body.model, &model_entry.id, &model_entry.value);
     let reservation = crate::quota::enforce(state, auth, Some(&model_rl)).await?;
@@ -383,6 +402,17 @@ async fn dispatch(
             // embed_resp into the JSON response — the handler needs
             // this for UsageEvent emission downstream (#226).
             let prompt_tokens = embed_resp.usage.prompt_tokens;
+            // Content capture (#700): the full response JSON, vectors
+            // included (LiteLLM parity); CapturedContent::new truncates to
+            // the cap.
+            let captured_content = match (&captured_prompt, content_cap) {
+                (Some(prompt), Some(cap)) => Some(CapturedContent::new(
+                    prompt,
+                    &serde_json::to_string(&embed_resp).unwrap_or_default(),
+                    cap as usize,
+                )),
+                _ => None,
+            };
             Ok(EmbedDispatchSuccess {
                 response: Json(embed_resp).into_response(),
                 provider: provider_label,
@@ -392,6 +422,7 @@ async fn dispatch(
                 redactions: redactions.clone(),
                 prompt_tokens,
                 upstream_called: true,
+                captured_content,
             })
         }
         Err(BridgeError::Config(msg)) if msg.contains("does not support embeddings") => {
@@ -411,6 +442,7 @@ async fn dispatch(
                 applied_guardrails: applied_guardrails.clone(),
                 redactions: redactions.clone(),
                 prompt_tokens: 0,
+                captured_content: None,
                 // No upstream call happened — the handler reads this
                 // and skips UsageEvent emission. Distinguished from
                 // `prompt_tokens == 0` so a 200 that legitimately
@@ -492,6 +524,9 @@ fn emit_usage_event(
     client: &ClientContext,
     // Per-detector PII mask counts (#932) applied to the input.
     redacted_entity_counts: crate::redact::RedactionCounts,
+    // Captured request/response content (#700). Forwarded only to `fan_out`,
+    // never to the CP sink.
+    content: Option<&CapturedContent>,
 ) {
     // Only populate fields meaningful to /v1/embeddings; rely on
     // UsageEvent's `#[derive(Default)]` for everything else. Wire-level
@@ -545,7 +580,7 @@ fn emit_usage_event(
     let exporters = snap.observability_exporters.entries();
     state
         .otlp_fan_out
-        .fan_out(&event, None, exporters.iter().map(|e| &e.value));
+        .fan_out(&event, content, exporters.iter().map(|e| &e.value));
 }
 
 #[cfg(test)]