From 275095c3e6002cc39022405f189467ec530a7d6b Mon Sep 17 00:00:00 2001 From: Jarvis Date: Tue, 18 Aug 2026 12:57:00 +0800 Subject: [PATCH 1/3] feat(passthrough): detect the body envelope per request, drop protocol/streaming config Remove the passthrough route's two pre-release knobs (the kind first ships in 0.10.0, so nothing released carries them): - `protocol` is gone. The envelope is detected once per exchange from the request body's top-level keys (`messages` -> chat, `input` -> Responses, `prompt` -> completions, anything else -> opaque) and drives guardrail extraction, capture, and usage for the request, the buffered response, and every stream frame. Detection never affects the relay, and an envelope that yields no text falls back to whole-body scanning, so a detected request is never scanned more narrowly than an opaque one. Opaque buffered responses are still not probed for usage (no phantom tokens from usage-shaped REST bodies). - `streaming` is gone. SSE upstream responses always relay incrementally; everything else buffers (guardrails and usage need the whole body). The exchange timeout keeps its streaming-safe semantics: header phase + non-SSE body reads, never a healthy relay. This collapses the forward-proxy Copilot setup to a single route: the CLI's /responses turns, FIM completions, MCP JSON-RPC and GitHub REST all sort themselves out per request with no per-protocol routes. E2E: new envelope-auto-detection case asserts, via a real OTLP exporter, that a route with no protocol config records chat usage from a buffered body and Responses usage from the terminal stream event, and that JSON-RPC traffic records zero tokens. --- crates/aisix-core/src/lib.rs | 4 +- crates/aisix-core/src/models/mod.rs | 4 +- .../src/models/passthrough_route.rs | 72 ++--- crates/aisix-core/src/models/schema.rs | 12 +- crates/aisix-proxy/src/passthrough_route.rs | 259 +++++++++++++----- .../resources/passthrough_route.schema.json | 53 +--- .../src/cases/passthrough-route-e2e.test.ts | 147 +++++++++- 7 files changed, 365 insertions(+), 186 deletions(-) diff --git a/crates/aisix-core/src/lib.rs b/crates/aisix-core/src/lib.rs index 1d34bcc3e..d67c5cab7 100644 --- a/crates/aisix-core/src/lib.rs +++ b/crates/aisix-core/src/lib.rs @@ -49,8 +49,8 @@ pub use models::{ GuardrailExecution, GuardrailHookPoint, GuardrailKind, GuardrailMetricsSink, GuardrailMonitorHit, KeywordConfig, KeywordPattern, McpAuthType, McpProtocolVersion, McpRateLimit, McpServer, McpServerType, McpTransport, Model, ObservabilityExporter, - ParamConstraints, PassthroughAuthMode, PassthroughCredentialMode, PassthroughProtocol, - PassthroughRoute, PolicyScope, PolicyWindow, ProviderKey, RateLimit, RateLimitPolicy, + ParamConstraints, PassthroughAuthMode, PassthroughCredentialMode, PassthroughRoute, + PolicyScope, PolicyWindow, ProviderKey, RateLimit, RateLimitPolicy, RequestOverrides, ResponseOverrides, Routing, RoutingStrategy, RoutingTarget, SchemaError, StreamDoneMarker, TelemetryKind, TelemetryTags, WhenAllUnavailablePolicy, DEFAULT_COOLDOWN_TRIGGER_STATUSES, diff --git a/crates/aisix-core/src/models/mod.rs b/crates/aisix-core/src/models/mod.rs index 84469328f..f723fa55e 100644 --- a/crates/aisix-core/src/models/mod.rs +++ b/crates/aisix-core/src/models/mod.rs @@ -62,9 +62,7 @@ pub use observability_exporter::{ ObjectStoreProvider, ObservabilityExporter, OtlpHttpConfig, SlsContentMode, }; pub use oidc_provider::{BoundClaimExpect, OidcProvider}; -pub use passthrough_route::{ - PassthroughAuthMode, PassthroughCredentialMode, PassthroughProtocol, PassthroughRoute, -}; +pub use passthrough_route::{PassthroughAuthMode, PassthroughCredentialMode, PassthroughRoute}; pub use policy_conditions::{ eval_condition_nodes, validate_condition_nodes, ConditionGroup, ConditionInput, ConditionLogic, ConditionNode, ConditionOperator, ConditionValue, GroupByDimension, PolicyAction, diff --git a/crates/aisix-core/src/models/passthrough_route.rs b/crates/aisix-core/src/models/passthrough_route.rs index 9064232cf..5456547f8 100644 --- a/crates/aisix-core/src/models/passthrough_route.rs +++ b/crates/aisix-core/src/models/passthrough_route.rs @@ -113,17 +113,6 @@ pub struct PassthroughRoute { #[schemars(length(min = 1))] pub provider_key_id: Option, - /// Body-shape hint for auditing, guardrails and usage extraction. - /// Parsing is best-effort: a body that does not match the declared - /// shape degrades to `raw` handling, it is never rejected for shape. - #[serde(default)] - pub protocol: PassthroughProtocol, - - /// Relay `text/event-stream` upstream responses incrementally. When - /// `false` streaming responses are fully buffered like any other body. - #[serde(default = "default_true")] - pub streaming: bool, - /// Optional header carrying the end-user identity injected by the /// upstream network device (e.g. `x-aisix-user`). Its value is recorded /// on the usage event for per-employee audit attribution and stripped @@ -135,12 +124,11 @@ pub struct PassthroughRoute { #[schemars(regex(pattern = "^[!#$%&'*+.^_`|~0-9a-z-]+$"), length(min = 1))] pub identity_header: Option, - /// Maximum time, in milliseconds, for a non-streaming upstream - /// exchange. On a streaming route it bounds the response-header phase - /// and any non-SSE body read, but never a healthy SSE relay (which - /// ends with the upstream stream or the client hanging up). When - /// omitted, the gateway default request timeout applies to - /// non-streaming exchanges only. + /// Maximum time, in milliseconds, for the upstream exchange. Bounds + /// the response-header phase and any non-SSE body read, but never a + /// healthy SSE relay (which ends with the upstream stream or the + /// client hanging up). When omitted, the gateway default request + /// timeout applies the same way. #[serde(default, skip_serializing_if = "Option::is_none")] #[schemars(range(min = 1))] pub timeout_ms: Option, @@ -193,30 +181,6 @@ pub enum PassthroughCredentialMode { ForwardClient, } -/// Body-shape hint for a passthrough route. -#[derive( - Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize, schemars::JsonSchema, -)] -#[serde(rename_all = "snake_case")] -pub enum PassthroughProtocol { - /// No parsing: bodies are opaque byte blobs (guardrails scan them as - /// one lossy-UTF-8 text). - #[default] - Raw, - /// OpenAI-compatible chat envelope (`messages`, streamed - /// `choices[].delta.content`, final-chunk / response `usage`). - OpenaiChat, - /// OpenAI-compatible legacy completions / FIM envelope (`prompt` [+ - /// `suffix`], streamed `choices[].text`, `usage`). - OpenaiCompletions, - /// OpenAI Responses API envelope: `input` items on the request, - /// `output` items on the response, `response.output_text.delta` - /// events while streaming, and `usage` in the - /// `input_tokens`/`output_tokens` spelling — carried on the terminal - /// `response.completed` event when the response streams. - OpenaiResponses, -} - impl Resource for PassthroughRoute { fn id(&self) -> &str { &self.runtime_id @@ -476,8 +440,6 @@ mod tests { let r = minimal(); assert_eq!(r.auth_mode, PassthroughAuthMode::GatewayKey); assert_eq!(r.credential_mode, PassthroughCredentialMode::Inject); - assert_eq!(r.protocol, PassthroughProtocol::Raw); - assert!(r.streaming); assert!(r.enabled); assert!(!r.preserve_host); } @@ -608,6 +570,30 @@ mod coupling_tests { assert!(validate_passthrough_route_lenient(&doc).is_ok()); } + #[test] + fn removed_protocol_and_streaming_fields_are_unknown() { + // The pre-0.10.0 dev cycle carried `protocol` / `streaming` route + // fields; both were removed before the kind ever shipped (the + // envelope is now detected per request, SSE always relays + // incrementally). The strict write path rejects them like any + // unknown field; the lenient etcd path tolerates-and-strips. + for (field, value) in [ + ("protocol", json!("openai_responses")), + ("streaming", json!(false)), + ] { + let mut doc = base(); + doc[field] = value; + assert!( + validate_passthrough_route(&doc).is_err(), + "strict must reject unknown field {field}" + ); + assert!( + validate_passthrough_route_lenient(&doc).is_ok(), + "lenient must tolerate unknown field {field}" + ); + } + } + #[test] fn cross_mode_leftover_companions_are_rejected() { // A companion outside its mode is never consulted at runtime, so diff --git a/crates/aisix-core/src/models/schema.rs b/crates/aisix-core/src/models/schema.rs index 42c3b25f4..088961c6d 100644 --- a/crates/aisix-core/src/models/schema.rs +++ b/crates/aisix-core/src/models/schema.rs @@ -715,7 +715,7 @@ pub fn claim_mapping_root_schema() -> Value { /// the [`PassthroughRoute`](crate::models::PassthroughRoute) struct. Uses the /// nullable `Option` representation (`true`) so unset optional fields accept /// an explicit `null` as well as being absent. The `auth_mode` / -/// `credential_mode` / `protocol` closed sets come from their enums; every +/// `credential_mode` closed sets come from their enums; every /// cross-field invariant (match dimensions, target shape, per-mode required /// companions) is injected as an `allOf` (see /// [`super::passthrough_route::passthrough_route_coupling`]) so the strict @@ -755,16 +755,6 @@ pub fn passthrough_route_root_schema() -> Value { ("forward_client", "Forward the caller's own credential"), ], ); - title_single_value_enum_variants( - defs, - "PassthroughProtocol", - &[ - ("raw", "Opaque body"), - ("openai_chat", "OpenAI-compatible chat"), - ("openai_completions", "OpenAI-compatible completions / FIM"), - ("openai_responses", "OpenAI Responses API"), - ], - ); } schema } diff --git a/crates/aisix-proxy/src/passthrough_route.rs b/crates/aisix-proxy/src/passthrough_route.rs index d9c13fbbe..a6b6c6d0d 100644 --- a/crates/aisix-proxy/src/passthrough_route.rs +++ b/crates/aisix-proxy/src/passthrough_route.rs @@ -2,10 +2,22 @@ //! //! Replaces the removed implicit `/passthrough/:provider/*rest` tunnel: a //! route binds a gateway entry (path prefix and/or inbound `Host`) to ONE -//! upstream target with its own gateway-auth mode, credential handling, -//! protocol hint, and streaming behavior. There is no implicit -//! provider→Model credential borrowing (AISIX-Cloud#1127) and no forced -//! `Authorization` replacement (AISIX-Cloud#1312). +//! upstream target with its own gateway-auth mode and credential handling. +//! There is no implicit provider→Model credential borrowing +//! (AISIX-Cloud#1127) and no forced `Authorization` replacement +//! (AISIX-Cloud#1312). +//! +//! ## Envelope detection +//! +//! The request body's envelope is detected once per exchange from its +//! top-level keys ([`detect_protocol`]) and drives guardrail text +//! extraction, content capture and usage extraction for both the request +//! and the response (buffered or streamed). Detection never affects the +//! relay itself — bodies are forwarded verbatim regardless — and every +//! extraction degrades to the whole lossy-UTF-8 body when the detected +//! shape yields no text, so a mis-detected envelope loses no audit +//! coverage. SSE upstream responses are always relayed incrementally; +//! anything else is buffered (guardrails and usage need the whole body). //! //! ## Entry points //! @@ -51,9 +63,7 @@ use axum::middleware::Next; use axum::response::{IntoResponse, Response}; use aisix_core::resource::ResourceEntry; -use aisix_core::{ - PassthroughAuthMode, PassthroughCredentialMode, PassthroughProtocol, PassthroughRoute, -}; +use aisix_core::{PassthroughAuthMode, PassthroughCredentialMode, PassthroughRoute}; use crate::auth::AuthenticatedKey; use crate::error::ProxyError; @@ -577,9 +587,13 @@ async fn dispatch( let resolved_chain = state.guardrail_index.resolve(&guardrail_ctx); let mut monitor_hits: Vec = Vec::new(); - // INPUT guardrails on the (protocol-extracted) request text. + // Envelope detection: once per exchange, from the request body's + // top-level keys; the response and stream frames reuse it. + let protocol = detect_protocol(&body_bytes); + + // INPUT guardrails on the (envelope-extracted) request text. if !resolved_chain.is_empty() { - let text = request_guardrail_text(route.protocol, &body_bytes); + let text = request_guardrail_text(protocol, &body_bytes); let chat = aisix_gateway::ChatFormat::new( route.name.clone(), vec![aisix_gateway::ChatMessage::user(text)], @@ -619,7 +633,7 @@ async fn dispatch( .iter() .map(|e| &e.value), ); - let captured_prompt = content_cap.map(|_| request_guardrail_text(route.protocol, &body_bytes)); + let captured_prompt = content_cap.map(|_| request_guardrail_text(protocol, &body_bytes)); // Rate limits AFTER the input guardrail so a content block doesn't burn // an RPM slot (matching the typed endpoints). The body's `model` field @@ -720,33 +734,27 @@ async fn dispatch( builder = builder.body(body_bytes.clone()); } - // Exchange bound. A non-streaming route carries a plain total-exchange - // timeout (route override, else the gateway default). A streaming - // route must not bound the relay itself — a healthy long-lived SSE + // Exchange bound. The timeout (route override, else the gateway + // default) must not bound the relay itself — a healthy long-lived SSE // stream is the point — but a blackholed upstream still can't pin the // connection: the header phase (and, below, a non-SSE body read) get - // the same bound via an explicit timer. + // the bound via an explicit timer. let exchange_timeout = route .timeout_ms .map(Duration::from_millis) .or(state.default_timeouts.request); - if !route.streaming { - if let Some(d) = exchange_timeout { - builder = builder.timeout(d); - } - } let bridge_timeout = |d: Duration| aisix_gateway::BridgeError::Timeout { elapsed_ms: d.as_millis().min(u64::MAX as u128) as u64, cause: "passthrough route upstream exchange".into(), }; let send_fut = builder.send(); - let sent = match (route.streaming, exchange_timeout) { - (true, Some(d)) => match tokio::time::timeout(d, send_fut).await { + let sent = match exchange_timeout { + Some(d) => match tokio::time::timeout(d, send_fut).await { Ok(r) => r, Err(_) => return Err(RouteError::of(ProxyError::Bridge(bridge_timeout(d)), &auth)), }, - _ => send_fut.await, + None => send_fut.await, }; let upstream_resp = sent.map_err(|e| { RouteError::of( @@ -799,9 +807,9 @@ async fn dispatch( emitted: false, }; - if route.streaming && is_sse { + if is_sse { return Ok(stream_response( - route.protocol, + protocol, resolved_chain, upstream_resp, resp_headers, @@ -813,19 +821,19 @@ async fn dispatch( // ----- buffered response ----- - // A streaming route reaching this branch got a non-SSE answer; its - // reqwest request carries no built-in timeout, so the body read gets - // the exchange bound explicitly (same blackhole guard as the send). + // A non-SSE answer: the reqwest request carries no built-in timeout, + // so the body read gets the exchange bound explicitly (same blackhole + // guard as the send). let body_fut = upstream_resp.bytes(); - let read = match (route.streaming, exchange_timeout) { - (true, Some(d)) => match tokio::time::timeout(d, body_fut).await { + let read = match exchange_timeout { + Some(d) => match tokio::time::timeout(d, body_fut).await { Ok(r) => r, Err(_) => { telemetry.emitted = true; return Err(RouteError::of(ProxyError::Bridge(bridge_timeout(d)), &auth)); } }, - _ => body_fut.await, + None => body_fut.await, }; let resp_body = read.map_err(|e| { telemetry.emitted = true; @@ -835,9 +843,9 @@ async fn dispatch( ) })?; - // OUTPUT guardrails on the (protocol-extracted) response text. + // OUTPUT guardrails on the (envelope-extracted) response text. if !resolved_chain.is_empty() { - let text = response_guardrail_text(route.protocol, &resp_body); + let text = response_guardrail_text(protocol, &resp_body); let synth = aisix_gateway::ChatResponse { id: String::new(), model: route.name.clone(), @@ -873,12 +881,12 @@ async fn dispatch( } } - if let Some((p, c)) = response_usage(route.protocol, &resp_body) { + if let Some((p, c)) = response_usage(protocol, &resp_body) { telemetry.prompt_tokens = p; telemetry.completion_tokens = c; } if telemetry.content_cap.is_some() { - telemetry.response_text = response_guardrail_text(route.protocol, &resp_body); + telemetry.response_text = response_guardrail_text(protocol, &resp_body); } let mut response = Response::builder() @@ -1007,43 +1015,92 @@ fn content_text(v: &serde_json::Value) -> String { } } -/// The request text a guardrail scans, per the route's protocol hint. -/// Parsing is best-effort: anything that doesn't match the declared shape -/// degrades to the raw lossy-UTF-8 body. +/// The body envelope detected for one exchange. Not configuration: +/// detected per request from the body's top-level keys +/// ([`detect_protocol`]) and sticky for the exchange — the buffered +/// response and every stream frame are read with the same detection. It +/// drives extraction (guardrail text, capture, usage) only; the relay +/// forwards bytes verbatim regardless. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum PassthroughProtocol { + /// No recognized envelope: bodies are opaque (guardrails scan them as + /// one lossy-UTF-8 text; buffered responses are not probed for usage). + Raw, + /// OpenAI-compatible chat envelope (`messages`, streamed + /// `choices[].delta.content`, final-chunk / response `usage`). + OpenaiChat, + /// OpenAI-compatible legacy completions / FIM envelope (`prompt` [+ + /// `suffix`], streamed `choices[].text`, `usage`). + OpenaiCompletions, + /// OpenAI Responses API envelope: `input` on the request, `output` + /// items on the response, `response.output_text.delta` events while + /// streaming, and `usage` in the `input_tokens`/`output_tokens` + /// spelling — carried on the terminal `response.completed` event when + /// the response streams. + OpenaiResponses, +} + +/// Detect the request envelope from the body's top-level keys. The three +/// LLM envelopes are structurally exclusive — `messages`, `input` and +/// `prompt` are each the required carrier field of exactly one API — so +/// real LLM traffic detects unambiguously, and everything else (JSON-RPC, +/// REST, non-JSON, empty/GET bodies) is `Raw`. An unknown API colliding +/// with a carrier key costs nothing: detection drives extraction only, +/// and extraction degrades to the whole body when the detected shape +/// yields no text. +fn detect_protocol(body: &[u8]) -> PassthroughProtocol { + let Ok(v) = serde_json::from_slice::(body) else { + return PassthroughProtocol::Raw; + }; + if v.get("messages").is_some_and(serde_json::Value::is_array) { + PassthroughProtocol::OpenaiChat + } else if v + .get("input") + .is_some_and(|i| i.is_string() || i.is_array()) + { + PassthroughProtocol::OpenaiResponses + } else if v + .get("prompt") + .is_some_and(|p| p.is_string() || p.is_array()) + { + PassthroughProtocol::OpenaiCompletions + } else { + PassthroughProtocol::Raw + } +} + +/// The request text a guardrail scans, per the detected envelope. +/// Extraction is best-effort: a shape that yields no text degrades to the +/// raw lossy-UTF-8 body, so detection never loses audit coverage. fn request_guardrail_text(protocol: PassthroughProtocol, body: &[u8]) -> String { let raw = || String::from_utf8_lossy(body).into_owned(); let Ok(v) = serde_json::from_slice::(body) else { return raw(); }; - match protocol { - PassthroughProtocol::Raw => raw(), - PassthroughProtocol::OpenaiChat => match v.get("messages").and_then(|m| m.as_array()) { - Some(msgs) => msgs - .iter() - .filter_map(|m| m.get("content").map(content_text)) - .filter(|t| !t.is_empty()) - .collect::>() - .join("\n"), - None => raw(), - }, + let extracted = match protocol { + PassthroughProtocol::Raw => return raw(), + PassthroughProtocol::OpenaiChat => v + .get("messages") + .and_then(|m| m.as_array()) + .map(|msgs| { + msgs.iter() + .filter_map(|m| m.get("content").map(content_text)) + .filter(|t| !t.is_empty()) + .collect::>() + .join("\n") + }) + .unwrap_or_default(), // Responses API: `input` is either a bare string or an array of // items whose `content` parts carry the text. PassthroughProtocol::OpenaiResponses => match v.get("input") { Some(serde_json::Value::String(t)) => t.clone(), - Some(serde_json::Value::Array(items)) => { - let joined = items - .iter() - .filter_map(|i| i.get("content").map(content_text)) - .filter(|t| !t.is_empty()) - .collect::>() - .join("\n"); - if joined.is_empty() { - raw() - } else { - joined - } - } - _ => raw(), + Some(serde_json::Value::Array(items)) => items + .iter() + .filter_map(|i| i.get("content").map(content_text)) + .filter(|t| !t.is_empty()) + .collect::>() + .join("\n"), + _ => String::new(), }, PassthroughProtocol::OpenaiCompletions => { let prompt = v.get("prompt").map(|p| match p { @@ -1055,20 +1112,20 @@ fn request_guardrail_text(protocol: PassthroughProtocol, body: &[u8]) -> String other => content_text(other), }); let suffix = v.get("suffix").and_then(|s| s.as_str()); - match (prompt, suffix) { - (None, None) => raw(), - (p, s) => { - let mut out = p.unwrap_or_default(); - if let Some(s) = s { - if !out.is_empty() { - out.push('\n'); - } - out.push_str(s); - } - out + let mut out = prompt.unwrap_or_default(); + if let Some(s) = suffix { + if !out.is_empty() { + out.push('\n'); } + out.push_str(s); } + out } + }; + if extracted.is_empty() { + raw() + } else { + extracted } } @@ -2250,6 +2307,62 @@ mod tests { request_guardrail_text(PassthroughProtocol::OpenaiChat, not_chat), r#"{"input":"x"}"# ); + // A detected envelope whose items carry no text ALSO degrades to + // the raw body — detection must never scan less than raw would. + let empty_chat = br#"{"messages":[{"role":"tool","tool_call_id":"1"}]}"#; + assert_eq!( + request_guardrail_text(PassthroughProtocol::OpenaiChat, empty_chat), + String::from_utf8_lossy(empty_chat) + ); + } + + #[test] + fn detect_protocol_from_request_envelope() { + // The real Copilot CLI surface, one shape per endpoint family. + let cases: [(&[u8], PassthroughProtocol); 8] = [ + // Chat: `messages` array. + ( + br#"{"model":"m","messages":[{"role":"user","content":"hi"}]}"#, + PassthroughProtocol::OpenaiChat, + ), + // Responses API: `input` as items or a bare string. + ( + br#"{"model":"m","input":[{"role":"user","content":"hi"}]}"#, + PassthroughProtocol::OpenaiResponses, + ), + ( + br#"{"model":"m","input":"hi"}"#, + PassthroughProtocol::OpenaiResponses, + ), + // FIM / legacy completions: `prompt`. + ( + br#"{"prompt":"def f(","suffix":"return"}"#, + PassthroughProtocol::OpenaiCompletions, + ), + // MCP JSON-RPC relays as raw. + ( + br#"{"jsonrpc":"2.0","id":1,"method":"tools/list"}"#, + PassthroughProtocol::Raw, + ), + // Plain REST / unrecognized JSON relays as raw. + (br#"{"ref":"main","inputs":{}}"#, PassthroughProtocol::Raw), + // Carrier keys of the wrong TYPE stay raw: only the API's own + // shape (array/string) counts as that envelope. + ( + br#"{"messages":"not-an-array"}"#, + PassthroughProtocol::Raw, + ), + // Non-JSON / empty (GET) bodies are raw. + (b"", PassthroughProtocol::Raw), + ]; + for (body, want) in cases { + assert_eq!( + detect_protocol(body), + want, + "body {:?}", + String::from_utf8_lossy(body) + ); + } } #[test] diff --git a/schemas/resources/passthrough_route.schema.json b/schemas/resources/passthrough_route.schema.json index 0223fdb6a..1b319d815 100644 --- a/schemas/resources/passthrough_route.schema.json +++ b/schemas/resources/passthrough_route.schema.json @@ -437,43 +437,6 @@ "type": "string" } ] - }, - "PassthroughProtocol": { - "description": "Body-shape hint for a passthrough route.", - "oneOf": [ - { - "description": "No parsing: bodies are opaque byte blobs (guardrails scan them as one lossy-UTF-8 text).", - "enum": [ - "raw" - ], - "title": "Opaque body", - "type": "string" - }, - { - "description": "OpenAI-compatible chat envelope (`messages`, streamed `choices[].delta.content`, final-chunk / response `usage`).", - "enum": [ - "openai_chat" - ], - "title": "OpenAI-compatible chat", - "type": "string" - }, - { - "description": "OpenAI-compatible legacy completions / FIM envelope (`prompt` [+ `suffix`], streamed `choices[].text`, `usage`).", - "enum": [ - "openai_completions" - ], - "title": "OpenAI-compatible completions / FIM", - "type": "string" - }, - { - "description": "OpenAI Responses API envelope: `input` items on the request, `output` items on the response, `response.output_text.delta` events while streaming, and `usage` in the `input_tokens`/`output_tokens` spelling — carried on the terminal `response.completed` event when the response streams.", - "enum": [ - "openai_responses" - ], - "title": "OpenAI Responses API", - "type": "string" - } - ] } }, "properties": { @@ -559,15 +522,6 @@ "description": "Derive the target from the request's own `Host` header (`https://`), for forward-proxy routes that fan one route out over several official hosts. Only legal when `hosts` is set — the matched allowlist is what makes the derived target non-attacker- controlled (SSRF guard).", "type": "boolean" }, - "protocol": { - "allOf": [ - { - "$ref": "#/definitions/PassthroughProtocol" - } - ], - "default": "raw", - "description": "Body-shape hint for auditing, guardrails and usage extraction. Parsing is best-effort: a body that does not match the declared shape degrades to `raw` handling, it is never rejected for shape." - }, "provider_key_id": { "description": "ProviderKey whose secret is injected upstream when `credential_mode` is `inject` (per-provider auth shape: `x-api-key` + `anthropic-version` for Anthropic, `Authorization: Bearer` otherwise; its `strip_headers` and TLS settings apply). Required for `inject`; forbidden for `forward_client`.", "minLength": 1, @@ -586,11 +540,6 @@ "null" ] }, - "streaming": { - "default": true, - "description": "Relay `text/event-stream` upstream responses incrementally. When `false` streaming responses are fully buffered like any other body.", - "type": "boolean" - }, "target_url": { "description": "Explicit upstream base URL, e.g. `https://api.openai.com`. The matched request's remainder path and query are appended. Exactly one of `target_url` / `preserve_host` must be configured.", "minLength": 1, @@ -600,7 +549,7 @@ ] }, "timeout_ms": { - "description": "Maximum time, in milliseconds, for a non-streaming upstream exchange. On a streaming route it bounds the response-header phase and any non-SSE body read, but never a healthy SSE relay (which ends with the upstream stream or the client hanging up). When omitted, the gateway default request timeout applies to non-streaming exchanges only.", + "description": "Maximum time, in milliseconds, for the upstream exchange. Bounds the response-header phase and any non-SSE body read, but never a healthy SSE relay (which ends with the upstream stream or the client hanging up). When omitted, the gateway default request timeout applies the same way.", "format": "uint64", "minimum": 1.0, "type": [ diff --git a/tests/e2e/src/cases/passthrough-route-e2e.test.ts b/tests/e2e/src/cases/passthrough-route-e2e.test.ts index 7e4af06c5..ddcc3ca89 100644 --- a/tests/e2e/src/cases/passthrough-route-e2e.test.ts +++ b/tests/e2e/src/cases/passthrough-route-e2e.test.ts @@ -10,6 +10,7 @@ import { type OpenAiUpstream, type SpawnedApp, } from "../harness/index.js"; +import { startMockOtlp, type MockOtlp } from "../harness/otlp-mock.js"; // E2E: explicit PassthroughRoute resources — the successor of the removed // implicit `/passthrough/{provider}/*rest` tunnel. A route binds a gateway @@ -36,6 +37,10 @@ import { // gated by `source_cidrs` (real TCP, so 127.0.0.1 resolves). // 6. SSE relay: a streaming upstream is forwarded as SSE with the // frames intact. +// 7. Envelope auto-detection: a route has NO protocol/streaming +// configuration — usage extraction follows the request body's own +// envelope (chat buffered, Responses streamed) and a non-LLM +// exchange is never probed for phantom usage. const CALLER_PLAINTEXT = "sk-ptr-e2e-caller"; const CALLER_KEY_HASH = createHash("sha256") @@ -47,6 +52,7 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone let seed: SeedClient | undefined; let etcdReachable = false; const upstreams: OpenAiUpstream[] = []; + const otlps: MockOtlp[] = []; beforeAll(async () => { const etcd = new EtcdClient(); @@ -65,6 +71,7 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone afterAll(async () => { await app?.exit(); await Promise.all(upstreams.map((u) => u.close())); + await Promise.all(otlps.map((o) => o.close())); }); test("inject route on the legacy prefix: /v1 dedup, verbatim body, Bearer injection", async (ctx) => { @@ -451,7 +458,6 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone path_prefix: "/sse-tunnel", target_url: upstream.baseUrl, provider_key_id: pk.id, - protocol: "openai_chat", }); const headers = { @@ -462,7 +468,11 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone fetch(`${app!.proxyUrl}/sse-tunnel/chat/completions`, { method: "POST", headers, - body: JSON.stringify({ model: "gpt-4o", stream: true }), + body: JSON.stringify({ + model: "gpt-4o", + messages: [{ role: "user", content: "hi" }], + stream: true, + }), }); await waitConfigPropagation(async () => { @@ -487,4 +497,137 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone expect(text).toContain('"content":"lo"'); expect(text).toContain("[DONE]"); }); + + test("envelope auto-detection: usage follows the request body, never the config", async (ctx) => { + if (!etcdReachable || !app || !seed) { + ctx.skip(); + return; + } + + // A route carries NO protocol or streaming configuration — the + // envelope is detected per exchange from the request body's top-level + // keys. Three exchanges through three identical-config routes: + // + // - chat-shaped request, buffered response → `usage` extracted + // - Responses-shaped request, SSE response → nested usage on the + // terminal `response.completed` event extracted (the shape the + // GitHub Copilot CLI streams) + // - JSON-RPC request (MCP) → raw: a usage-looking + // object in the response is NOT probed (no phantom tokens) + const otlp = await startMockOtlp(); + otlps.push(otlp); + await seed.createObservabilityExporter({ + name: "ptr-auto-otlp", + enabled: true, + kind: "otlp_http", + endpoint: otlp.url, + }); + + const chatUpstream = await startOpenAiUpstream({ + nonStreamBody: { + id: "chatcmpl-auto", + choices: [{ message: { role: "assistant", content: "ok" } }], + usage: { prompt_tokens: 7, completion_tokens: 3 }, + }, + }); + const responsesUpstream = await startOpenAiUpstream({ + streamEvents: [ + JSON.stringify({ type: "response.output_text.delta", delta: "he" }), + JSON.stringify({ type: "response.output_text.delta", delta: "y" }), + JSON.stringify({ + type: "response.completed", + response: { usage: { input_tokens: 11, output_tokens: 4 } }, + }), + ], + }); + const rpcUpstream = await startOpenAiUpstream({ + nonStreamBody: { + jsonrpc: "2.0", + id: 1, + result: { usage: { prompt_tokens: 99, completion_tokens: 99 } }, + }, + }); + upstreams.push(chatUpstream, responsesUpstream, rpcUpstream); + + const pk = await seed.createProviderKey({ + display_name: "ptr-auto-pk", + secret: "sk-mock", + api_base: "http://unused-on-routes", + }); + for (const [name, prefix, upstream] of [ + ["ptr-auto-chat", "/auto-chat", chatUpstream], + ["ptr-auto-resp", "/auto-resp", responsesUpstream], + ["ptr-auto-rpc", "/auto-rpc", rpcUpstream], + ] as const) { + await seed.createPassthroughRoute({ + name, + path_prefix: prefix, + target_url: upstream.baseUrl, + provider_key_id: pk.id, + }); + } + + const headers = { + authorization: `Bearer ${CALLER_PLAINTEXT}`, + "content-type": "application/json", + }; + const post = (path: string, body: unknown) => + fetch(`${app!.proxyUrl}${path}`, { + method: "POST", + headers, + body: JSON.stringify(body), + }); + + await waitConfigPropagation(async () => { + try { + const r = await post("/auto-chat/chat/completions", { + model: "m", + messages: [{ role: "user", content: "probe" }], + }); + await r.text(); + return r.status === 200; + } catch { + return false; + } + }); + + const responsesRes = await post("/auto-resp/responses", { + model: "m", + input: [{ role: "user", content: "hi" }], + stream: true, + }); + expect(responsesRes.status).toBe(200); + await responsesRes.text(); + const rpcRes = await post("/auto-rpc/mcp", { + jsonrpc: "2.0", + id: 1, + method: "tools/list", + }); + expect(rpcRes.status).toBe(200); + await rpcRes.text(); + + const spanFor = async (route: string) => { + const deadline = Date.now() + 10_000; + while (Date.now() < deadline) { + const hit = otlp.spans.find( + (s) => s.attributes["aisix.passthrough.route_name"] === route, + ); + if (hit) return hit; + await new Promise((r) => setTimeout(r, 50)); + } + throw new Error(`no OTLP span for route ${route}`); + }; + + const chatSpan = await spanFor("ptr-auto-chat"); + expect(chatSpan.attributes["gen_ai.usage.input_tokens"]).toBe(7); + expect(chatSpan.attributes["gen_ai.usage.output_tokens"]).toBe(3); + + const respSpan = await spanFor("ptr-auto-resp"); + expect(respSpan.attributes["gen_ai.usage.input_tokens"]).toBe(11); + expect(respSpan.attributes["gen_ai.usage.output_tokens"]).toBe(4); + + const rpcSpan = await spanFor("ptr-auto-rpc"); + expect(rpcSpan.attributes["gen_ai.usage.input_tokens"] ?? 0).toBe(0); + expect(rpcSpan.attributes["gen_ai.usage.output_tokens"] ?? 0).toBe(0); + }); }); From 793517f58ef1bfb9654bc122f07d54bcb45906f6 Mon Sep 17 00:00:00 2001 From: Jarvis Date: Tue, 18 Aug 2026 13:02:14 +0800 Subject: [PATCH 2/3] style: rustfmt --- crates/aisix-core/src/lib.rs | 7 +++---- crates/aisix-proxy/src/passthrough_route.rs | 5 +---- 2 files changed, 4 insertions(+), 8 deletions(-) diff --git a/crates/aisix-core/src/lib.rs b/crates/aisix-core/src/lib.rs index d67c5cab7..5ca08d34c 100644 --- a/crates/aisix-core/src/lib.rs +++ b/crates/aisix-core/src/lib.rs @@ -50,10 +50,9 @@ pub use models::{ GuardrailMonitorHit, KeywordConfig, KeywordPattern, McpAuthType, McpProtocolVersion, McpRateLimit, McpServer, McpServerType, McpTransport, Model, ObservabilityExporter, ParamConstraints, PassthroughAuthMode, PassthroughCredentialMode, PassthroughRoute, - PolicyScope, PolicyWindow, ProviderKey, RateLimit, RateLimitPolicy, - RequestOverrides, ResponseOverrides, Routing, RoutingStrategy, RoutingTarget, SchemaError, - StreamDoneMarker, TelemetryKind, TelemetryTags, WhenAllUnavailablePolicy, - DEFAULT_COOLDOWN_TRIGGER_STATUSES, + PolicyScope, PolicyWindow, ProviderKey, RateLimit, RateLimitPolicy, RequestOverrides, + ResponseOverrides, Routing, RoutingStrategy, RoutingTarget, SchemaError, StreamDoneMarker, + TelemetryKind, TelemetryTags, WhenAllUnavailablePolicy, DEFAULT_COOLDOWN_TRIGGER_STATUSES, }; pub use resource::{Resource, ResourceEntry}; pub use snapshot::{ResourceTable, SnapshotHandle}; diff --git a/crates/aisix-proxy/src/passthrough_route.rs b/crates/aisix-proxy/src/passthrough_route.rs index a6b6c6d0d..963663382 100644 --- a/crates/aisix-proxy/src/passthrough_route.rs +++ b/crates/aisix-proxy/src/passthrough_route.rs @@ -2348,10 +2348,7 @@ mod tests { (br#"{"ref":"main","inputs":{}}"#, PassthroughProtocol::Raw), // Carrier keys of the wrong TYPE stay raw: only the API's own // shape (array/string) counts as that envelope. - ( - br#"{"messages":"not-an-array"}"#, - PassthroughProtocol::Raw, - ), + (br#"{"messages":"not-an-array"}"#, PassthroughProtocol::Raw), // Non-JSON / empty (GET) bodies are raw. (b"", PassthroughProtocol::Raw), ]; From a8a531c02ca297e1d47973fe8778a4eb0b2c1a3b Mon Sep 17 00:00:00 2001 From: Jarvis Date: Tue, 18 Aug 2026 13:11:36 +0800 Subject: [PATCH 3/3] review: heartbeat + request-span on the SSE relay; sentinel readiness gate in e2e The relay stream now composes with_heartbeat + in_request_span like every typed streaming endpoint (pre-existing gap from the redesign, surfaced on review). The envelope-detection e2e gates readiness on a sentinel route seeded last instead of probing an exchange the test asserts on, and fires the chat measurement explicitly. --- crates/aisix-proxy/src/passthrough_route.rs | 10 ++++++++- .../src/cases/passthrough-route-e2e.test.ts | 22 +++++++++++++++---- 2 files changed, 27 insertions(+), 5 deletions(-) diff --git a/crates/aisix-proxy/src/passthrough_route.rs b/crates/aisix-proxy/src/passthrough_route.rs index 963663382..b1690612e 100644 --- a/crates/aisix-proxy/src/passthrough_route.rs +++ b/crates/aisix-proxy/src/passthrough_route.rs @@ -1623,9 +1623,17 @@ fn stream_response( telemetry.emit(); }; + // Re-attach the request span (the body is polled after the request-id + // middleware returns, so end-of-stream telemetry would otherwise log + // without a request_id) and heartbeat silence gaps — this branch is + // SSE-only, where a comment frame is protocol-legal and identical to + // what the typed endpoints emit; relayed frames are untouched. let mut response = Response::builder() .status(status) - .body(Body::from_stream(stream)) + .body(Body::from_stream(crate::sse_keepalive::with_heartbeat( + crate::request_id::in_request_span(stream), + crate::sse_keepalive::interval(), + ))) .unwrap(); copy_safe_headers(&resp_headers, response.headers_mut()); // The relay re-chunks the body; a stale upstream length must not ride diff --git a/tests/e2e/src/cases/passthrough-route-e2e.test.ts b/tests/e2e/src/cases/passthrough-route-e2e.test.ts index ddcc3ca89..e267bb82b 100644 --- a/tests/e2e/src/cases/passthrough-route-e2e.test.ts +++ b/tests/e2e/src/cases/passthrough-route-e2e.test.ts @@ -567,6 +567,17 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone }); } + // Readiness sentinel, seeded LAST: the snapshot watch applies etcd + // revisions in order, so this route serving proves every resource + // seeded before it (exporter, the three routes under test) is live — + // without the probe exercising any exchange the test asserts on. + await seed.createPassthroughRoute({ + name: "ptr-auto-ready", + path_prefix: "/auto-ready", + target_url: rpcUpstream.baseUrl, + provider_key_id: pk.id, + }); + const headers = { authorization: `Bearer ${CALLER_PLAINTEXT}`, "content-type": "application/json", @@ -580,10 +591,7 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone await waitConfigPropagation(async () => { try { - const r = await post("/auto-chat/chat/completions", { - model: "m", - messages: [{ role: "user", content: "probe" }], - }); + const r = await post("/auto-ready/ping", {}); await r.text(); return r.status === 200; } catch { @@ -591,6 +599,12 @@ describe("passthrough-route e2e: explicit routes, BYO credentials, 410 tombstone } }); + const chatRes = await post("/auto-chat/chat/completions", { + model: "m", + messages: [{ role: "user", content: "hi" }], + }); + expect(chatRes.status).toBe(200); + await chatRes.text(); const responsesRes = await post("/auto-resp/responses", { model: "m", input: [{ role: "user", content: "hi" }],