diff --git a/Cargo.lock b/Cargo.lock index e5a8ed0e..e5845df0 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -334,6 +334,7 @@ dependencies = [ "reqwest 0.12.28", "serde", "serde_json", + "sha2 0.10.9", "thiserror 1.0.69", "tokio", "tower 0.5.3", diff --git a/crates/aisix-proxy/Cargo.toml b/crates/aisix-proxy/Cargo.toml index 06d1a424..8807de6a 100644 --- a/crates/aisix-proxy/Cargo.toml +++ b/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 diff --git a/crates/aisix-proxy/src/audio.rs b/crates/aisix-proxy/src/audio.rs index 226ed1ce..47a4409a 100644 --- a/crates/aisix-proxy/src/audio.rs +++ b/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, } // ───────────────────────────────────────────────────────────────────────────── @@ -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, crate::redact::RedactionCounts, + Option, ), 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) { diff --git a/crates/aisix-proxy/src/embeddings.rs b/crates/aisix-proxy/src/embeddings.rs index 7c41ea3d..0556fae3 100644 --- a/crates/aisix-proxy/src/embeddings.rs +++ b/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, /// 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, 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)] diff --git a/crates/aisix-proxy/src/images.rs b/crates/aisix-proxy/src/images.rs index c8c9878f..2276a89d 100644 --- a/crates/aisix-proxy/src/images.rs +++ b/crates/aisix-proxy/src/images.rs @@ -12,7 +12,7 @@ use aisix_core::AppliedGuardrail; use aisix_gateway::{BridgeContext, BridgeError}; -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}; @@ -54,6 +54,11 @@ struct ImageDispatchSuccess { /// Per-detector PII mask counts (#932/#696) applied to the prompt. /// Attached to the emitted UsageEvent. Empty = no redaction. redactions: crate::redact::RedactionCounts, + /// Captured request/response content for content-capturing exporters + /// (#700, LiteLLM parity: the full image response JSON — url or + /// b64_json — truncated at the capture cap). `Some` only when an + /// exporter opted into `content_mode = full`. + captured_content: Option, } pub async fn image_generations( @@ -114,6 +119,7 @@ pub async fn image_generations( completion_tokens, &client, success.redactions.clone(), + success.captured_content.as_ref(), ); } success.response @@ -235,6 +241,19 @@ async fn dispatch( // Pre-#696 a mask-action detector was a silent no-op here. let redactions = crate::redact::redact_images_request(&resolved_chain, &mut body); + // Content capture (#700): the client-facing request body + // (post-#932-redaction) is the prompt; the response is the image JSON + // (url or b64_json — LiteLLM logs it unstripped too), truncated at the + // cap. + 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(model_name, &model_entry.id, &model_entry.value); let reservation = crate::quota::enforce(state, auth, Some(&model_rl)).await?; @@ -288,6 +307,16 @@ async fn dispatch( .map(|(prompt, completion)| u64::from(prompt) + u64::from(completion)) .unwrap_or(0); reservation.commit_tokens(total_tokens).await; + // Content capture (#700): the full response JSON (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(&resp_json).unwrap_or_default(), + cap as usize, + )), + _ => None, + }; Ok(ImageDispatchSuccess { response: Json(resp_json).into_response(), provider: provider_label, @@ -297,6 +326,7 @@ async fn dispatch( usage, upstream_called: true, redactions, + captured_content, }) } Err(BridgeError::Config(msg)) if msg.contains("does not support image generation") => { @@ -313,6 +343,7 @@ async fn dispatch( // No upstream call happened → handler skips emit. upstream_called: false, redactions, + captured_content: None, }) } Err(e) => { @@ -365,6 +396,9 @@ fn emit_usage_event( client: &ClientContext, // Per-detector PII mask counts (#932/#696). Empty = no redaction. redacted_entity_counts: crate::redact::RedactionCounts, + // 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 { @@ -390,7 +424,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 emit_access_log( diff --git a/crates/aisix-proxy/src/rerank.rs b/crates/aisix-proxy/src/rerank.rs index b8b9b3b9..7a4db372 100644 --- a/crates/aisix-proxy/src/rerank.rs +++ b/crates/aisix-proxy/src/rerank.rs @@ -10,7 +10,7 @@ //! The gateway appends `/v1/rerank`. use aisix_core::AppliedGuardrail; -use aisix_obs::{AccessLog, RequestOutcome, UsageEvent}; +use aisix_obs::{content_capture_cap, AccessLog, CapturedContent, RequestOutcome, UsageEvent}; use axum::extract::State; use axum::http::HeaderValue; use axum::response::{IntoResponse, Response}; @@ -47,6 +47,10 @@ struct RerankDispatchSuccess { /// Per-detector PII mask counts (#932/#696) applied to the request. /// Attached to the emitted UsageEvent. Empty = no redaction. redactions: crate::redact::RedactionCounts, + /// Captured request/response content for content-capturing exporters + /// (#700, LiteLLM parity: the full rerank response JSON). `Some` only + /// when an exporter opted into `content_mode = full`. + captured_content: Option, } /// Rerank has no completion side — only the input (query + docs) @@ -118,6 +122,7 @@ pub async fn rerank( &usage, &client, success.redactions.clone(), + success.captured_content.as_ref(), ); } success.response @@ -250,6 +255,18 @@ async fn dispatch( // upstream. Pre-#696 a mask-action detector was a silent no-op here. let redactions = crate::redact::redact_rerank_request(&resolved_chain, body); + // Content capture (#700): the client-facing request body + // (post-#932-redaction) is the prompt; the response is the upstream's + // rerank JSON verbatim (LiteLLM parity), both truncated at the cap. + 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(&model_name, &model_entry.id, &model_entry.value); let reservation = crate::quota::enforce(state, auth, Some(&model_rl)).await?; @@ -444,6 +461,17 @@ async fn dispatch( } }; + // Content capture (#700): the relayed response bytes are the JSON the + // caller sees; non-UTF-8 (unexpected) degrades to lossy text. + 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 resp = axum::response::Response::new(axum::body::Body::from(body_bytes)); // Forward content-type from upstream. @@ -475,6 +503,7 @@ async fn dispatch( applied_guardrails: applied_guardrails.clone(), usage, redactions, + captured_content, }) } @@ -537,6 +566,9 @@ fn emit_usage_event( client: &ClientContext, // Per-detector PII mask counts (#932/#696). Empty = no redaction. redacted_entity_counts: crate::redact::RedactionCounts, + // 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 { @@ -563,7 +595,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)); } /// Default upstream host for the rerank-supporting providers, diff --git a/tests/e2e/src/cases/sls-content-capture-nontext-e2e.test.ts b/tests/e2e/src/cases/sls-content-capture-nontext-e2e.test.ts new file mode 100644 index 00000000..05ff1a35 --- /dev/null +++ b/tests/e2e/src/cases/sls-content-capture-nontext-e2e.test.ts @@ -0,0 +1,285 @@ +import { createHash } from "node:crypto"; +import { afterAll, beforeAll, describe, expect, test } from "vitest"; +import { + AdminClient, + decodedTextFor, + EtcdClient, + spawnApp, + startMockSls, + startOpenAiUpstream, + waitConfigPropagation, + waitForToken, + type MockSls, + type OpenAiUpstream, + type SpawnedApp, +} from "../harness/index.js"; + +// ai-gateway#700 (follow-up to AISIX-Cloud#947): content capture for the +// non-text-generation endpoints, LiteLLM-parity scope — embeddings / rerank / +// images capture the post-redaction request JSON as the prompt and the full +// response JSON (vectors / scores / image url|b64) as the response; audio +// speech captures the input text only (binary response not logged); audio +// transcription represents the file by its sha256 and captures the transcript. + +const CALLER_PLAINTEXT = "sk-content-capture-nontext-PLAINTEXT"; +const CALLER_KEY_HASH = createHash("sha256").update(CALLER_PLAINTEXT).digest("hex"); +const PROVIDER_SECRET = "sk-mock-nontext"; + +const CREDENTIAL_REF = "mock"; +const SLS_PROJECT = "aisix-e2e-obs"; +const FULL_LOGSTORE = "full-events-nontext"; +const META_LOGSTORE = "meta-events-nontext"; + +const EMBED_PROMPT_TOKEN = "embed-prompt-tok-11aa22"; +const EMBED_RESPONSE_MARK = 0.987654; // survives into the captured response JSON +const RERANK_PROMPT_TOKEN = "rerank-prompt-tok-33bb44"; +const RERANK_DOC_TOKEN = "rerank-doc-tok-55cc66"; +const RERANK_RESPONSE_MARK = 0.918273; +const IMAGES_PROMPT_TOKEN = "images-prompt-tok-77dd88"; +const IMAGES_RESPONSE_URL = "https://img.example/images-response-tok-99ee00.png"; +const SPEECH_PROMPT_TOKEN = "speech-prompt-tok-aa11bb"; +const TRANSCRIPT_PROMPT_TOKEN = "transcribe-prompt-tok-cc22dd"; +const TRANSCRIPT_RESPONSE_TOKEN = "transcribe-response-tok-ee33ff"; +const AUDIO_FILE_BYTES = "ID3fakeaudio-nontext"; + +async function postJson(app: SpawnedApp, path: string, body: unknown): Promise { + return fetch(`${app.proxyUrl}${path}`, { + method: "POST", + headers: { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }, + body: JSON.stringify(body), + }); +} + +describe("sls content capture e2e (#700): embeddings / rerank / images / audio", () => { + let etcdReachable = false; + let embedUpstream: OpenAiUpstream | undefined; + let rerankUpstream: OpenAiUpstream | undefined; + let imagesUpstream: OpenAiUpstream | undefined; + let speechUpstream: OpenAiUpstream | undefined; + let transcribeUpstream: OpenAiUpstream | undefined; + let sls: MockSls | undefined; + const apps: SpawnedApp[] = []; + + beforeAll(async () => { + etcdReachable = await new EtcdClient().ping(); + if (!etcdReachable) return; + embedUpstream = await startOpenAiUpstream({ + nonStreamBody: { + object: "list", + data: [{ object: "embedding", index: 0, embedding: [EMBED_RESPONSE_MARK, 0.1] }], + model: "text-embedding-3-small", + usage: { prompt_tokens: 4, total_tokens: 4 }, + }, + }); + rerankUpstream = await startOpenAiUpstream({ + nonStreamBody: { + results: [{ index: 0, relevance_score: RERANK_RESPONSE_MARK }], + usage: { total_tokens: 6 }, + }, + }); + imagesUpstream = await startOpenAiUpstream({ + nonStreamBody: { + created: 1_700_000_000, + data: [{ url: IMAGES_RESPONSE_URL }], + }, + }); + speechUpstream = await startOpenAiUpstream({ + nonStreamBody: { fake: "binary-audio-placeholder" }, + }); + transcribeUpstream = await startOpenAiUpstream({ + nonStreamBody: { + text: `the speaker said ${TRANSCRIPT_RESPONSE_TOKEN}`, + usage: { type: "tokens", input_tokens: 12, output_tokens: 5, total_tokens: 17 }, + }, + }); + sls = await startMockSls(); + }); + + afterAll(async () => { + await Promise.all(apps.map((a) => a.exit())); + await embedUpstream?.close(); + await rerankUpstream?.close(); + await imagesUpstream?.close(); + await speechUpstream?.close(); + await transcribeUpstream?.close(); + await sls?.close(); + }); + + test( + "full-capture exporter logs request + response content on all four endpoint families", + async (ctx) => { + if ( + !etcdReachable || + !embedUpstream || + !rerankUpstream || + !imagesUpstream || + !speechUpstream || + !transcribeUpstream || + !sls + ) { + ctx.skip(); + return; + } + const app = await spawnApp({ + extraEnv: { + [`SLS_CRED_${CREDENTIAL_REF.toUpperCase()}_AK_ID`]: "mock-akid", + [`SLS_CRED_${CREDENTIAL_REF.toUpperCase()}_AK_SECRET`]: "mock-secret", + }, + }); + apps.push(app); + const admin = new AdminClient(app.adminUrl, app.adminKey); + await admin.createObservabilityExporter({ + name: "sls-full-nontext", + enabled: true, + kind: "aliyun_sls", + endpoint: sls.url, + project: SLS_PROJECT, + logstore: FULL_LOGSTORE, + credential_ref: CREDENTIAL_REF, + content_mode: "full", + }); + await admin.createObservabilityExporter({ + name: "sls-meta-nontext", + enabled: true, + kind: "aliyun_sls", + endpoint: sls.url, + project: SLS_PROJECT, + logstore: META_LOGSTORE, + credential_ref: CREDENTIAL_REF, + content_mode: "metadata_only", + }); + const seed = async (name: string, modelName: string, upstream: OpenAiUpstream) => { + const pk = await admin.createProviderKey({ + display_name: `${name}-pk`, + secret: PROVIDER_SECRET, + api_base: `${upstream.baseUrl}/v1`, + }); + await admin.createModel({ + display_name: name, + provider: "openai", + model_name: modelName, + provider_key_id: pk.id, + }); + }; + await seed("cc700-embed", "text-embedding-3-small", embedUpstream); + await seed("cc700-rerank", "gpt-rerank", rerankUpstream); + await seed("cc700-images", "dall-e-3", imagesUpstream); + await seed("cc700-speech", "tts-1", speechUpstream); + await seed("cc700-transcribe", "whisper-1", transcribeUpstream); + await admin.createApiKey({ + key_hash: CALLER_KEY_HASH, + allowed_models: [ + "cc700-embed", + "cc700-rerank", + "cc700-images", + "cc700-speech", + "cc700-transcribe", + ], + }); + + await waitConfigPropagation(async () => { + try { + const r = await postJson(app, "/v1/embeddings", { + model: "cc700-embed", + input: "warmup", + }); + await r.text(); + return r.status === 200; + } catch { + return false; + } + }); + + // -- embeddings: prompt = request JSON, response = vectors JSON -- + const embedRes = await postJson(app, "/v1/embeddings", { + model: "cc700-embed", + input: `embed the token ${EMBED_PROMPT_TOKEN}`, + }); + expect(embedRes.status).toBe(200); + await embedRes.text(); + await waitForToken(sls, FULL_LOGSTORE, EMBED_PROMPT_TOKEN); + let fullText = decodedTextFor(sls, FULL_LOGSTORE); + expect(fullText).toContain(EMBED_PROMPT_TOKEN); + expect(fullText).toContain(String(EMBED_RESPONSE_MARK)); // vector captured + + // -- rerank: prompt = query + documents, response = scores JSON -- + const rerankRes = await postJson(app, "/v1/rerank", { + model: "cc700-rerank", + query: `find ${RERANK_PROMPT_TOKEN}`, + documents: [`doc about ${RERANK_DOC_TOKEN}`], + }); + expect(rerankRes.status).toBe(200); + await rerankRes.text(); + await waitForToken(sls, FULL_LOGSTORE, RERANK_PROMPT_TOKEN); + fullText = decodedTextFor(sls, FULL_LOGSTORE); + expect(fullText).toContain(RERANK_PROMPT_TOKEN); + expect(fullText).toContain(RERANK_DOC_TOKEN); + expect(fullText).toContain(String(RERANK_RESPONSE_MARK)); + + // -- images: prompt captured, response url captured -- + const imagesRes = await postJson(app, "/v1/images/generations", { + model: "cc700-images", + prompt: `paint ${IMAGES_PROMPT_TOKEN}`, + }); + expect(imagesRes.status).toBe(200); + await imagesRes.text(); + await waitForToken(sls, FULL_LOGSTORE, IMAGES_PROMPT_TOKEN); + fullText = decodedTextFor(sls, FULL_LOGSTORE); + expect(fullText).toContain(IMAGES_PROMPT_TOKEN); + expect(fullText).toContain(IMAGES_RESPONSE_URL); + + // -- audio speech: input text captured; binary response not logged -- + const speechRes = await postJson(app, "/v1/audio/speech", { + model: "cc700-speech", + input: `say ${SPEECH_PROMPT_TOKEN}`, + voice: "alloy", + }); + expect(speechRes.status).toBe(200); + await speechRes.arrayBuffer(); + await waitForToken(sls, FULL_LOGSTORE, SPEECH_PROMPT_TOKEN); + + // -- audio transcription: file → sha256, prompt field + transcript -- + const form = new FormData(); + form.set("model", "cc700-transcribe"); + form.set("prompt", `context: ${TRANSCRIPT_PROMPT_TOKEN}`); + form.set("file", new Blob([AUDIO_FILE_BYTES], { type: "audio/mpeg" }), "a.mp3"); + const trRes = await fetch(`${app.proxyUrl}/v1/audio/transcriptions`, { + method: "POST", + headers: { authorization: `Bearer ${CALLER_PLAINTEXT}` }, + body: form, + }); + expect(trRes.status).toBe(200); + await trRes.text(); + await waitForToken(sls, FULL_LOGSTORE, TRANSCRIPT_PROMPT_TOKEN); + fullText = decodedTextFor(sls, FULL_LOGSTORE); + expect(fullText).toContain(TRANSCRIPT_PROMPT_TOKEN); + expect(fullText).toContain(TRANSCRIPT_RESPONSE_TOKEN); + const expectedSha = createHash("sha256").update(AUDIO_FILE_BYTES).digest("hex"); + expect(fullText).toContain(expectedSha); // file represented by checksum + expect(fullText).not.toContain(AUDIO_FILE_BYTES); // never the raw bytes + + // -- metadata_only exporter received none of the content -- + // Wait until the LAST request's metadata (its requested_model, which + // both content modes carry) has landed in the meta logstore, so the + // negative assertions below can't pass trivially against an empty or + // lagging store. + await waitForToken(sls, META_LOGSTORE, "cc700-transcribe"); + const metaText = decodedTextFor(sls, META_LOGSTORE); + for (const token of [ + EMBED_PROMPT_TOKEN, + RERANK_PROMPT_TOKEN, + RERANK_DOC_TOKEN, + IMAGES_PROMPT_TOKEN, + SPEECH_PROMPT_TOKEN, + TRANSCRIPT_PROMPT_TOKEN, + TRANSCRIPT_RESPONSE_TOKEN, + ]) { + expect(metaText).not.toContain(token); + } + }, + 120_000, + ); +});