diff --git a/.github/ISSUE_TEMPLATE/bug_report.yml b/.github/ISSUE_TEMPLATE/bug_report.yml index 1aa2ac130..2158aa77e 100644 --- a/.github/ISSUE_TEMPLATE/bug_report.yml +++ b/.github/ISSUE_TEMPLATE/bug_report.yml @@ -94,8 +94,17 @@ body: attributes: label: 请求与响应 description: | - 接口问题请提供脱敏后的 cURL 或最小请求体、HTTP 状态码、响应片段，以及请求 ID / 发生时间。 - 如需详细报文，可按部署文档开启 `host.logging.request_dump`，仅摘取定位所需片段并脱敏，不要直接上传完整转储。 + 接口问题请提供脱敏后的最小请求、网关实际状态/响应片段和客户端终端显示，区分两者。 + 优先附管理端默认诊断包并检查其缺失说明；原始错误与日志片段需另行审阅脱敏。 + 先按部署文档的“请求错误排查”检查关联 ID、错误详情与 application 日志，不默认开启完整 dump。 + placeholder: | + 发生时间与时区： + 传输（HTTP JSON / SSE / WebSocket）： + 网关关联 ID（x-gateway-request-id / 配置的入口头）： + 上游关联 ID（x-request-id / x-oai-request-id，注明来源）： + 网关实际状态与响应片段： + 客户端终端显示： + 诊断包或无法取得的信息： render: text - type: textarea diff --git a/backend/crates/gateway-api/src/openai/endpoint.rs b/backend/crates/gateway-api/src/openai/endpoint.rs index bc6191840..81390c5fb 100644 --- a/backend/crates/gateway-api/src/openai/endpoint.rs +++ b/backend/crates/gateway-api/src/openai/endpoint.rs @@ -5,7 +5,7 @@ use axum::{ http::{HeaderName, HeaderValue, StatusCode, header::CONTENT_TYPE}, response::{IntoResponse, Response}, }; -use gateway_core::engine::execution::StartedExecution; +use gateway_core::engine::execution::{ExecutionSession, StartedExecution}; use gateway_core::event::{ProviderEvent, ProviderResponseHeader}; use super::{ @@ -15,7 +15,12 @@ use super::{ /// 收集一个 Provider 原生 JSON 响应，并在 Core commit 后按原始 bytes 交付。 pub(super) async fn collect_raw_json_response(started: StartedExecution) -> Response { - let mut execution = PendingExecution::new(started.session); + let response = collect_raw_json_session(started.session).await; + super::with_model_request_id(response, &started.request_id) +} + +async fn collect_raw_json_session(session: Box) -> Response { + let mut execution = PendingExecution::new(session); let Some(session) = execution.session_mut() else { return invalid_upstream_response(); }; diff --git a/backend/crates/gateway-api/src/openai/error.rs b/backend/crates/gateway-api/src/openai/error.rs index ee53f8003..9ed000288 100644 --- a/backend/crates/gateway-api/src/openai/error.rs +++ b/backend/crates/gateway-api/src/openai/error.rs @@ -190,7 +190,15 @@ pub fn engine_error_response(error: &EngineError) -> Response { } return response; } - gateway_error_response(&gateway_error_from_engine(error)) + let mut response = gateway_error_response(&gateway_error_from_engine(error)); + // 正文未完整取得时不能透传残缺响应，但已确认的上游关联 ID 仍应交付。 + if let EngineError::Provider(error) = error + && let Some(request_id) = error.upstream_request_id() + && let Ok(value) = HeaderValue::from_str(request_id.as_str()) + { + response.headers_mut().insert("x-request-id", value); + } + response } /// Gateway 错误稳定映射，供 HTTP、SSE 和 WebSocket 共用。 diff --git a/backend/crates/gateway-api/src/openai/mod.rs b/backend/crates/gateway-api/src/openai/mod.rs index 1ba553eb7..638ce3c2b 100644 --- a/backend/crates/gateway-api/src/openai/mod.rs +++ b/backend/crates/gateway-api/src/openai/mod.rs @@ -9,3 +9,28 @@ pub mod responses; pub mod router; pub mod search; pub(crate) mod service; + +/// 客户端可读的 ID 必须能按现有请求记录检索；不以无关入口 ID 补位。 +fn with_model_request_id( + mut response: axum::response::Response, + request_id: &gateway_core::engine::ModelRequestId, +) -> axum::response::Response { + use axum::http::HeaderValue; + + let Ok(gateway_id) = HeaderValue::from_str(request_id.as_str()) else { + return response; + }; + let headers = response.headers_mut(); + let usable = |value: &HeaderValue| value.to_str().is_ok_and(|id| !id.trim().is_empty()); + if !headers.get("x-request-id").is_some_and(usable) { + // 只有别名时沿用同一个上游值，避免外层 middleware 再填入入口 ID。 + let client_id = headers + .get("x-oai-request-id") + .filter(|value| usable(value)) + .unwrap_or(&gateway_id) + .clone(); + headers.insert("x-request-id", client_id); + } + headers.insert("x-gateway-request-id", gateway_id); + response +} diff --git a/backend/crates/gateway-api/src/openai/responses/http.rs b/backend/crates/gateway-api/src/openai/responses/http.rs index d70089233..5cd23029e 100644 --- a/backend/crates/gateway-api/src/openai/responses/http.rs +++ b/backend/crates/gateway-api/src/openai/responses/http.rs @@ -93,20 +93,17 @@ pub(crate) async fn responses( "ingressRequestId": ingress_id.as_ref().and_then(|Extension(id)| id.header_value().to_str().ok()), }), headers.iter().map(|(name, value)| (name.as_str(), value.as_bytes()))); trace.capture("client.request.body", &body); - let request_id_header = HeaderValue::from_str(started.request_id.as_str()).ok(); + let request_id = started.request_id; let StartedExecution { stream, session, .. } = started; - let mut response = if stream { + let response = if stream { stream_execution_response(session, connection_guard).await } else { drop(connection_guard); collect_execution_response(session).await }; - if let Some(id) = request_id_header { - response.headers_mut().insert("x-gateway-request-id", id); - } - response + crate::openai::with_model_request_id(response, &request_id) } /// 从 socket 与标准转发头提取旧 Usage 页面使用的诊断事实。 @@ -271,7 +268,26 @@ fn engine_error_response_with_headers( error: &EngineError, response_headers: &[ProviderResponseHeader], ) -> Response { - apply_response_headers(engine_error_response(error), response_headers) + let response = engine_error_response(error); + let failure_request_ids: Vec<_> = ["x-request-id", "x-oai-request-id"] + .into_iter() + .flat_map(|name| { + response + .headers() + .get_all(name) + .iter() + .cloned() + .map(move |value| (name, value)) + }) + .collect(); + // 失败后仍须交付已采集的 turn state；只隔离 opening 身份，不丢弃会话状态。 + let mut response = apply_response_headers(response, response_headers); + response.headers_mut().remove("x-request-id"); + response.headers_mut().remove("x-oai-request-id"); + for (name, value) in failure_request_ids { + response.headers_mut().append(name, value); + } + response } /// 编码首个 SSE frame 后提交下游，再持续驱动同一执行会话。 diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs b/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs index c3fad5a1a..f472cafc7 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/forward.rs @@ -14,7 +14,7 @@ use crate::openai::error::{gateway_error_contract, gateway_error_from_engine}; use super::{ super::{DecodedResponsesRequest, OpenAiResponsesEncoder, PendingExecution, ProtocolErrorBody}, connection::{FramePhase, ResponsesWebSocketConnection, WriteContext}, - protocol::{error_event, response_metadata_event}, + protocol::{error_event, initial_engine_error_event, response_metadata_event}, }; #[derive(Clone, Copy, PartialEq, Eq)] @@ -273,8 +273,12 @@ async fn send_initial_engine_error( { return ForwardOutcome::Disconnect; } - let error = gateway_error_from_engine(error); - send_gateway_error(connection, &error, request_id).await + send_error_message( + connection, + initial_engine_error_event(error, request_id, &response_headers), + request_id, + ) + .await } fn commit_connection_replay( @@ -393,10 +397,31 @@ async fn send_error( message: &str, param: Option<&str>, request_id: &Arc, +) -> ForwardOutcome { + send_error_message( + connection, + error_event( + status, + error_type, + code, + message, + param, + Some(request_id), + serde_json::Map::new(), + ), + request_id, + ) + .await +} + +async fn send_error_message( + connection: &mut ResponsesWebSocketConnection, + message: String, + request_id: &Arc, ) -> ForwardOutcome { if connection .send_text( - error_event(status, error_type, code, message, param, Some(request_id)), + message, WriteContext::request(request_id, FramePhase::Error), ) .await diff --git a/backend/crates/gateway-api/src/openai/responses/websocket/protocol.rs b/backend/crates/gateway-api/src/openai/responses/websocket/protocol.rs index 97243467e..e06e7efce 100644 --- a/backend/crates/gateway-api/src/openai/responses/websocket/protocol.rs +++ b/backend/crates/gateway-api/src/openai/responses/websocket/protocol.rs @@ -1,10 +1,13 @@ //! Responses WebSocket 入站请求与下行事件的纯协议映射。 -use axum::http::{HeaderName, StatusCode}; +use axum::http::{HeaderName, HeaderValue, StatusCode}; +use gateway_core::engine::EngineError; use gateway_core::event::ProviderResponseHeader; use serde_json::{Map, Value, json}; use thiserror::Error; +use crate::openai::error::{gateway_error_contract, gateway_error_from_engine}; + use super::super::{ DecodedResponsesRequest, ProtocolErrorBody, RequestDecodeError, request::{OpenAiRequestHeaders, RequestDecodeSource}, @@ -91,21 +94,7 @@ pub(super) fn response_metadata_event( request_id: &str, response_headers: &[ProviderResponseHeader], ) -> String { - let connection_options = super::super::response_connection_options(response_headers); - let mut headers = Map::new(); - for header in response_headers { - if !super::super::response_header_is_forwardable(header.name(), &connection_options) { - continue; - } - let Ok(name) = HeaderName::from_bytes(header.name().as_bytes()) else { - continue; - }; - let Ok(value) = std::str::from_utf8(header.value()) else { - continue; - }; - // 官方 response.metadata 使用 string map，无法表达同名多值；后值保持既有覆盖语义。 - headers.insert(name.as_str().to_owned(), Value::String(value.to_owned())); - } + let mut headers = response_event_headers(response_headers); headers .entry("x-request-id".to_owned()) .or_insert_with(|| Value::String(request_id.to_owned())); @@ -116,6 +105,84 @@ pub(super) fn response_metadata_event( .to_string() } +fn response_event_headers(response_headers: &[ProviderResponseHeader]) -> Map { + let connection_options = super::super::response_connection_options(response_headers); + let mut headers = Map::new(); + for header in response_headers { + if !super::super::response_header_is_forwardable(header.name(), &connection_options) { + continue; + } + let Ok(name) = HeaderName::from_bytes(header.name().as_bytes()) else { + continue; + }; + let Ok(http_value) = HeaderValue::from_bytes(header.value()) else { + continue; + }; + let Ok(value) = std::str::from_utf8(header.value()) else { + continue; + }; + if matches!(name.as_str(), "x-request-id" | "x-oai-request-id") + && (value.trim().is_empty() || http_value.to_str().is_err()) + { + continue; + } + // 官方 WS 事件使用 string map，无法表达同名多值；后值保持既有覆盖语义。 + headers.insert(name.as_str().to_owned(), Value::String(value.to_owned())); + } + headers +} + +pub(super) fn initial_engine_error_event( + error: &EngineError, + request_id: &str, + response_headers: &[ProviderResponseHeader], +) -> String { + let gateway = gateway_error_from_engine(error); + let (default_status, default_type, default_code) = gateway_error_contract(gateway.kind()); + let provider = match error { + EngineError::Provider(provider) => Some(provider), + _ => None, + }; + let upstream = provider.and_then(|error| error.client_visible_upstream_response()); + // 最终失败响应优先于会话快照，避免混入先前 attempt 或 opening 的关联头。 + let mut headers = + response_event_headers(upstream.map_or(response_headers, |response| response.headers())); + if upstream.is_none() { + // 会话 opening 身份不证明当前失败属于该请求；仅保留其他允许的会话头。 + headers.remove("x-request-id"); + headers.remove("x-oai-request-id"); + } + let status = upstream + .map(|response| response.status()) + .or_else(|| provider.and_then(|error| error.upstream_status())) + .and_then(|value| StatusCode::from_u16(value).ok()) + .filter(|value| !value.is_success() && !value.is_informational()) + .unwrap_or(default_status); + // 无原响应关联头时使用 Provider 的失败事实；不能让旧会话 ID 遮住当前失败。 + if !has_request_id(&headers) + && let Some(request_id) = provider.and_then(|error| error.upstream_request_id()) + && let Ok(value) = HeaderValue::from_str(request_id.as_str()) + && let Ok(value) = value.to_str() + && !value.trim().is_empty() + { + headers.insert("x-request-id".to_owned(), Value::String(value.to_owned())); + } + // 只桥接 Provider 已提取的结构化错误；原始 HTML/文本正文不成为安全 fallback 的 message。 + error_event( + status, + gateway.client_error_type().unwrap_or(default_type), + gateway.client_error_code().unwrap_or(default_code), + gateway.client_message(), + None, + Some(request_id), + headers, + ) +} + +fn has_request_id(headers: &Map) -> bool { + headers.contains_key("x-request-id") || headers.contains_key("x-oai-request-id") +} + pub(super) fn error_event( status: StatusCode, error_type: &str, @@ -123,6 +190,7 @@ pub(super) fn error_event( message: &str, param: Option<&str>, request_id: Option<&str>, + mut headers: Map, ) -> String { let mut error = Map::new(); error.insert("type".to_owned(), Value::String(error_type.to_owned())); @@ -136,11 +204,26 @@ pub(super) fn error_event( event.insert("status".to_owned(), Value::Number(status.as_u16().into())); event.insert("error".to_owned(), Value::Object(error)); if let Some(request_id) = request_id { + // Codex 从本 error 事件的 headers 读 ID，独立 metadata 和顶层 request_id 不能替代。 + // 网关身份独立保存；有上游关联头时不冒充或覆盖上游身份。 + if !has_request_id(&headers) { + headers.insert( + "x-request-id".to_owned(), + Value::String(request_id.to_owned()), + ); + } + headers.insert( + "x-gateway-request-id".to_owned(), + Value::String(request_id.to_owned()), + ); event.insert( "request_id".to_owned(), Value::String(request_id.to_owned()), ); } + if !headers.is_empty() { + event.insert("headers".to_owned(), Value::Object(headers)); + } Value::Object(event).to_string() } @@ -152,5 +235,6 @@ pub(super) fn connection_limit_event() -> String { "Responses websocket connection limit reached (60 minutes). Create a new websocket connection to continue.", None, None, + Map::new(), ) } diff --git a/backend/crates/gateway-api/tests/architecture.rs b/backend/crates/gateway-api/tests/architecture.rs index 3a2927e49..3dfe53841 100644 --- a/backend/crates/gateway-api/tests/architecture.rs +++ b/backend/crates/gateway-api/tests/architecture.rs @@ -114,6 +114,7 @@ fn test_tree_should_match_frozen_rust_mirror() { "tests/openai/responses/mod.rs", "tests/openai/responses/request.rs", "tests/openai/responses/websocket/connection.rs", + "tests/openai/responses/websocket/forward.rs", "tests/openai/responses/websocket/mod.rs", "tests/openai/responses/websocket/protocol.rs", "tests/openai/router.rs", diff --git a/backend/crates/gateway-api/tests/openai/images/mod.rs b/backend/crates/gateway-api/tests/openai/images/mod.rs index 71797d09f..be14ecd8b 100644 --- a/backend/crates/gateway-api/tests/openai/images/mod.rs +++ b/backend/crates/gateway-api/tests/openai/images/mod.rs @@ -258,6 +258,7 @@ async fn image_routes_should_not_decode_bodies_and_should_preserve_both_directio .oneshot( Request::post(*path) .header(AUTHORIZATION, "Bearer sk_images_test") + .header("x-request-id", "caller-image-request") .header("content-type", "application/json") .header("x-codex-image-turn-id", "turn_image_route") .header("session-id", "root-image-session") @@ -268,6 +269,11 @@ async fn image_routes_should_not_decode_bodies_and_should_preserve_both_directio .await .expect("image response"); assert_eq!(response.status(), StatusCode::CREATED); + assert_eq!(response.headers()["x-request-id"], "req_images_test"); + assert_eq!( + response.headers()["x-gateway-request-id"], + "req_images_test" + ); assert_eq!( response.headers().get("x-image-rate-limit"), Some(&"42".parse().expect("header value")) @@ -315,6 +321,11 @@ async fn image_route_should_return_the_exact_upstream_error_response() { .expect("image response"); assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); + assert_eq!(response.headers()["x-request-id"], "req_images_test"); + assert_eq!( + response.headers()["x-gateway-request-id"], + "req_images_test" + ); assert_eq!( response.headers().get("content-type"), Some(&"application/problem+json".parse().expect("content type")) diff --git a/backend/crates/gateway-api/tests/openai/responses/http.rs b/backend/crates/gateway-api/tests/openai/responses/http.rs index 83ac98023..87967e0e6 100644 --- a/backend/crates/gateway-api/tests/openai/responses/http.rs +++ b/backend/crates/gateway-api/tests/openai/responses/http.rs @@ -22,7 +22,7 @@ use gateway_core::engine::execution::{ AuthenticatedClient, ClientAuthenticationError, ExecutionService, ExecutionSession, StartExecution, StartProviderExecution, StartedExecution, }; -use gateway_core::engine::{CommitRequirement, CoordinatedEvent, EngineError}; +use gateway_core::engine::{CommitRequirement, CoordinatedEvent, EngineError, ModelRequestId}; use gateway_core::error::{ ClientVisibleUpstreamError, ClientVisibleUpstreamResponse, GatewayError, GatewayErrorKind, ProviderError, ProviderErrorKind, @@ -33,7 +33,7 @@ use gateway_core::event::{ }; use gateway_core::operation::{Operation, OperationKind}; use gateway_core::routing::PublicModelId; -use gateway_core::upstream::UpstreamSendState; +use gateway_core::upstream::{OpaqueUpstreamValue, UpstreamSendState}; use serde_json::{Value, json}; use gateway_api::openai::responses::{collect_execution_response, stream_execution_response}; @@ -1819,3 +1819,223 @@ fn response_status_from_body(body: &[u8]) -> Option { let value: serde_json::Value = serde_json::from_slice(body).ok()?; value.pointer("/error/code")?.as_str().map(str::to_owned) } + +const MODEL_REQUEST_ID: &str = "req_model_correlation"; + +struct SessionExecution { + client: AuthenticatedClient, + session: Mutex>>, +} + +impl ExecutionService for SessionExecution { + fn authenticate(&self, _: &str) -> Result { + Ok(self.client.clone()) + } + + fn public_models(&self, _: &AuthenticatedClient) -> Vec { + Vec::new() + } + + fn contains_public_model(&self, _: &AuthenticatedClient, _: &PublicModelId) -> bool { + true + } + + fn start( + &self, + request: StartExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async move { + Ok(StartedExecution { + request_id: ModelRequestId::new(MODEL_REQUEST_ID).unwrap(), + created_at: std::time::SystemTime::now(), + stream: request.metadata.stream, + session: self.session.lock().unwrap().take().expect("one execution"), + }) + }) + } + + fn start_provider_endpoint( + &self, + _: StartProviderExecution, + ) -> BoxFuture<'_, Result> { + Box::pin(async { panic!("Responses must use model execution") }) + } +} + +async fn response( + provider: &str, + session: FakeSession, + streaming: bool, +) -> axum::response::Response { + let execution = Arc::new(SessionExecution { + client: authenticated_client_for_provider("sk_correlation_test", provider), + session: Mutex::new(Some(Box::new(session))), + }); + api_router(execution) + .await + .oneshot( + Request::post("/v1/responses") + .header(AUTHORIZATION, "Bearer sk_correlation_test") + .header("x-request-id", "caller-chosen-not-a-model-id") + .header("user-agent", "AnotherTerminal/1.0") + .body(Body::from( + json!({ + "model": "model-a", "input": "synthetic", "stream": streaming + }) + .to_string(), + )) + .unwrap(), + ) + .await + .unwrap() +} + +fn headers(values: &[(&str, &'static str)]) -> Vec { + values + .iter() + .map(|(name, value)| { + ProviderResponseHeader::new(*name, Bytes::from_static(value.as_bytes())) + }) + .collect() +} + +fn failed_session(streaming: bool, error: EngineError) -> FakeSession { + let trace = Arc::new(Trace::default()); + if streaming { + FakeSession::streaming(trace, vec![NextStep::Error(error)]) + } else { + FakeSession::buffered(trace, Vec::new()).with_collect_error(error) + } +} + +#[tokio::test] +async fn response_ids_are_provider_and_client_independent() { + let cases: &[(&[(&str, &str)], &str)] = &[ + (&[], MODEL_REQUEST_ID), + (&[("x-request-id", "upstream-primary")], "upstream-primary"), + (&[("x-oai-request-id", "upstream-alias")], "upstream-alias"), + ( + &[ + ("x-request-id", "upstream-primary"), + ("x-oai-request-id", "upstream-alias"), + ], + "upstream-primary", + ), + ( + &[ + ("x-request-id", " "), + ("x-oai-request-id", "upstream-alias"), + ], + "upstream-alias", + ), + (&[("x-request-id", "")], MODEL_REQUEST_ID), + ]; + for provider in ["openai", "xai"] { + for streaming in [false, true] { + for (upstream_headers, expected) in cases { + let trace = Arc::new(Trace::default()); + let session = if streaming { + FakeSession::streaming( + trace, + vec![ + NextStep::Event(delivery( + started(), + CommitRequirement::CommitBeforeDelivery, + )), + NextStep::Event(delivery( + completed(), + CommitRequirement::AlreadyCommitted, + )), + NextStep::FinalizeSuccess, + ], + ) + } else { + FakeSession::buffered(trace, vec![started(), completed()]) + } + .with_response_headers(headers(upstream_headers)); + let response = response(provider, session, streaming).await; + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(response.headers()["x-request-id"], *expected); + assert_eq!(response.headers()["x-gateway-request-id"], MODEL_REQUEST_ID); + assert!( + !to_bytes(response.into_body(), 64 * 1024) + .await + .unwrap() + .is_empty() + ); + } + } + } +} + +#[tokio::test] +async fn failure_ids_prefer_the_actual_error_over_session_headers() { + for provider in ["openai", "xai"] { + for streaming in [false, true] { + for upstream_headers in [vec![], headers(&[("x-oai-request-id", "actual-error")])] { + let expected = if upstream_headers.is_empty() { + MODEL_REQUEST_ID + } else { + "actual-error" + }; + let raw = + Bytes::from_static(br#"{"error":{"message":"synthetic upstream detail"}}"#); + let error = + ProviderError::new(ProviderErrorKind::InvalidRequest, UpstreamSendState::Sent) + .with_client_visible_upstream_response( + ClientVisibleUpstreamResponse::new( + 422, + Some(b"application/json".to_vec()), + raw.clone(), + ) + .with_headers(upstream_headers), + ); + let session = failed_session(streaming, EngineError::Provider(error)) + .with_response_headers(headers(&[("x-request-id", "earlier-session")])); + let response = response(provider, session, streaming).await; + assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); + assert_eq!(response.headers()["x-request-id"], expected); + assert_eq!(response.headers()["x-gateway-request-id"], MODEL_REQUEST_ID); + assert_eq!(to_bytes(response.into_body(), 4096).await.unwrap(), raw); + } + + let error = + ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous) + .with_status(503) + .with_upstream_request_id(OpaqueUpstreamValue::new("body-read-error")); + let session = failed_session(streaming, EngineError::Provider(error)) + .with_response_headers(headers(&[ + ("x-request-id", "earlier-session"), + ("x-oai-request-id", "earlier-alias"), + ])); + let response = response(provider, session, streaming).await; + assert_eq!(response.status(), StatusCode::BAD_GATEWAY); + assert_eq!(response.headers()["x-request-id"], "body-read-error"); + assert_eq!(response.headers().get_all("x-request-id").iter().count(), 1); + assert!(response.headers().get("x-oai-request-id").is_none()); + assert_eq!(response.headers()["x-gateway-request-id"], MODEL_REQUEST_ID); + } + } +} + +#[tokio::test] +async fn local_execution_failure_uses_the_existing_model_id() { + for provider in ["openai", "xai"] { + for streaming in [false, true] { + let session = + failed_session(streaming, EngineError::Deadline).with_response_headers(headers(&[ + ("x-request-id", "opening-primary"), + ("x-oai-request-id", "opening-alias"), + ("x-codex-turn-state", "retained-turn-state"), + ])); + let response = response(provider, session, streaming).await; + assert_eq!(response.headers()["x-request-id"], MODEL_REQUEST_ID); + assert_eq!(response.headers()["x-gateway-request-id"], MODEL_REQUEST_ID); + assert!(response.headers().get("x-oai-request-id").is_none()); + assert_eq!( + response.headers()["x-codex-turn-state"], + "retained-turn-state" + ); + } + } +} diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs new file mode 100644 index 000000000..5101cb11d --- /dev/null +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/forward.rs @@ -0,0 +1,449 @@ +use std::collections::VecDeque; +use std::sync::{Arc, Mutex, atomic::Ordering}; +use std::time::Duration; + +use axum::http::{StatusCode, header::AUTHORIZATION}; +use bytes::Bytes; +use futures::{SinkExt, StreamExt}; +use gateway_core::engine::{CommitRequirement, CoordinatedEvent, EngineError}; +use gateway_core::error::{ + ClientVisibleUpstreamError, ClientVisibleUpstreamResponse, ProviderError, ProviderErrorKind, +}; +use gateway_core::event::{ProtocolWireEvent, ProviderEvent, ProviderResponseHeader}; +use gateway_core::upstream::{OpaqueUpstreamValue, UpstreamSendState}; +use serde_json::{Value, json}; +use tokio::net::TcpStream; +use tokio_tungstenite::{ + MaybeTlsStream, WebSocketStream, connect_async, + tungstenite::{Message, client::IntoClientRequest}, +}; + +use super::{AtomicFailureExecution, AtomicFailureTrace}; +use crate::openai::{api_router, authenticated_client}; + +type TestSocket = WebSocketStream>; + +// 全部使用合成响应事实和真实 API 路由，不读取账号、日志或请求转储。 +struct TestServer(tokio::task::JoinHandle<()>); + +impl Drop for TestServer { + fn drop(&mut self) { + self.0.abort(); + } +} + +async fn connect( + trace: Arc, + response_headers: Vec, +) -> (TestSocket, TestServer) { + let execution = Arc::new(AtomicFailureExecution { + client: authenticated_client("sk_ws_atomic"), + trace, + response_headers, + fail_before_first_event: false, + }); + let app = api_router(execution).await; + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("bind local API"); + let address = listener.local_addr().unwrap(); + let server = TestServer(tokio::spawn(async move { + axum::serve(listener, app).await.expect("serve local API"); + })); + let mut request = format!("ws://{address}/v1/responses") + .into_client_request() + .unwrap(); + request + .headers_mut() + .insert(AUTHORIZATION, "Bearer sk_ws_atomic".parse().unwrap()); + let (socket, response) = connect_async(request).await.expect("upgrade WebSocket"); + assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS); + (socket, server) +} + +async fn send_request(socket: &mut TestSocket) { + socket + .send(Message::Text( + json!({"type": "response.create", "model": "model-a", "input": "synthetic"}) + .to_string() + .into(), + )) + .await + .unwrap(); +} + +async fn next_event(socket: &mut TestSocket) -> Value { + let frame = tokio::time::timeout(Duration::from_secs(2), socket.next()) + .await + .expect("event timeout") + .expect("connection remains open") + .expect("valid frame"); + let Message::Text(text) = frame else { + panic!("expected a JSON text frame, got {frame:?}"); + }; + serde_json::from_str(&text).expect("JSON event") +} + +async fn next_error(socket: &mut TestSocket) -> Value { + let first = next_event(socket).await; + let error = if first["type"] == "response.metadata" { + next_event(socket).await + } else { + first + }; + assert_eq!(error["type"], "error"); + error +} + +async fn assert_no_duplicate_failure(socket: &mut TestSocket) { + // 后续入站帧只会在本轮收敛后处理，比固定等待窗口更可靠地检查没有重复失败终态。 + socket.send(Message::Text("{".into())).await.unwrap(); + let error = next_event(socket).await; + assert_eq!(error["error"]["code"], "invalid_json"); + assert_eq!(error["status"], 400); + assert_eq!(error["headers"]["x-request-id"], error["request_id"]); + assert_eq!( + error["headers"]["x-gateway-request-id"], + error["request_id"] + ); +} + +async fn initial_error(error: EngineError, response_headers: Vec) -> Value { + let trace = Arc::new(AtomicFailureTrace { + initial_errors: Mutex::new(VecDeque::from([error])), + ..AtomicFailureTrace::default() + }); + let (mut socket, _server) = connect(Arc::clone(&trace), response_headers).await; + send_request(&mut socket).await; + let error = next_error(&mut socket).await; + assert!(!trace.committed.load(Ordering::Acquire)); + assert!(trace.finalized.load(Ordering::Acquire)); + assert_eq!(error["request_id"], "req_ws_atomic"); + assert_eq!(error["headers"]["x-gateway-request-id"], "req_ws_atomic"); + assert_no_duplicate_failure(&mut socket).await; + socket.close(None).await.unwrap(); + error +} + +fn header(name: &str, value: &'static [u8]) -> ProviderResponseHeader { + ProviderResponseHeader::new(name, Bytes::from_static(value)) +} + +fn upstream_failure( + status: u16, + body: &'static [u8], + headers: Vec, +) -> ProviderError { + ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::Sent) + .with_status(status) + .with_client_visible_upstream_response( + ClientVisibleUpstreamResponse::new(status, None, Bytes::from_static(body)) + .with_headers(headers), + ) +} + +#[tokio::test] +async fn initial_http_or_opening_failure_preserves_status_and_structured_error() { + // 分类故意都用 Unavailable：不能把收到的 401/429/500 等按网关分类改成 502。 + for status in [302, 400, 401, 403, 429, 500, 502, 503] { + let provider = upstream_failure( + status, + br#"{"error":{"message":"synthetic upstream marker","code":"custom_code","type":"custom_type"}}"#, + vec![header("x-request-id", b"req_current"), header("retry-after", b"7")], + ) + .with_client_visible_upstream_error(ClientVisibleUpstreamError::new( + "synthetic upstream marker", + Some("custom_code".to_owned()), + Some("custom_type".to_owned()), + )); + let error = initial_error(EngineError::Provider(provider), Vec::new()).await; + assert_eq!(error["status"], status); + assert_eq!( + error["error"], + json!({ + "message": "synthetic upstream marker", "code": "custom_code", "type": "custom_type", + }) + ); + assert_eq!(error["headers"]["x-request-id"], "req_current"); + assert_eq!(error["headers"]["retry-after"], "7"); + } +} + +#[tokio::test] +async fn non_structured_bodies_use_safe_fallback_without_losing_status_or_ids() { + for body in [ + b"synthetic-private-body-marker".as_slice(), + b"synthetic-private-body-marker", + b"", + br#""synthetic-private-body-marker""#, + br#"{"error":"synthetic-private-body-marker"}"#, + br#"{"detail":"synthetic-private-body-marker"}"#, + br#"{"error":{"code":"synthetic-private-body-marker"}}"#, + ] { + let provider = upstream_failure( + 503, + body, + vec![header("x-request-id", b"req_body_fallback")], + ); + let error = initial_error(EngineError::Provider(provider), Vec::new()).await; + assert_eq!(error["status"], 503); + assert_eq!(error["headers"]["x-request-id"], "req_body_fallback"); + assert_eq!( + error["error"], + json!({ + "message": "upstream service is unavailable", + "code": "upstream_unavailable", + "type": "server_error", + }) + ); + assert!(!error.to_string().contains("synthetic-private-body-marker")); + } +} + +#[tokio::test] +async fn final_failure_headers_take_precedence_over_observed_opening_headers() { + let provider = upstream_failure(403, b"", vec![header("x-oai-request-id", b"req_final")]) + .with_status(502) + .with_upstream_request_id(OpaqueUpstreamValue::new("req_stale_fact")); + let error = initial_error( + EngineError::Provider(provider), + vec![ + header("x-request-id", b"req_old_opening"), + header("x-codex-turn-state", b"old-turn-state"), + ], + ) + .await; + assert_eq!(error["status"], 403); + assert_eq!(error["headers"]["x-oai-request-id"], "req_final"); + assert!(error["headers"].get("x-request-id").is_none()); + assert!(!error.to_string().contains("req_old_opening")); + assert!(!error.to_string().contains("req_stale_fact")); + assert!(!error.to_string().contains("old-turn-state")); +} + +#[tokio::test] +async fn provider_failure_facts_survive_without_a_complete_http_response() { + let provider = ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous) + .with_status(429) + .with_upstream_request_id(OpaqueUpstreamValue::new("req_body_read_failed")); + let error = initial_error( + EngineError::Provider(provider), + vec![ + header("x-request-id", b"req_old"), + header("x-oai-request-id", b"req_old_alias"), + header("x-codex-turn-state", b"observed-turn-state"), + ], + ) + .await; + assert_eq!(error["status"], 429); + assert_eq!(error["headers"]["x-request-id"], "req_body_read_failed"); + assert_eq!( + error["headers"]["x-codex-turn-state"], + "observed-turn-state" + ); + assert!(error["headers"].get("x-oai-request-id").is_none()); + assert_eq!(error["error"]["message"], "upstream service is unavailable"); +} + +#[tokio::test] +async fn initial_error_headers_apply_the_response_security_boundary() { + let provider = upstream_failure( + 502, + b"", + vec![ + header("X-Request-Id", b"req_safe"), + header("x-gateway-request-id", b"upstream-cannot-claim-gateway-id"), + header("x-future-header", b"first"), + header("x-future-header", b"second"), + header("cf-ray", b"synthetic-ray"), + header("retry-after", b"11"), + header("authorization", b"synthetic-private-header-marker"), + header("set-cookie", b"synthetic-private-header-marker"), + header("chatgpt-account-id", b"synthetic-private-header-marker"), + header( + "x-codex-installation-id", + b"synthetic-private-header-marker", + ), + header("connection", b"x-private-hop"), + header("x-private-hop", b"synthetic-private-header-marker"), + header("sec-websocket-accept", b"synthetic-private-header-marker"), + header("content-length", b"123"), + header("content-type", b"text/html"), + header("bad\0name", b"synthetic-private-header-marker"), + header("x-bad-value", b"synthetic-private-header-marker\r\n"), + header("x-non-utf8", b"\xff"), + ], + ); + let error = initial_error(EngineError::Provider(provider), Vec::new()).await; + assert_eq!( + error["headers"], + json!({ + "x-request-id": "req_safe", + "x-gateway-request-id": "req_ws_atomic", + "x-future-header": "second", + "cf-ray": "synthetic-ray", + "retry-after": "11", + }) + ); +} + +#[tokio::test] +async fn initial_error_without_upstream_facts_keeps_the_local_contract() { + for (error, status, code) in [ + (EngineError::Deadline, 504, "request_timeout"), + (EngineError::EmptyRoutingPlan, 503, "no_available_provider"), + ( + EngineError::Provider(ProviderError::new( + ProviderErrorKind::Transport, + UpstreamSendState::NotSent, + )), + 502, + "upstream_unavailable", + ), + ] { + let error = initial_error(error, Vec::new()).await; + assert_eq!(error["status"], status); + assert_eq!(error["error"]["code"], code); + assert_eq!(error["headers"]["x-request-id"], "req_ws_atomic"); + } +} + +#[tokio::test] +async fn initial_error_never_uses_a_success_or_upgrade_status_as_an_error_status() { + for status in [0, 99, 101, 200, 204] { + let error = initial_error( + EngineError::Provider(upstream_failure(status, b"", Vec::new())), + Vec::new(), + ) + .await; + assert_eq!(error["status"], 502); + } +} + +#[tokio::test] +async fn opening_ids_stay_in_metadata_but_do_not_identify_initial_failures() { + for (failure, status) in [ + (EngineError::Deadline, 504), + ( + EngineError::Provider(ProviderError::new( + ProviderErrorKind::Transport, + UpstreamSendState::Ambiguous, + )), + 502, + ), + ] { + let trace = Arc::new(AtomicFailureTrace { + initial_errors: Mutex::new(VecDeque::from([failure])), + ..AtomicFailureTrace::default() + }); + let (mut socket, _server) = connect( + trace, + vec![ + header("x-request-id", b"req_old_opening"), + header("x-oai-request-id", b"req_observed"), + header("x-codex-turn-state", b"observed-turn-state"), + header("x-future-header", b"future-value"), + ], + ) + .await; + send_request(&mut socket).await; + // metadata 仍交付原会话观测；opening ID 不能因此成为当前失败的请求身份。 + assert_eq!( + next_event(&mut socket).await, + json!({ + "type": "response.metadata", + "headers": { + "x-request-id": "req_old_opening", + "x-oai-request-id": "req_observed", + "x-codex-turn-state": "observed-turn-state", + "x-future-header": "future-value", + }, + }) + ); + let error = next_event(&mut socket).await; + assert_eq!(error["type"], "error"); + assert_eq!(error["status"], status); + assert_eq!(error["request_id"], "req_ws_atomic"); + assert_eq!( + error["headers"], + json!({ + "x-request-id": "req_ws_atomic", + "x-gateway-request-id": "req_ws_atomic", + "x-codex-turn-state": "observed-turn-state", + "x-future-header": "future-value", + }) + ); + assert_no_duplicate_failure(&mut socket).await; + socket.close(None).await.unwrap(); + } +} + +#[tokio::test] +async fn errors_on_a_reused_connection_keep_their_own_upstream_ids() { + let errors = ["req_first", "req_second"].map(|request_id| { + EngineError::Provider( + upstream_failure(502, b"", Vec::new()) + .with_upstream_request_id(OpaqueUpstreamValue::new(request_id)), + ) + }); + let trace = Arc::new(AtomicFailureTrace { + initial_errors: Mutex::new(VecDeque::from(errors)), + ..AtomicFailureTrace::default() + }); + let (mut socket, _server) = connect(trace, Vec::new()).await; + for request_id in ["req_first", "req_second"] { + send_request(&mut socket).await; + let error = next_error(&mut socket).await; + assert_eq!(error["headers"]["x-request-id"], request_id); + assert_eq!(error["headers"]["x-gateway-request-id"], "req_ws_atomic"); + } + assert_no_duplicate_failure(&mut socket).await; + socket.close(None).await.unwrap(); +} + +#[tokio::test] +async fn deliverable_business_failures_are_not_rewritten_or_followed_by_a_second_error() { + for raw in [ + json!({ + "type": "error", + "status": 429, + "headers": {"x-request-id": "req_raw"}, + "error": {"message": "raw marker", "type": "custom_type", "code": "custom_code"}, + "future": {"keep": true}, + }), + json!({ + "type": "response.failed", + "response": { + "id": "resp_raw", + "status": "failed", + "error": {"message": "raw marker", "code": "custom_code"}, + }, + "future": {"keep": true}, + }), + ] { + let event = ProviderEvent::wire( + ProtocolWireEvent::json( + "openai", + Some(raw["type"].as_str().unwrap().to_owned()), + raw.clone(), + ) + .unwrap(), + ); + let trace = Arc::new(AtomicFailureTrace { + first_batch: Mutex::new(Some( + CoordinatedEvent::try_batch(vec![event], CommitRequirement::CommitBeforeDelivery) + .unwrap(), + )), + ..AtomicFailureTrace::default() + }); + let (mut socket, _server) = connect(Arc::clone(&trace), Vec::new()).await; + send_request(&mut socket).await; + assert_eq!(next_event(&mut socket).await["type"], "response.metadata"); + assert_eq!(next_event(&mut socket).await, raw); + assert_no_duplicate_failure(&mut socket).await; + assert!(trace.committed.load(Ordering::Acquire)); + assert!(trace.finalized.load(Ordering::Acquire)); + assert_eq!(trace.next_calls.load(Ordering::Acquire), 2); + socket.close(None).await.unwrap(); + } +} diff --git a/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs b/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs index 115a2c582..c4c96985d 100644 --- a/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs +++ b/backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs @@ -1,4 +1,5 @@ mod connection; +mod forward; mod protocol; use std::sync::{ @@ -307,6 +308,8 @@ struct AtomicFailureTrace { committed: AtomicBool, finalized: AtomicBool, downstream_websocket_connection_ids: Mutex>, + initial_errors: Mutex>, + first_batch: Mutex>, } struct AtomicFailureSession { @@ -319,6 +322,10 @@ impl ExecutionSession for AtomicFailureSession { fn next_event(&mut self) -> BoxFuture<'_, Result, EngineError>> { Box::pin(async move { let next_call = self.trace.next_calls.fetch_add(1, Ordering::AcqRel); + if let Some(error) = self.trace.initial_errors.lock().unwrap().pop_front() { + self.trace.finalized.store(true, Ordering::Release); + return Err(error); + } if self.fail_before_first_event && next_call == 0 { return Err(EngineError::Provider(ProviderError::new( ProviderErrorKind::Unavailable, @@ -326,7 +333,14 @@ impl ExecutionSession for AtomicFailureSession { ))); } match next_call { - 0 => Ok(Some(atomic_failure_batch())), + 0 => Ok(Some( + self.trace + .first_batch + .lock() + .unwrap() + .take() + .unwrap_or_else(atomic_failure_batch), + )), 1 => { self.trace.finalized.store(true, Ordering::Release); Err(EngineError::Provider(ProviderError::new( diff --git a/backend/crates/gateway-api/tests/openai/search/mod.rs b/backend/crates/gateway-api/tests/openai/search/mod.rs index 69374ea07..aab238ebf 100644 --- a/backend/crates/gateway-api/tests/openai/search/mod.rs +++ b/backend/crates/gateway-api/tests/openai/search/mod.rs @@ -171,6 +171,11 @@ async fn search_route_should_preserve_request_and_success_response_bytes() { .expect("search response"); assert_eq!(response.status(), StatusCode::CREATED); + assert_eq!(response.headers()["x-request-id"], "req_search_test"); + assert_eq!( + response.headers()["x-gateway-request-id"], + "req_search_test" + ); assert_eq!( response.headers().get("x-search-rate-limit"), Some(&"42".parse().expect("header value")) @@ -216,6 +221,11 @@ async fn search_route_should_return_the_exact_upstream_error_response() { .expect("search response"); assert_eq!(response.status(), StatusCode::UNPROCESSABLE_ENTITY); + assert_eq!(response.headers()["x-request-id"], "req_search_test"); + assert_eq!( + response.headers()["x-gateway-request-id"], + "req_search_test" + ); assert_eq!( response.headers().get("content-type"), Some(&"application/problem+json".parse().expect("content type")) diff --git a/backend/crates/gateway-core/src/diagnostics/capture.rs b/backend/crates/gateway-core/src/diagnostics/capture.rs index e4f3d3258..5182f4d27 100644 --- a/backend/crates/gateway-core/src/diagnostics/capture.rs +++ b/backend/crates/gateway-core/src/diagnostics/capture.rs @@ -1,4 +1,4 @@ -//! 诊断内容的统一筛选边界。未知字符串只保留长度和摘要。 +//! 诊断内容的统一筛选边界。未知键名与值只保留有界结构和摘要。 use serde_json::{Map, Value, json}; use sha2::{Digest, Sha256}; @@ -6,8 +6,12 @@ use sha2::{Digest, Sha256}; /// 原始字节的长度与 SHA-256，不复制或保留正文。 #[must_use] pub fn body_fingerprint(bytes: &[u8]) -> Value { + json!({"bytes": bytes.len(), "sha256": fingerprint_hex(bytes)}) +} + +fn fingerprint_hex(bytes: &[u8]) -> String { let digest = Sha256::digest(bytes); - let hex: String = digest + digest .iter() .flat_map(|byte| { let alphabet = b"0123456789abcdef"; @@ -16,17 +20,40 @@ pub fn body_fingerprint(bytes: &[u8]) -> Value { char::from(alphabet[usize::from(byte & 15)]), ] }) - .collect(); - json!({"bytes": bytes.len(), "sha256": hex}) + .collect() } -/// 保留可诊断的协议结构；凭据永不进入普通日志或数据库。 +/// 仅保留有限协议骨架；任意 JSON（包括 headers 和 metadata）不能声明自己可信。 #[must_use] pub fn diagnostic_json(value: &Value) -> Value { - capture(value, "", 0, &mut 96) + capture(value, "", Context::Protocol, 0, &mut 96) } -fn capture(value: &Value, key: &str, depth: usize, budget: &mut usize) -> Value { +/// 只有 transport 的上游 metadata / error 帧顶层才是受控头部边界。 +pub(super) fn diagnostic_event_json(value: &Value, stage: &str) -> Value { + let context = if stage == "upstream.event" + && value.get("headers").is_some_and(Value::is_object) + && matches!( + value.get("type").and_then(Value::as_str), + Some("response.metadata" | "codex.response.metadata" | "error") + ) { + Context::UpstreamEvent + } else { + Context::Protocol + }; + capture(value, "", context, 0, &mut 96) +} + +#[derive(Clone, Copy)] +enum Context { + Protocol, + UpstreamEvent, + Opaque, + Headers, + HeaderValue, +} + +fn capture(value: &Value, key: &str, context: Context, depth: usize, budget: &mut usize) -> Value { if *budget == 0 || depth > 6 { return json!({"omitted": true}); } @@ -42,7 +69,11 @@ fn capture(value: &Value, key: &str, depth: usize, budget: &mut usize) -> Value if *budget == 0 { break; } - result.insert(bounded(name, 96), capture(value, name, depth + 1, budget)); + let (captured_name, child_context) = field(context, name); + result.insert( + captured_name, + capture(value, name, child_context, depth + 1, budget), + ); } if result.len() < object.len() { result.insert( @@ -55,16 +86,86 @@ fn capture(value: &Value, key: &str, depth: usize, budget: &mut usize) -> Value Value::Array(items) => json!({ "length": items.len(), "sample": items.iter().take(4) - .map(|item| capture(item, key.as_str(), depth + 1, budget)) + .map(|item| { + // 受控头值只接受一层字符串数组；畸形嵌套不能继承头部放行规则。 + let context = if matches!(context, Context::HeaderValue) && !item.is_string() { + Context::Opaque + } else { + context + }; + capture(item, key.as_str(), context, depth + 1, budget) + }) .collect::>() }), Value::String(text) - if safe_header(&key) || safe_value_key(&key) && protocol_label(text) => + if matches!(context, Context::HeaderValue) && safe_header(&key) + || matches!(context, Context::Protocol | Context::UpstreamEvent) + && protocol_value(&key, text) => { Value::String(bounded(text, 256)) } Value::String(text) => body_fingerprint(text.as_bytes()), - other => other.clone(), + Value::Number(number) + if matches!(context, Context::Protocol) + && key == "status" + && number + .as_u64() + .is_some_and(|status| (100..=599).contains(&status)) => + { + value.clone() + } + other => body_fingerprint(other.to_string().as_bytes()), + } +} + +fn field(context: Context, name: &str) -> (String, Context) { + let child = match (context, name) { + (Context::UpstreamEvent, "headers") => Some(Context::Headers), + (Context::Headers, _) if safe_header(&name.to_ascii_lowercase()) => { + Some(Context::HeaderValue) + } + ( + Context::Protocol | Context::UpstreamEvent, + "response" | "error" | "input" | "output" | "item" | "part" | "content" | "type" + | "role" | "status", + ) => Some(Context::Protocol), + ( + Context::Protocol | Context::UpstreamEvent, + "id" | "model" | "message" | "text" | "delta" | "code" | "metadata" | "headers", + ) => Some(Context::Opaque), + _ => None, + }; + match child { + Some(context) => (name.to_owned(), context), + None => ( + format!("field_{}", fingerprint_hex(name.as_bytes())), + Context::Opaque, + ), + } +} + +fn protocol_value(key: &str, value: &str) -> bool { + match key { + "type" => { + known_event_type(value) + || matches!( + value, + "message" + | "function_call" + | "function_call_output" + | "reasoning" + | "input_text" + | "output_text" + | "input_image" + | "refusal" + ) + } + "role" => matches!( + value, + "system" | "developer" | "user" | "assistant" | "tool" + ), + "status" => matches!(value, "in_progress" | "completed" | "incomplete" | "failed"), + _ => false, } } @@ -131,14 +232,6 @@ fn safe_header(key: &str) -> bool { ) } -fn safe_value_key(key: &str) -> bool { - safe_header(key) - || matches!( - key, - "type" | "event" | "status" | "code" | "model" | "role" | "object" | "service_tier" - ) -} - pub(super) fn bounded(value: &str, limit: usize) -> String { value .chars() @@ -147,11 +240,39 @@ pub(super) fn bounded(value: &str, limit: usize) -> String { .collect() } -/// Event names/codes are protocol labels, never arbitrary upstream text. -pub(super) fn protocol_label(value: &str) -> bool { - !value.is_empty() - && value.len() <= 128 - && value - .bytes() - .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'.' | b'_' | b'-' | b':')) +pub(super) fn diagnostic_event_type(value: &str) -> Value { + if known_event_type(value) { + Value::String(value.to_owned()) + } else { + body_fingerprint(value.as_bytes()) + } +} + +/// 这里只决定诊断可明文展示的阶段，不校验协议；未知事件仍采集摘要，不能按字符形状放行。 +fn known_event_type(value: &str) -> bool { + matches!( + value, + "error" + | "response.create" + | "response.created" + | "response.in_progress" + | "response.completed" + | "response.failed" + | "response.incomplete" + | "response.metadata" + | "codex.response.metadata" + | "codex.rate_limits" + | "response.output_item.added" + | "response.output_item.done" + | "response.content_part.added" + | "response.content_part.done" + | "response.output_text.delta" + | "response.output_text.done" + | "response.function_call_arguments.delta" + | "response.function_call_arguments.done" + | "response.reasoning_summary_text.delta" + | "response.reasoning_summary_text.done" + | "response.reasoning_text.delta" + | "response.reasoning_text.done" + ) } diff --git a/backend/crates/gateway-core/src/diagnostics/trace.rs b/backend/crates/gateway-core/src/diagnostics/trace.rs index 624dbb122..f692b1bc5 100644 --- a/backend/crates/gateway-core/src/diagnostics/trace.rs +++ b/backend/crates/gateway-core/src/diagnostics/trace.rs @@ -10,8 +10,8 @@ use base64::{Engine as _, engine::general_purpose::STANDARD}; use serde::Serialize; use serde_json::{Value, json}; -use super::capture::bounded; -use super::{body_fingerprint, diagnostic_headers, diagnostic_json}; +use super::capture::{bounded, diagnostic_event_json, diagnostic_event_type}; +use super::{body_fingerprint, diagnostic_headers}; const MAX_EVENTS: usize = 128; const HEAD_EVENTS: usize = 8; @@ -124,7 +124,7 @@ impl TraceContext { } /// 同一个头部边界只采集一次：默认脱敏，显式开启 dump 时另存完整字节。 - /// facts 必须为调用方构造的安全对象；原始头值只进入独立 dump。 + /// facts 必须为调用方构造的安全对象；仅受控诊断头可明文保留，完整头部另存 dump。 pub fn headers<'a>( &self, stage: &'static str, @@ -182,27 +182,27 @@ impl TraceContext { .as_ref() .and_then(|v| v.get("type")) .and_then(Value::as_str) - .or(name) - .filter(|kind| super::capture::protocol_label(kind)); + .or(name); + // 后缀只用于观测合并，不授予明文权限；未知事件用摘要区分，避免合并成同一类。 let delta = event_type.is_some_and(|kind| kind.ends_with(".delta")); let data = json!({ "body": fingerprint, - "eventType": event_type.map(|kind| bounded(kind, 128)), + "eventType": event_type.map(diagnostic_event_type), "jsonValid": parsed.is_some(), - "metadata": parsed.as_ref().filter(|_| !delta).map(diagnostic_json), + "metadata": parsed.as_ref().filter(|_| !delta) + .map(|value| diagnostic_event_json(value, stage)), }); self.push(stage, data, delta); } /// 只记录业务解析边界；metadata 已由 transport 在解析前捕获。 pub fn wire_event(&self, protocol: &str, event_type: Option<&str>) { - let event_type = event_type.filter(|kind| super::capture::protocol_label(kind)); let delta = event_type.is_some_and(|kind| kind.ends_with(".delta")); self.push( "provider.event", json!({ "protocol": protocol, - "eventType": event_type.map(|kind| bounded(kind, 128)), + "eventType": event_type.map(diagnostic_event_type), }), delta, ); diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 2bfa24523..23875c021 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/backend/crates/gateway-core/src/engine/coordinator.rs @@ -230,11 +230,22 @@ struct CurrentAttempt { response_observation: Option, } +impl CurrentAttempt { + fn upstream_request_id(&self) -> Option<&str> { + self.response_observation + .as_ref() + .and_then(ProviderResponseObservation::request_id) + .or_else(|| self.metadata.upstream_request_id()) + .map(|id| id.as_str()) + } +} + struct FailureFinalization { outcome: ExecutionOutcome, send_state: UpstreamSendState, error: GatewayError, upstream_status_code: Option, + upstream_request_id: Option, provider_error_code: Option, raw_upstream_error: Option, retry_after_ms: Option, @@ -698,6 +709,7 @@ where send_state: self.current_send_state(), error, upstream_status_code: None, + upstream_request_id: None, provider_error_code: None, raw_upstream_error: None, retry_after_ms: None, @@ -871,6 +883,7 @@ where send_state: self.current_send_state(), error, upstream_status_code: None, + upstream_request_id: None, provider_error_code: None, raw_upstream_error: None, retry_after_ms: None, @@ -902,6 +915,7 @@ where send_state: self.current_send_state(), error, upstream_status_code: None, + upstream_request_id: None, provider_error_code: None, raw_upstream_error: None, retry_after_ms: None, @@ -923,6 +937,7 @@ where send_state: self.current_send_state(), error, upstream_status_code: None, + upstream_request_id: None, provider_error_code: None, raw_upstream_error: None, retry_after_ms: None, @@ -947,6 +962,7 @@ where send_state: self.current_send_state(), error, upstream_status_code: None, + upstream_request_id: None, provider_error_code: None, raw_upstream_error: None, retry_after_ms: None, @@ -1220,15 +1236,17 @@ where provider_kind: current.metadata.provider().clone(), account_id: Some(current.metadata.provider_account_id().clone()), upstream_model_id: current.metadata.upstream_model().cloned(), - upstream_status_code: current - .response_observation - .as_ref() - .and_then(ProviderResponseObservation::status_code), - upstream_request_id: current - .response_observation - .as_ref() - .and_then(ProviderResponseObservation::request_id) - .map(|value| value.as_str().to_owned()), + upstream_status_code: error.upstream_status().or_else(|| { + current + .response_observation + .as_ref() + .and_then(ProviderResponseObservation::status_code) + }), + upstream_request_id: error + .upstream_request_id() + .map(|id| id.as_str()) + .or_else(|| current.upstream_request_id()) + .map(str::to_owned), latency: current.started_at.elapsed().unwrap_or_default(), error, }), @@ -1376,22 +1394,11 @@ where let completed_at = SystemTime::now(); self.finalized_at = Some(completed_at); self.observation.finish(); - let upstream_request_id = self.current.as_ref().and_then(|current| { - current - .response_observation - .as_ref() - .and_then(|observation| { - observation - .request_id() - .map(|value| value.as_str().to_owned()) - }) - .or_else(|| { - current - .metadata - .upstream_request_id() - .map(|value| value.as_str().to_owned()) - }) - }); + let upstream_request_id = self + .current + .as_ref() + .and_then(CurrentAttempt::upstream_request_id) + .map(str::to_owned); let upstream_status_code = self .current .as_ref() @@ -1469,6 +1476,7 @@ where send_state, error: GatewayError::from_provider(error), upstream_status_code: error.upstream_status(), + upstream_request_id: error.upstream_request_id().map(|id| id.as_str().to_owned()), provider_error_code: error.upstream_code().map(|code| code.as_str().to_owned()), raw_upstream_error: error .raw_upstream_error() @@ -1531,6 +1539,7 @@ where send_state, error: gateway_error, upstream_status_code: None, + upstream_request_id: None, provider_error_code: None, raw_upstream_error: None, retry_after_ms: None, @@ -1561,22 +1570,11 @@ where return Ok(()); } self.observation.finish(); - let upstream_request_id = self.current.as_ref().and_then(|current| { - current - .response_observation - .as_ref() - .and_then(|observation| { - observation - .request_id() - .map(|value| value.as_str().to_owned()) - }) - .or_else(|| { - current - .metadata - .upstream_request_id() - .map(|value| value.as_str().to_owned()) - }) - }); + let upstream_request_id = self + .current + .as_ref() + .and_then(CurrentAttempt::upstream_request_id) + .map(str::to_owned); let observed_status_code = self .current .as_ref() @@ -1602,7 +1600,7 @@ where .upstream_status_code .or(observed_status_code), client_response_id: self.observation.client_response_id.clone(), - upstream_request_id, + upstream_request_id: finalization.upstream_request_id.or(upstream_request_id), upstream_response_id: self.observation.upstream_response_id.clone(), upstream_transport, http_version, @@ -1908,6 +1906,7 @@ fn record_trace_error(trace: &TraceContext, error: &ProviderError) { "stage": diagnostic.stage(), "code": diagnostic.code(), "message": diagnostic.as_str(), })), "upstreamStatus": error.upstream_status(), + "upstreamRequestId": error.upstream_request_id().map(|id| id.as_str()), "upstreamCode": error.upstream_code().map(|code| code.as_str()), "rawError": error.raw_upstream_error().map(|raw| { let value = serde_json::from_str(raw.as_str()).unwrap_or_else(|_| json!(raw.as_str())); diff --git a/backend/crates/gateway-core/tests/diagnostics/capture.rs b/backend/crates/gateway-core/tests/diagnostics/capture.rs index 0ea7cd6d4..5c3d72ff4 100644 --- a/backend/crates/gateway-core/tests/diagnostics/capture.rs +++ b/backend/crates/gateway-core/tests/diagnostics/capture.rs @@ -1,5 +1,5 @@ -use gateway_core::diagnostics::{diagnostic_headers, diagnostic_json}; -use serde_json::json; +use gateway_core::diagnostics::{body_fingerprint, diagnostic_headers, diagnostic_json}; +use serde_json::{Value, json}; #[test] fn credentials_and_content_never_enter_diagnostics_but_trace_ids_survive() { @@ -42,8 +42,175 @@ fn arbitrary_event_type_text_is_fingerprinted_instead_of_logged() { let snapshot = trace.snapshot().unwrap(); assert_eq!( snapshot["events"][0]["data"]["eventType"], - serde_json::Value::Null + body_fingerprint(b"secret@example.com") ); assert!(!snapshot.to_string().contains("secret@example.com")); assert!(!snapshot.to_string().contains("private-content")); } + +#[test] +fn user_map_keys_and_header_named_values_are_not_diagnostic_facts() { + let content = diagnostic_json(&json!({ + "type": "response.create", + "metadata": { + "PRIVATE_OWNER": "PRIVATE_VALUE", + "x-request-id": "PRIVATE_NESTED_REQUEST", + "headers": {"x-oai-request-id": "PRIVATE_NESTED_HEADER"}, + "type": "PRIVATE_TYPE", + "model": "PRIVATE_MODEL", + "account": 1234567890123_u64, + }, + "input": [{ + "role": "user", + "content": [{"type": "input_text", "text": "PRIVATE_PROMPT"}], + "PRIVATE_ITEM_KEY": [{"cf-ray": "PRIVATE_ARRAY_HEADER"}], + }], + })); + assert!( + !content.to_string().contains("PRIVATE_"), + "user keys and values must be summarized: {content}" + ); + assert!(!content.to_string().contains("1234567890123")); + assert_eq!(content["type"], "response.create"); + assert_eq!(content["input"]["length"], 1); + assert_eq!(content["input"]["sample"][0]["role"], "user"); + assert_eq!( + content["input"]["sample"][0]["content"]["sample"][0]["type"], + "input_text" + ); +} + +#[test] +fn json_headers_never_gain_trust_from_their_shape_or_event_type() { + for value in [ + json!({"headers": {"x-request-id": "PRIVATE_ROOT_HEADER"}}), + json!({ + "type": "codex.response.metadata", + "headers": {"x-request-id": "PRIVATE_ROOT_HEADER"}, + }), + json!([{"x-request-id": "PRIVATE_ARRAY_HEADER"}]), + ] { + let content = diagnostic_json(&value); + assert!(!content.to_string().contains("PRIVATE_"), "{content}"); + } +} + +#[test] +fn protocol_shaped_strings_are_not_a_safety_guarantee() { + let content = diagnostic_json(&json!({ + "type": "PRIVATE_TYPE", + "code": "PRIVATE_CODE", + "model": "PRIVATE_MODEL", + "role": "PRIVATE_ROLE", + "status": "PRIVATE_STATUS", + "object": "PRIVATE_OBJECT", + "service_tier": "PRIVATE_TIER", + })); + assert!(!content.to_string().contains("PRIVATE_"), "{content}"); + assert_eq!(content["type"], body_fingerprint(b"PRIVATE_TYPE")); +} + +#[test] +fn opaque_maps_do_not_inherit_even_known_protocol_keys_or_values() { + let content = diagnostic_json(&json!({ + "status": 429, + "metadata": { + "type": "response.completed", + "role": "user", + "status": 429, + "Authorization_PRIVATE_OWNER": "PRIVATE_CREDENTIAL", + "私人标记@example.test": "PRIVATE_VALUE", + }, + })); + assert_eq!(content["status"], 429); + let metadata = content["metadata"].as_object().unwrap(); + for value in ["response.completed", "user", "429"] { + assert!( + metadata + .values() + .any(|v| *v == body_fingerprint(value.as_bytes())) + ); + } + let text = content.to_string(); + for original in ["PRIVATE_", "私人标记", "response.completed"] { + assert!(!text.contains(original), "{text}"); + } +} + +#[test] +fn diagnostic_capture_keeps_bounded_samples_of_unknown_keys_without_collisions() { + let prefix = "PRIVATE_".repeat(16); + let value = json!({ + format!("{prefix}one"): "PRIVATE_ONE", + format!("{prefix}two"): "PRIVATE_TWO", + }); + let content = diagnostic_json(&value); + let entries = content.as_object().unwrap(); + assert_eq!(entries.len(), 2); + assert!( + entries + .values() + .any(|v| *v == body_fingerprint(b"PRIVATE_ONE")) + ); + assert!( + entries + .values() + .any(|v| *v == body_fingerprint(b"PRIVATE_TWO")) + ); + assert!(!content.to_string().contains("PRIVATE_")); +} + +#[test] +fn diagnostic_capture_bounds_fields_arrays_depth_and_total_nodes() { + let fields: serde_json::Map<_, _> = (0..100) + .map(|i| (format!("PRIVATE_FIELD_{i}"), json!("PRIVATE_VALUE"))) + .collect(); + let content = diagnostic_json(&Value::Object(fields)); + assert_eq!(content.as_object().unwrap().len(), 49); + assert_eq!(content["_omittedFields"], 52); + + let array = diagnostic_json(&json!(vec!["PRIVATE_VALUE"; 8])); + assert_eq!(array["length"], 8); + assert_eq!(array["sample"].as_array().unwrap().len(), 4); + + let mut deep = json!("PRIVATE_DEEP"); + for _ in 0..12 { + deep = json!({"PRIVATE_DEPTH": deep}); + } + let content = diagnostic_json(&deep); + assert!(content.to_string().contains("\"omitted\":true")); + assert!(!content.to_string().contains("PRIVATE_")); + + let wide: serde_json::Map<_, _> = (0..48) + .map(|i| (format!("PRIVATE_NODE_{i}"), json!(vec!["PRIVATE_VALUE"; 4]))) + .collect(); + let content = diagnostic_json(&Value::Object(wide)); + assert!(content["_omittedFields"].as_u64().unwrap() > 0); + assert!(content.to_string().len() < 32 * 1024); + assert!(!content.to_string().contains("PRIVATE_")); +} + +#[test] +fn credentials_remain_redacted_at_nested_json_and_real_header_boundaries() { + for name in [ + "Authorization", + "Cookie", + "Set-Cookie", + "api_key", + "api-key", + "access_token", + "refresh_token", + "password", + "client_secret", + "attestation", + "x-oai-is", + "x-oai-is-update", + ] { + let content = diagnostic_json(&json!({"metadata": {name: "PRIVATE_CREDENTIAL"}})); + assert!(!content.to_string().contains("PRIVATE_")); + assert!(content.to_string().contains("")); + let headers = diagnostic_headers([(name, "PRIVATE_CREDENTIAL")]); + assert!(!headers.to_string().contains("PRIVATE_")); + assert_eq!(headers[name.to_ascii_lowercase()][0], ""); + } +} diff --git a/backend/crates/gateway-core/tests/diagnostics/stream.rs b/backend/crates/gateway-core/tests/diagnostics/stream.rs index 6346fc9c4..664349188 100644 --- a/backend/crates/gateway-core/tests/diagnostics/stream.rs +++ b/backend/crates/gateway-core/tests/diagnostics/stream.rs @@ -1,4 +1,4 @@ -use gateway_core::diagnostics::{StreamCapture, StreamFormat, TraceContext}; +use gateway_core::diagnostics::{StreamCapture, StreamFormat, TraceContext, body_fingerprint}; #[test] fn sse_capture_reassembles_every_chunk_boundary_and_preserves_unknown_events() { @@ -12,11 +12,13 @@ fn sse_capture_reassembles_every_chunk_boundary_and_preserves_unknown_events() { let snapshot = trace.snapshot().unwrap(); let events = snapshot["events"].as_array().unwrap(); assert_eq!(events.len(), 3, "boundary {boundary}"); - assert_eq!(events[0]["data"]["eventType"], "future.metadata"); + // 未知 SSE 名称和伪装成头部的 JSON 仍有摘要，但不能获得协议字段的明文权限。 assert_eq!( - events[0]["data"]["metadata"]["headers"]["x-request-id"], - "up-1" + events[0]["data"]["eventType"], + body_fingerprint(b"future.metadata") ); + assert!(!snapshot.to_string().contains("up-1")); + assert!(!snapshot.to_string().contains("中文")); assert_eq!(events[1]["data"]["eventType"], "response.completed"); assert_eq!(events[2]["data"]["partialFrame"], false); } diff --git a/backend/crates/gateway-core/tests/diagnostics/trace.rs b/backend/crates/gateway-core/tests/diagnostics/trace.rs index 00d57432e..c319136c3 100644 --- a/backend/crates/gateway-core/tests/diagnostics/trace.rs +++ b/backend/crates/gateway-core/tests/diagnostics/trace.rs @@ -1,5 +1,15 @@ -use gateway_core::diagnostics::TraceContext; +use std::{ + fmt::Write as _, + sync::{Arc, Mutex}, +}; + +use gateway_core::diagnostics::{TraceContext, body_fingerprint}; use serde_json::json; +use tracing::{ + Event, Metadata, Subscriber, + field::{Field, Visit}, + span::{Attributes, Id, Record}, +}; #[test] fn bounded_history_preserves_start_failure_and_final_result() { @@ -140,3 +150,232 @@ fn a_long_successful_retry_keeps_the_earlier_failure_ahead_of_routine_events() { assert_eq!(events.last().unwrap()["stage"], "request.finished"); assert!(events.len() <= 128); } + +#[test] +fn only_top_level_headers_in_controlled_upstream_frames_keep_correlation_values() { + for event_type in ["response.metadata", "codex.response.metadata", "error"] { + let trace = TraceContext::new("req_headers"); + let body = serde_json::to_vec(&json!({ + "type": event_type, + "status": 429, + "headers": { + "x-request-id": "upstream-controlled", + "x-oai-request-id": ["upstream-one", "upstream-two"], + "authorization": "PRIVATE_CREDENTIAL", + "PRIVATE_HEADER_KEY": "PRIVATE_HEADER_VALUE", + "cf-ray": [{"x-request-id": "PRIVATE_MALFORMED_HEADER"}], + "request-id": [["PRIVATE_NESTED_ARRAY"]], + }, + "metadata": { + "x-request-id": "PRIVATE_METADATA_VALUE", + "headers": {"x-request-id": "PRIVATE_NESTED_VALUE"}, + }, + "response": { + "type": "codex.response.metadata", + "headers": {"x-request-id": "PRIVATE_RESPONSE_HEADER"}, + }, + })) + .unwrap(); + trace.capture("upstream.event", &body); + let snapshot = trace.snapshot().unwrap(); + let metadata = &snapshot["events"][0]["data"]["metadata"]; + assert_eq!(metadata["headers"]["x-request-id"], "upstream-controlled"); + assert_eq!( + metadata["headers"]["x-oai-request-id"]["sample"], + json!(["upstream-one", "upstream-two"]) + ); + assert_eq!(metadata["status"], 429); + assert!(!snapshot.to_string().contains("PRIVATE_"), "{snapshot}"); + } +} + +#[test] +fn body_capture_never_infers_controlled_headers_from_client_supplied_type() { + for stage in [ + "client.request.body", + "upstream.request.body", + "upstream.response.body", + "upstream.error.body", + ] { + let trace = TraceContext::new("req_untrusted"); + trace.capture( + stage, + br#"{"type":"codex.response.metadata","headers":{"x-request-id":"PRIVATE_HEADER"}}"#, + ); + assert!( + !trace.snapshot().unwrap().to_string().contains("PRIVATE_"), + "{stage}" + ); + } + let trace = TraceContext::new("req_malformed"); + trace.capture( + "upstream.event", + br#"{"type":"codex.response.metadata","headers":[{"x-request-id":"PRIVATE_HEADER"}]}"#, + ); + assert!(!trace.snapshot().unwrap().to_string().contains("PRIVATE_")); +} + +#[test] +fn unknown_event_names_are_fingerprinted_at_all_trace_entry_points() { + let trace = TraceContext::new("req_names"); + trace.capture("upstream.event", br#"{"type":"PRIVATE_EVENT.delta"}"#); + trace.capture_named_event("upstream.event", b"{}", Some("PRIVATE_SSE_EVENT")); + trace.wire_event("openai", Some("PRIVATE_WIRE_EVENT")); + let snapshot = trace.snapshot().unwrap(); + for (event, name) in snapshot["events"].as_array().unwrap().iter().zip([ + "PRIVATE_EVENT.delta", + "PRIVATE_SSE_EVENT", + "PRIVATE_WIRE_EVENT", + ]) { + assert_eq!( + event["data"]["eventType"], + body_fingerprint(name.as_bytes()) + ); + } + assert!(!snapshot.to_string().contains("PRIVATE_")); +} + +#[test] +fn unknown_deltas_coalesce_by_summary_without_merging_distinct_event_types() { + let trace = TraceContext::new("req_unknown_delta"); + for name in ["PRIVATE_FIRST.delta", "PRIVATE_SECOND.delta"] { + let bytes = serde_json::to_vec(&json!({"type": name, "delta": "PRIVATE_TEXT"})).unwrap(); + for _ in 0..100 { + trace.capture("upstream.event", &bytes); + trace.wire_event("openai", Some(name)); + } + } + let snapshot = trace.snapshot().unwrap(); + let events = snapshot["events"].as_array().unwrap(); + assert_eq!(events.len(), 4); + assert!(events.iter().all(|event| event["count"] == 100)); + assert_eq!(snapshot["totalEvents"], 400); + assert_eq!(snapshot["droppedEvents"], 0); + assert_ne!( + events[0]["data"]["eventType"], + events[2]["data"]["eventType"] + ); + assert!(!snapshot.to_string().contains("PRIVATE_")); +} + +#[test] +fn size_gaps_and_truncated_events_do_not_reintroduce_user_content() { + let trace = TraceContext::new("req_limits"); + let private = "PRIVATE_OVERSIZE".repeat(10_000); + trace.capture("client.request.body", private.as_bytes()); + let fields: serde_json::Map<_, _> = (0..48) + .map(|i| (format!("PRIVATE_FIELD_{i}"), json!("PRIVATE_VALUE"))) + .collect(); + let bytes = serde_json::to_vec(&fields).unwrap(); + trace.capture_named_event("upstream.event", &bytes, Some("PRIVATE_EVENT")); + let snapshot = trace.snapshot().unwrap(); + assert_eq!(snapshot["events"][0]["stage"], "capture.gap"); + assert_eq!( + snapshot["events"][0]["data"]["body"], + body_fingerprint(private.as_bytes()) + ); + assert_eq!(snapshot["events"][1]["data"]["truncated"], true); + assert_eq!( + snapshot["events"][1]["data"]["eventType"], + body_fingerprint(b"PRIVATE_EVENT") + ); + assert!(!snapshot.to_string().contains("PRIVATE_")); +} + +#[test] +fn ordinary_trace_logs_and_serialized_snapshots_share_the_sanitized_capture_boundary() { + // tracing 的 callsite 缓存是进程级的；隔离并行测试的注册，不修改生产日志或全局 subscriber。 + const CHILD: &str = "GATEWAY_CORE_DIAGNOSTICS_LOG_TEST"; + if std::env::var_os(CHILD).is_none() { + let thread = std::thread::current(); + let output = std::process::Command::new(std::env::current_exe().unwrap()) + .args(["--exact", thread.name().unwrap()]) + .env(CHILD, "1") + .output() + .unwrap(); + assert!( + output.status.success() && String::from_utf8_lossy(&output.stdout).contains("1 passed"), + "isolated log regression failed:\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + return; + } + // 仅捕获本线程的普通 trace 事件，不打开 request_dump，也不读写真实日志文件。 + let logs = Arc::new(Mutex::new(Vec::new())); + let snapshot = tracing::subscriber::with_default(TraceLog(Arc::clone(&logs)), || { + let trace = TraceContext::new("req_feedback"); + let exchange = trace.attempt(2).exchange("websocket"); + exchange.capture( + "client.request.body", + br#"{"type":"PRIVATE_EVENT","metadata":{"PRIVATE_KEY":"PRIVATE_VALUE","x-request-id":"PRIVATE_HEADER"},"input":[{"role":"user","content":[{"type":"input_text","text":"PRIVATE_PROMPT"}]}]}"#, + ); + exchange.capture_named_event("upstream.event", b"{}", Some("PRIVATE_SSE_EVENT")); + exchange.wire_event("openai", Some("PRIVATE_WIRE_EVENT")); + exchange.headers( + "upstream.response.headers", + json!({"status": 429}), + [ + ("x-request-id", b"upstream-controlled".as_slice()), + ("authorization", b"PRIVATE_CREDENTIAL".as_slice()), + ("set-cookie", b"PRIVATE_COOKIE".as_slice()), + ], + ); + let snapshot = trace.snapshot().unwrap(); + assert_eq!(snapshot["wireDumpEnabled"], false); + assert_eq!(snapshot["wireFrames"], 0); + snapshot + }); + // API / 反馈导出使用的是这个 Value 的序列化结果，而不只是 Debug 表示。 + let exported = String::from_utf8(serde_json::to_vec(&snapshot).unwrap()).unwrap(); + assert!(!exported.contains("PRIVATE_"), "{exported}"); + let logs = logs.lock().unwrap(); + let output = logs.join("\n"); + assert!(!logs.is_empty(), "必须真实捕获 tracing 事件，不能空集通过"); + assert!(!output.contains("PRIVATE_"), "{output}"); + for fact in [ + "req_feedback", + "upstream-controlled", + "upstream.response.headers", + ] { + assert!(exported.contains(fact), "{exported}"); + assert!(output.contains(fact), "{output}"); + } + assert!(output.contains("attempt_index=2")); + assert!(output.contains("exchange_id=1")); + assert!(output.contains("request.trace.closed")); +} + +struct TraceLog(Arc>>); + +impl Subscriber for TraceLog { + fn enabled(&self, metadata: &Metadata<'_>) -> bool { + metadata.target() == "request_trace" + } + + fn new_span(&self, _span: &Attributes<'_>) -> Id { + Id::from_u64(1) + } + + fn record(&self, _span: &Id, _values: &Record<'_>) {} + + fn record_follows_from(&self, _span: &Id, _follows: &Id) {} + + fn event(&self, event: &Event<'_>) { + let mut text = String::new(); + event.record(&mut LogFields(&mut text)); + self.0.lock().unwrap().push(text); + } + + fn enter(&self, _span: &Id) {} + + fn exit(&self, _span: &Id) {} +} + +struct LogFields<'a>(&'a mut String); + +impl Visit for LogFields<'_> { + fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) { + write!(self.0, "{}={value:?} ", field.name()).unwrap(); + } +} diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 4edb984bf..06a49d62c 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/backend/crates/gateway-core/tests/engine/coordinator.rs @@ -997,6 +997,50 @@ fn failed_image_request_should_persist_failure() { ); } +#[test] +fn failure_request_id_prefers_error_then_response_observation() { + for error_request_id in [None, Some("upstream-error")] { + let operation = generate_operation(); + let route_plan = plan(&operation); + let observation = ProviderResponseObservation::new( + UpstreamTransport::new("http_sse").expect("transport"), + ) + .with_status_code(200) + .with_request_id(OpaqueUpstreamValue::new("upstream-opening")); + let mut error = + ProviderError::new(ProviderErrorKind::InvalidRequest, UpstreamSendState::Sent) + .with_status(400); + if let Some(request_id) = error_request_id { + error = error.with_upstream_request_id(OpaqueUpstreamValue::new(request_id)); + } + let (coordinator, store, _) = coordinator(vec![Script::ObservedStream { + account_id: "acct_observed", + items: vec![Ok(ProviderEvent::observation(observation)), Err(error)], + }]); + let mut session = block_on(coordinator.start( + model_request(&operation, SystemTime::now() + Duration::from_secs(30)), + operation, + route_plan, + None, + None, + CancellationToken::new(), + )) + .expect("start execution"); + + assert!(matches!( + block_on(session.collect_uncommitted()), + Err(EngineError::Provider(error)) if error.kind() == ProviderErrorKind::InvalidRequest + )); + let state = store.state.lock().expect("store lock"); + let finalization = &state.finalizations[0]; + assert_eq!( + finalization.upstream_request_id.as_deref(), + Some(error_request_id.unwrap_or("upstream-opening")) + ); + assert_eq!(finalization.upstream_status_code, Some(400)); + } +} + #[test] fn response_observation_is_persisted_but_never_delivered() { let operation = generate_operation(); @@ -1292,6 +1336,8 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() { Ok(ProviderEvent::observation(first_observation)), Err( ProviderError::new(ProviderErrorKind::Unavailable, UpstreamSendState::Sent) + .with_status(504) + .with_upstream_request_id(OpaqueUpstreamValue::new("discarded-error")) .with_replay_safe() .with_diagnostic( gateway_core::error::ProviderDiagnostic::new( @@ -1363,10 +1409,10 @@ fn discarded_attempt_observation_does_not_leak_into_retry_result() { .first_event_ms .is_some_and(|elapsed| elapsed < 987_654) ); - assert_eq!(state.intermediate_status_codes, vec![Some(503)]); + assert_eq!(state.intermediate_status_codes, vec![Some(504)]); assert_eq!( state.intermediate_request_ids, - vec![Some("discarded-request".to_owned())] + vec![Some("discarded-error".to_owned())] ); } diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 9c32e830d..69f6bc2f4 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/backend/crates/providers/openai/src/provider/execution.rs @@ -681,7 +681,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { &response, &request, ); - if let Some(observation) = observation_state.observation() { + if let Some(observation) = observation_state.observation(None) { yield ProviderEvent::observation(observation); } if let Some(etag) = response.response_metadata.models_etag.as_deref() @@ -710,7 +710,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { let response_transport = response.transport; let websocket_connection_id = response.websocket_connection_id; let mut body = response.body; - let failure_diagnostics = response.diagnostics.clone(); + let mut failure_diagnostics = response.diagnostics.clone(); + if response_transport == CodexBackendTransport::WebSocket { + // opening ID 标识连接，不可作为缺失请求级错误头时的当前请求 ID。 + failure_diagnostics.request_id = None; + } let failure_set_cookie_headers = response.set_cookie_headers.clone(); let failure_rate_limit_headers = response.rate_limit_headers.clone(); let mut passive_quota_observation = @@ -791,7 +795,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { ) .await; let observation_event = if rate_limits_changed || turn_state_merge.is_some() { - observation_state.observation().map(ProviderEvent::observation) + observation_state.observation(None).map(ProviderEvent::observation) } else { None }; @@ -934,8 +938,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { || service_tier_changed || timing_changed || turn_state_changed - || terminal_changed) - && let Some(observation) = observation_state.observation() + || terminal_changed + || (response_transport == CodexBackendTransport::WebSocket && terminal_failure.is_some())) + && let Some(observation) = observation_state.observation( + terminal_failure.as_ref().map(|(failure, _)| &failure.error) + ) { yield ProviderEvent::observation(observation); } @@ -1054,8 +1061,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream { || timing_changed || rate_limits_changed || turn_state_changed - || terminal_changed) - && let Some(observation) = observation_state.observation() + || terminal_changed + || (response_transport == CodexBackendTransport::WebSocket && terminal_failure.is_some())) + && let Some(observation) = observation_state.observation( + terminal_failure.as_ref().map(|(failure, _)| &failure.error) + ) { yield ProviderEvent::observation(observation); } diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 7500ca02a..9f14ad7cf 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -664,6 +664,7 @@ pub(super) fn stream_transport_allows_pre_delivery_retry(error: &CodexClientErro match error { CodexClientError::Http(_) | CodexClientError::HttpJson(_) + | CodexClientError::ErrorBodyRead { .. } | CodexClientError::StreamIdleTimeout { .. } => true, CodexClientError::WebSocket(error) => !matches!( error.classified(), @@ -766,6 +767,27 @@ pub(super) fn map_client_error( ProviderErrorKind::Protocol, UpstreamSendState::Sent, )), + CodexClientError::ErrorBodyRead { + source, + status, + diagnostics, + .. + } => { + // 已知响应头只补充观测事实；不将残缺响应重新分类成可换号重放的上游拒绝。 + let mut error = provider_error( + if source.is_timeout() { + ProviderErrorKind::Timeout + } else { + ProviderErrorKind::Transport + }, + uncertain_state, + ) + .with_status(status.as_u16()); + if let Some(request_id) = diagnostics.request_id { + error = error.with_upstream_request_id(OpaqueUpstreamValue::new(request_id)); + } + MappedProviderFailure::plain(error) + } CodexClientError::Http(error) | CodexClientError::HttpJson(error) => { let send_state = if error.is_connect() { UpstreamSendState::NotSent @@ -807,10 +829,9 @@ pub(super) fn map_client_error( "websocket_close_{close_code}" ))); } - if matches!( - error.classified(), - CodexWebSocketExchangeError::ConnectionLimitReached - ) { + if let CodexWebSocketExchangeError::ConnectionLimitReached(upstream) = + error.classified() + { failure.error = failure .error @@ -818,6 +839,14 @@ pub(super) fn map_client_error( .with_upstream_code(OpaqueUpstreamValue::new( WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE.to_owned(), )); + if let Some(status) = upstream.explicit_status_code { + failure.error = failure.error.with_status(status); + } + if let Some(request_id) = &upstream.request_id { + failure.error = failure + .error + .with_upstream_request_id(OpaqueUpstreamValue::new(request_id.clone())); + } } if let Some(client_visible_error) = client_visible_error { failure.error = failure @@ -854,6 +883,15 @@ pub(super) fn map_client_error( fn client_diagnostic(error: &CodexClientError) -> Option { let (stage, code, message) = match error { CodexClientError::WebSocket(error) => return Some(websocket_diagnostic(error)), + CodexClientError::ErrorBodyRead { status, source, .. } => { + return Some( + ProviderDiagnostic::new(format!( + "OpenAI HTTP error response body read failed: status={status}; complete body unavailable" + )) + .with_io_cause(source) + .with_classification("receive", "body_read_failed"), + ); + } CodexClientError::Http(error) | CodexClientError::HttpJson(error) => { let (stage, code) = if error.is_builder() { ("prepare", "http_request_build_failed") @@ -945,7 +983,7 @@ fn websocket_diagnostic(error: &CodexWebSocketExchangeError) -> ProviderDiagnost CodexWebSocketExchangeError::ClosedBeforeTerminal(_) => { ("receive", "closed_before_terminal") } - CodexWebSocketExchangeError::ConnectionLimitReached => { + CodexWebSocketExchangeError::ConnectionLimitReached(_) => { ("upstream", "connection_limit_reached") } CodexWebSocketExchangeError::ContinuationUnavailable { .. } => { @@ -1028,7 +1066,7 @@ fn websocket_diagnostic_message(error: &CodexWebSocketExchangeError) -> Provider "OpenAI WebSocket opening was rejected with status {}", upstream.status_code ), - CodexWebSocketExchangeError::ConnectionLimitReached => { + CodexWebSocketExchangeError::ConnectionLimitReached(_) => { "OpenAI requested a new WebSocket connection".to_owned() } CodexWebSocketExchangeError::ContinuationUnavailable { reason } => { @@ -1053,6 +1091,9 @@ fn websocket_diagnostic_message(error: &CodexWebSocketExchangeError) -> Provider } fn websocket_raw_error(error: &CodexWebSocketExchangeError) -> Option { + if let CodexWebSocketExchangeError::ConnectionLimitReached(failure) = error.classified() { + return Some(RawUpstreamError::new(failure.raw_body().to_owned())); + } let close = error.close_before_terminal()?; Some(RawUpstreamError::new( json!({ @@ -1068,14 +1109,11 @@ fn websocket_raw_error(error: &CodexWebSocketExchangeError) -> Option Option { - if matches!( - error.classified(), - CodexWebSocketExchangeError::ConnectionLimitReached - ) { + if let CodexWebSocketExchangeError::ConnectionLimitReached(failure) = error.classified() { return Some(ClientVisibleUpstreamError::new( - "websocket connection limit reached", - Some(WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE.to_owned()), - Some("websocket_error".to_owned()), + failure.message.clone(), + failure.upstream_code.clone(), + failure.upstream_type.clone(), )); } let close = error.close_before_terminal()?; @@ -1288,7 +1326,7 @@ pub(super) fn websocket_send_state(error: &CodexWebSocketExchangeError) -> Upstr | CodexWebSocketExchangeError::SharedConnectFailed | CodexWebSocketExchangeError::ContinuationUnavailable { .. } => UpstreamSendState::NotSent, CodexWebSocketExchangeError::Upstream(_) - | CodexWebSocketExchangeError::ConnectionLimitReached + | CodexWebSocketExchangeError::ConnectionLimitReached(_) | CodexWebSocketExchangeError::InvalidSse(_) | CodexWebSocketExchangeError::UnexpectedBinaryEvent => UpstreamSendState::Sent, CodexWebSocketExchangeError::Transport(_) @@ -1320,7 +1358,7 @@ pub(super) fn websocket_error_kind(error: &CodexWebSocketExchangeError) -> Provi ProviderErrorKind::Unavailable } CodexWebSocketExchangeError::Upstream(_) => ProviderErrorKind::Unavailable, - CodexWebSocketExchangeError::ConnectionLimitReached => ProviderErrorKind::RateLimited, + CodexWebSocketExchangeError::ConnectionLimitReached(_) => ProviderErrorKind::RateLimited, CodexWebSocketExchangeError::Transport(_) | CodexWebSocketExchangeError::Connect(_) | CodexWebSocketExchangeError::PostSendAmbiguous { .. } diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index 8f821107b..01e4b7e52 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -88,15 +88,32 @@ impl OpenAiResponseObservationState { } } - pub(super) fn observation(&self) -> Option { + pub(super) fn observation( + &self, + failure: Option<&ProviderError>, + ) -> Option { + // 失败快照的请求 ID 来自当前错误；连接诊断仍保持 opening 原貌，不维护第二份可变 ID。 + let failure_diagnostics = failure + .filter(|_| self.transport == CodexBackendTransport::WebSocket) + .map(|error| { + let mut diagnostics = self.diagnostics.clone(); + diagnostics.request_id = + error.upstream_request_id().map(|id| id.as_str().to_owned()); + diagnostics + }); let mut observation = codex_response_observation( self.transport, - &self.diagnostics, + failure_diagnostics.as_ref().unwrap_or(&self.diagnostics), &self.response_metadata, &self.metrics, self.websocket_pool_decision, self.timings, )?; + if self.transport == CodexBackendTransport::WebSocket + && let Some(status) = failure.and_then(ProviderError::upstream_status) + { + observation = observation.with_status_code(status); + } if let Some(metadata) = self.provider_metadata() { observation = observation.with_provider_metadata(metadata); } @@ -206,6 +223,11 @@ impl OpenAiResponseObservationState { pub(super) fn provider_metadata(&self) -> Option { let mut metadata = Map::new(); metadata.insert("schemaVersion".to_owned(), json!(2)); + if self.transport == CodexBackendTransport::WebSocket + && let Some(request_id) = &self.diagnostics.request_id + { + metadata.insert("websocketOpeningRequestId".to_owned(), json!(request_id)); + } if let Some(model) = self .response_metadata .effective_model @@ -330,6 +352,13 @@ pub(super) fn codex_error_observation( transport, transport_metrics, .. + } + | CodexClientError::ErrorBodyRead { + status, + diagnostics, + transport, + transport_metrics, + .. } => { observation = codex_response_observation( *transport, @@ -341,14 +370,21 @@ pub(super) fn codex_error_observation( )? .with_status_code(status.as_u16()); } - CodexClientError::WebSocket(error) - if matches!(error.classified(), CodexWebSocketExchangeError::Upstream(_)) => - { - let CodexWebSocketExchangeError::Upstream(upstream) = error.classified() else { - unreachable!("websocket error was checked above") + CodexClientError::WebSocket(error) => { + let (status, request_id) = match error.classified() { + CodexWebSocketExchangeError::ConnectionLimitReached(failure) => { + (failure.explicit_status_code, failure.request_id.as_deref()) + } + CodexWebSocketExchangeError::Upstream(upstream) => ( + Some(upstream.status_code), + upstream.diagnostics.request_id.as_deref(), + ), + _ => return Some(observation), }; - observation = observation.with_status_code(upstream.status_code); - if let Some(request_id) = upstream.diagnostics.request_id.as_deref() { + if let Some(status) = status { + observation = observation.with_status_code(status); + } + if let Some(request_id) = request_id { observation = observation.with_request_id(OpaqueUpstreamValue::new(request_id.to_owned())); } diff --git a/backend/crates/providers/openai/src/transport/client.rs b/backend/crates/providers/openai/src/transport/client.rs index 742528c2a..7073beadf 100644 --- a/backend/crates/providers/openai/src/transport/client.rs +++ b/backend/crates/providers/openai/src/transport/client.rs @@ -194,6 +194,16 @@ pub enum CodexClientError { /// 非流式 JSON 请求的 Reqwest 传输失败。 #[error("HTTP JSON transport error: {0}")] HttpJson(#[source] reqwest::Error), + /// 已收到错误响应头，但未能读取完整正文；不能作为可透传的 HTTP 原响应。 + #[error("upstream error response body read failed for status {status}")] + ErrorBodyRead { + #[source] + source: reqwest::Error, + status: StatusCode, + diagnostics: Box, + transport: CodexBackendTransport, + transport_metrics: Box, + }, /// 自定义 CA 构建失败。 #[error("custom CA transport error: {0}")] CustomCa(#[from] CustomCaError), @@ -258,6 +268,13 @@ impl fmt::Debug for CodexClientError { match self { Self::Http(_) => formatter.write_str("CodexClientError::Http([REDACTED])"), Self::HttpJson(_) => formatter.write_str("CodexClientError::HttpJson([REDACTED])"), + Self::ErrorBodyRead { + status, transport, .. + } => formatter + .debug_struct("CodexClientError::ErrorBodyRead") + .field("status", status) + .field("transport", transport) + .finish_non_exhaustive(), Self::CustomCa(_) => formatter.write_str("CodexClientError::CustomCa([REDACTED])"), Self::InvalidHeaderName(_) => { formatter.write_str("CodexClientError::InvalidHeaderName([REDACTED])") @@ -312,7 +329,9 @@ impl CodexClientError { | Self::ModelCatalog(_) => Some(CodexBackendTransport::HttpSse), Self::HttpJson(_) => Some(CodexBackendTransport::HttpJson), Self::WebSocket(_) => Some(CodexBackendTransport::WebSocket), - Self::Upstream { transport, .. } => Some(*transport), + Self::Upstream { transport, .. } | Self::ErrorBodyRead { transport, .. } => { + Some(*transport) + } Self::CustomCa(_) | Self::InvalidHeaderName(_) | Self::InvalidHeaderValue(_) diff --git a/backend/crates/providers/openai/src/transport/client_json.rs b/backend/crates/providers/openai/src/transport/client_json.rs index 89be0b020..aa7bd9334 100644 --- a/backend/crates/providers/openai/src/transport/client_json.rs +++ b/backend/crates/providers/openai/src/transport/client_json.rs @@ -92,9 +92,15 @@ impl CodexBackendClient { .get(CONTENT_TYPE) .map(|value| value.as_bytes().to_vec()); let client_headers = response_meta::client_headers(response.headers()); - let raw_body = read_error_response_body(response) - .await - .map_err(CodexClientError::HttpJson)?; + let raw_body = read_error_response_body(response).await.map_err(|source| { + CodexClientError::ErrorBodyRead { + source, + status, + diagnostics: Box::new(diagnostics.clone()), + transport: CodexBackendTransport::HttpJson, + transport_metrics: Box::new(transport_metrics.clone()), + } + })?; trace.capture("upstream.error.body", &raw_body); let error_body = String::from_utf8_lossy(&raw_body).into_owned(); return Err(CodexClientError::Upstream { diff --git a/backend/crates/providers/openai/src/transport/client_sse.rs b/backend/crates/providers/openai/src/transport/client_sse.rs index 1b3143b70..fb8013d19 100644 --- a/backend/crates/providers/openai/src/transport/client_sse.rs +++ b/backend/crates/providers/openai/src/transport/client_sse.rs @@ -24,9 +24,11 @@ use crate::transport::{ headers::websocket_header_pairs, profile::CodexWireProfileState, protocol::{ - responses::{CodexResponsesRequest, TransportRequirement, transport_requirement}, + responses::{ + CodexResponsesRequest, ResponsesSseFailure, TransportRequirement, transport_requirement, + }, websocket::{ - websocket_audit_artifact_from_attempt, websocket_connection_limit_message, + websocket_audit_artifact_from_attempt, websocket_connection_limit_failure, websocket_payload_audit_snapshot, }, }, @@ -146,7 +148,19 @@ impl CodexBackendClient { .get(CONTENT_TYPE) .map(|value| value.as_bytes().to_vec()); let client_headers = response_meta::client_headers(response.headers()); - let raw_body = read_error_response_body(response).await?; + let raw_body = read_error_response_body(response).await.map_err(|source| { + CodexClientError::ErrorBodyRead { + source, + status, + diagnostics: Box::new(diagnostics.clone()), + transport: CodexBackendTransport::HttpSse, + transport_metrics: Box::new(CodexTransportMetrics { + upstream_headers_ms: Some(upstream_headers_ms), + http_version: Some(http_version.clone()), + ..CodexTransportMetrics::default() + }), + } + })?; trace.capture("upstream.error.body", &raw_body); let body = String::from_utf8_lossy(&raw_body).into_owned(); let retry_after_seconds = @@ -424,9 +438,9 @@ impl CodexBackendClient { if requirement.allows_connection_restart() { match await_websocket_delivery_boundary(&mut exchange).await { Ok(DeliveryBoundary::Ready) => {} - Ok(DeliveryBoundary::ConnectionLimitReached) => { + Ok(DeliveryBoundary::ConnectionLimitReached(failure)) => { return Err(CodexClientError::WebSocket( - CodexWebSocketExchangeError::ConnectionLimitReached, + CodexWebSocketExchangeError::ConnectionLimitReached(failure), )); } Err(error) => { @@ -531,7 +545,7 @@ enum DeliveryBoundary { /// 已越过边界，可开始向下游投递。 Ready, /// 首个可投递帧是上游连接寿命限制错误。 - ConnectionLimitReached, + ConnectionLimitReached(Box), } async fn await_websocket_delivery_boundary( @@ -542,14 +556,14 @@ async fn await_websocket_delivery_boundary( match exchange.body.next().await { Some(Ok(frame)) if is_websocket_lifecycle_prelude(&frame) => prelude.push(frame), Some(Ok(frame)) => { - let connection_limit_reached = websocket_connection_limit_message(&frame).is_some(); + let connection_limit_failure = websocket_connection_limit_failure(&frame); prelude.push(frame); let remaining = std::mem::replace(&mut exchange.body, Box::pin(futures::stream::empty())); exchange.body = Box::pin(futures::stream::iter(prelude.into_iter().map(Ok)).chain(remaining)); - return Ok(if connection_limit_reached { - DeliveryBoundary::ConnectionLimitReached + return Ok(if let Some(failure) = connection_limit_failure { + DeliveryBoundary::ConnectionLimitReached(Box::new(failure)) } else { DeliveryBoundary::Ready }); diff --git a/backend/crates/providers/openai/src/transport/diagnostics.rs b/backend/crates/providers/openai/src/transport/diagnostics.rs index 7ed350f66..08992adc8 100644 --- a/backend/crates/providers/openai/src/transport/diagnostics.rs +++ b/backend/crates/providers/openai/src/transport/diagnostics.rs @@ -209,7 +209,10 @@ impl CodexUpstreamFailure { identity_error_code: diagnostics.identity_error_code.clone(), usage_limit_resets_at: None, retry_after_seconds: failure.retry_after_seconds, - request_id: diagnostics.request_id.clone(), + request_id: failure + .request_id + .clone() + .or_else(|| diagnostics.request_id.clone()), set_cookie_headers: set_cookie_headers.to_vec(), rate_limit_headers: rate_limit_headers.to_vec(), send_phase, @@ -326,6 +329,29 @@ pub struct CodexUpstreamDiagnostics { } impl CodexUpstreamDiagnostics { + pub(crate) fn error_event_request_id(value: &Value) -> Option { + let headers = value.get("headers")?.as_object()?; + let mut mapped = HeaderMap::new(); + for (name, value) in headers { + let Ok(name) = reqwest::header::HeaderName::from_bytes(name.as_bytes()) else { + continue; + }; + if !UPSTREAM_REQUEST_ID_HEADERS.contains(&name.as_str()) { + continue; + } + // 与官方 wrapped error 一致，只接受合法 HTTP 标量头；任意对象和数组不作 ID。 + let value = match value { + Value::String(value) => value.clone(), + Value::Number(_) | Value::Bool(_) => value.to_string(), + _ => continue, + }; + if let Ok(value) = reqwest::header::HeaderValue::from_str(&value) { + mapped.insert(name, value); + } + } + first_header(&mapped, UPSTREAM_REQUEST_ID_HEADERS) + } + pub fn from_headers(status_code: Option, headers: &HeaderMap) -> Self { Self { status_code, diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index e664b402c..5e65dfe04 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -404,6 +404,8 @@ pub struct ResponsesSseFailure { pub explicit_status_code: Option, /// 上游显式重试间隔，或从官方限流消息中解析出的重试间隔。 pub retry_after_seconds: Option, + /// 当前错误事件自身携带的请求 ID；不是连接 opening ID。 + pub(crate) request_id: Option, /// 上游错误事件的原始 JSON data；只应在明确的失败审计边界读取。 raw_body: String, } @@ -435,6 +437,10 @@ impl ResponsesSseFailure { upstream_type: failure_type(value), explicit_status_code: failure_explicit_status_code(value), retry_after_seconds: events::retry_after_seconds_from_value(value), + request_id: + crate::transport::diagnostics::CodexUpstreamDiagnostics::error_event_request_id( + value, + ), raw_body: raw_body.to_owned(), } } diff --git a/backend/crates/providers/openai/src/transport/protocol/websocket.rs b/backend/crates/providers/openai/src/transport/protocol/websocket.rs index 703b81876..3447e59aa 100644 --- a/backend/crates/providers/openai/src/transport/protocol/websocket.rs +++ b/backend/crates/providers/openai/src/transport/protocol/websocket.rs @@ -2,7 +2,9 @@ use gateway_protocol::openai::sse::encode_sse_event; use serde::{Deserialize, Serialize}; use serde_json::{Map, Value}; -use crate::transport::protocol::responses::{CodexResponsesRequest, transport_requirement}; +use crate::transport::protocol::responses::{ + CodexResponsesRequest, ResponsesSseFailure, transport_requirement, +}; const REDACTED_PAYLOAD_VALUE: &str = ""; @@ -103,8 +105,8 @@ fn is_websocket_metadata_event(event: Option<&str>) -> bool { pub(crate) const WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE: &str = "websocket_connection_limit_reached"; -/// 若首个可投递 SSE 帧是上游连接寿命限制错误帧，返回其 message。 -pub(crate) fn websocket_connection_limit_message(frame: &[u8]) -> Option { +/// 若首个可投递 SSE 帧是连接寿命限制错误，保留原始失败事实而不改变恢复分类。 +pub(crate) fn websocket_connection_limit_failure(frame: &[u8]) -> Option { let text = std::str::from_utf8(frame).ok()?; let event = gateway_protocol::openai::sse::parse_sse_events(text) .ok()? @@ -116,11 +118,15 @@ pub(crate) fn websocket_connection_limit_message(frame: &[u8]) -> Option let value = serde_json::from_str::(&event.data).ok()?; let code = value.pointer("/error/code").and_then(Value::as_str)?; (code == WEBSOCKET_CONNECTION_LIMIT_REACHED_CODE).then(|| { - value + let mut failure = ResponsesSseFailure::from_raw_event("error", &event.data, &value); + if value .pointer("/error/message") .and_then(Value::as_str) - .unwrap_or("websocket connection limit reached") - .to_owned() + .is_none() + { + failure.message = "websocket connection limit reached".to_owned(); + } + failure }) } diff --git a/backend/crates/providers/openai/src/transport/websocket/error.rs b/backend/crates/providers/openai/src/transport/websocket/error.rs index 907c9037a..d779ebe2d 100644 --- a/backend/crates/providers/openai/src/transport/websocket/error.rs +++ b/backend/crates/providers/openai/src/transport/websocket/error.rs @@ -9,6 +9,7 @@ use uuid::Uuid; use crate::transport::client::CodexClientVisibleUpstreamResponse; use crate::transport::diagnostics::CodexUpstreamDiagnostics; use crate::transport::diagnostics::CodexUpstreamSendPhase; +use crate::transport::protocol::responses::ResponsesSseFailure; use super::PreviousResponseUnavailableReason; use super::pump::WebSocketConnectionObservation; @@ -63,7 +64,7 @@ pub enum CodexWebSocketExchangeError { Upstream(Box), /// 上游要求结束当前 WebSocket 连接并在新连接上重试。 #[error("websocket connection limit reached")] - ConnectionLimitReached, + ConnectionLimitReached(Box), /// 请求依赖的连接本地 previous response 无法在当前连接满足。 #[error("websocket continuation unavailable: {reason}")] ContinuationUnavailable { diff --git a/backend/crates/providers/openai/tests/provider/contract.rs b/backend/crates/providers/openai/tests/provider/contract.rs index 83fc0965f..c7531a38f 100644 --- a/backend/crates/providers/openai/tests/provider/contract.rs +++ b/backend/crates/providers/openai/tests/provider/contract.rs @@ -6413,7 +6413,7 @@ async fn connection_limit_rejection_requests_one_provider_retry_and_reconnects_i let mut ws = accept_codex_test_websocket(stream).await; ws.next().await.expect("request").expect("valid frame"); let event = if first { - json!({"type":"error","status":400,"error":{"code":"websocket_connection_limit_reached","type":"invalid_request_error","message":"connection expired"}}) + json!({"type":"error","status":400,"headers":{"x-request-id":"req-limit-retry"},"error":{"code":"websocket_connection_limit_reached","type":"invalid_request_error","message":"connection expired"}}) } else { json!({"type":"response.completed","response":{"id":"resp_reconnected","model":"gpt-5.4","status":"completed","output":[]}}) }; @@ -6453,6 +6453,23 @@ async fn connection_limit_rejection_requests_one_provider_retry_and_reconnects_i assert!( matches!(error.pre_delivery_retry(), Some(PreDeliveryRetry::SameAccountTransportRetry { retry_index, .. }) if retry_index.get() == 1) ); + assert_eq!(error.upstream_status(), Some(400)); + assert_eq!( + error.upstream_request_id().map(|id| id.as_str()), + Some("req-limit-retry") + ); + let visible = error + .client_visible_upstream_error() + .expect("original limit error"); + assert_eq!(visible.message(), "connection expired"); + assert_eq!(visible.error_type(), Some("invalid_request_error")); + assert!( + error + .raw_upstream_error() + .expect("original frame") + .as_str() + .contains("req-limit-retry") + ); drop(stream); let retry_context = context("req_rejected", CancellationToken::new()).with_transport( AttemptTransport::Retry(NonZeroU32::new(1).expect("retry index")), @@ -6471,3 +6488,381 @@ async fn connection_limit_rejection_requests_one_provider_retry_and_reconnects_i assert!(pooled); server.await.expect("server"); } + +#[tokio::test] +async fn error_body_read_failure_preserves_http_sse_response_facts_without_replay() { + assert_error_body_read_failure(false).await; +} + +#[tokio::test] +async fn error_body_read_failure_preserves_http_json_response_facts_without_replay() { + assert_error_body_read_failure(true).await; +} + +async fn assert_error_body_read_failure(image: bool) { + for status in [401, 429, 503] { + for chunked in [false, true] { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let listener = TcpListener::bind("127.0.0.1:0").await.expect("listener"); + let base_url = format!("http://{}", listener.local_addr().expect("address")); + let server = tokio::spawn(async move { + let (mut stream, _) = listener.accept().await.expect("accept HTTP"); + read_http_request(&mut stream).await; + let framing = if chunked { + "transfer-encoding: chunked" + } else { + "content-length: 4096" + }; + stream + .write_all(format!( + "HTTP/1.1 {status} Error\r\n{framing}\r\ncontent-type: application/json\r\nx-request-id: req-body-truncated\r\nretry-after: 99\r\nconnection: close\r\n\r\n" + ).as_bytes()) + .await + .expect("headers"); + if chunked { + stream.write_all(b"1000\r\n").await.expect("chunk size"); + } + stream + .write_all(b"{\"error\":{\"message\":\"synthetic-private-body") + .await + .expect("partial body"); + stream.shutdown().await.expect("truncate response"); + }); + let provider = provider_with_base_url_and_retry_budget(&store, base_url, 0); + let request = if image { + let payload = RawJsonPayload::new( + "openai", + Bytes::from_static(br#"{"model":"gpt-image-2","prompt":"test"}"#), + ) + .expect("image payload"); + planned_provider_endpoint_request( + "openai", + Operation::GenerateImage(ImageRequest::from_raw_json( + ImageRequestKind::Generation, + payload, + )), + ) + } else { + planned_request("openai", http_generate_operation()) + }; + let mut stream = provider + .execute( + request, + context("req_body_read_failure", CancellationToken::new()), + ) + .await + .expect("prepare"); + let mut observation = None; + let error = timeout(Duration::from_secs(5), async { + loop { + match stream.next().await { + Some(Ok(event)) => { + if let Some(current) = event.response_observation() { + observation = Some(current.clone()); + } + } + Some(Err(error)) => break error, + None => panic!("expected body read failure"), + } + } + }) + .await + .expect("bounded body read"); + server.await.expect("server"); + assert_eq!( + error.upstream_status(), + Some(status), + "image={image}, chunked={chunked}" + ); + assert_eq!( + error.upstream_request_id().map(|id| id.as_str()), + Some("req-body-truncated") + ); + assert_eq!(error.kind(), ProviderErrorKind::Transport); + assert_eq!(error.send_state(), UpstreamSendState::Ambiguous); + assert!(!error.replay_is_safe()); + assert!(error.pre_delivery_retry().is_none()); + assert!(error.retry_after().is_none()); + assert!(error.client_visible_upstream_response().is_none()); + assert!(error.raw_upstream_error().is_none()); + let diagnostic = error.diagnostic().expect("body read diagnostic"); + assert_eq!(diagnostic.stage(), Some("receive")); + assert_eq!(diagnostic.code(), Some("body_read_failed")); + assert!(!diagnostic.as_str().contains("synthetic-private")); + let observation = observation.expect("known HTTP response facts"); + assert_eq!(observation.status_code(), Some(status)); + assert_eq!( + observation.request_id().map(|id| id.as_str()), + Some("req-body-truncated") + ); + } + } +} + +#[tokio::test] +async fn websocket_failure_headers_override_opening_id_without_changing_raw_events() { + assert_websocket_failure_headers( + json!({"X-Request-Id": "req-current-error", "authorization": "synthetic-private-token"}), + Some("req-current-error"), + false, + ) + .await; +} + +#[tokio::test] +async fn websocket_failure_without_valid_headers_does_not_claim_opening_request_id() { + for headers in [ + Value::Null, + json!({}), + json!({"x-request-id": " "}), + json!({"x-request-id": ["invalid"], "x-oai-request-id": "\r\ninvalid"}), + ] { + assert_websocket_failure_headers(headers, None, false).await; + } +} + +#[tokio::test] +async fn reused_websocket_response_failed_uses_current_request_id() { + assert_websocket_failure_headers( + json!({"x-request-id": "", "X-Oai-Request-Id": "req-reused-error"}), + Some("req-reused-error"), + true, + ) + .await; +} + +async fn assert_websocket_failure_headers(headers: Value, request_id: Option<&str>, reuse: bool) { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let listener = TcpListener::bind("127.0.0.1:0").await.expect("listener"); + let base_url = format!("http://{}", listener.local_addr().expect("address")); + let upstream_error = json!({ + "code":"invalid_request","type":"invalid_request_error","message":"synthetic failure" + }); + let raw = if reuse { + json!({ + "type":"response.failed", "status_code":400, "headers":headers, + "response":{"id":"resp-failed","status":"failed","error":upstream_error} + }) + } else { + json!({"type":"error", "status":400, "headers":headers, "error":upstream_error}) + } + .to_string(); + let server_raw = raw.clone(); + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("accept WS"); + let mut ws = crate::transport::accept_codex_test_websocket_with(stream, |_, response| { + response + .headers_mut() + .insert("x-request-id", "req-opening".parse().expect("ID")); + }) + .await; + if reuse { + ws.next() + .await + .expect("first request") + .expect("valid frame"); + ws.send(Message::Text( + json!({ + "type":"response.created","response":{"id":"resp-initial","model":"gpt-5.4"} + }) + .to_string() + .into(), + )) + .await + .expect("created"); + ws.send(Message::Text(json!({ + "type":"response.completed","response":{"id":"resp-initial","model":"gpt-5.4","status":"completed","output":[]} + }).to_string().into())).await.expect("completed"); + } + ws.next().await.expect("request").expect("valid frame"); + ws.send(Message::Text( + json!({ + "type":"response.metadata","headers":{"x-request-id":"req-metadata"} + }) + .to_string() + .into(), + )) + .await + .expect("metadata"); + ws.send(Message::Text(server_raw.into())) + .await + .expect("error frame"); + }); + let provider = provider_with_base_url_and_retry_budget(&store, base_url, 0); + let operation = || { + Operation::Generate(generate_with_persisted_session_context( + "acct_provider_contract", + "conversation-error-headers", + "session-error-headers", + "thread", + )) + }; + if reuse { + let mut first = provider + .execute( + planned_request("openai", operation()), + context("req_ws_initial", CancellationToken::new()), + ) + .await + .expect("prepare first request"); + timeout(Duration::from_secs(5), async { + while let Some(event) = first.next().await { + event.expect("successful first request"); + } + }) + .await + .expect("bounded first request"); + } + let mut stream = provider + .execute( + planned_request("openai", operation()), + context("req_ws_headers", CancellationToken::new()), + ) + .await + .expect("prepare"); + let mut observation = None; + let mut error = timeout(Duration::from_secs(5), async { + loop { + match stream.next().await { + Some(Ok(event)) => { + if let Some(current) = event.response_observation() { + observation = Some(current.clone()); + } + } + Some(Err(error)) => break error, + None => panic!("expected error"), + } + } + }) + .await + .expect("bounded failure"); + server.await.expect("server"); + assert_eq!( + error.upstream_request_id().map(|id| id.as_str()), + request_id + ); + let observation = observation.expect("failure observation"); + assert_eq!(observation.status_code(), Some(400)); + let metadata: Value = serde_json::from_str( + observation + .provider_metadata() + .expect("provider facts") + .as_json(), + ) + .expect("metadata JSON"); + assert_eq!(metadata["websocketOpeningRequestId"], "req-opening"); + assert_eq!(observation.request_id().map(|id| id.as_str()), request_id); + if reuse { + assert_eq!(observation.websocket_pool(), Some(WebSocketPoolKind::Reuse)); + } + assert_eq!(error.upstream_status(), Some(400)); + assert_eq!(error.kind(), ProviderErrorKind::InvalidRequest); + assert_eq!(error.send_state(), UpstreamSendState::Sent); + assert!(!error.replay_is_safe()); + assert_eq!(error.raw_upstream_error().expect("raw error").as_str(), raw); + let events = error.take_atomic_client_events(); + let event_type = if reuse { "response.failed" } else { "error" }; + assert_eq!( + events + .iter() + .filter_map(|event| event.wire_event()?.event_type()) + .collect::>(), + vec![event_type], + "original failure stays atomically deliverable", + ); + let wire = events + .iter() + .find_map(|event| event.wire_event()?.raw_sse_frame()) + .expect("raw frame"); + assert_eq!( + wire.as_ref(), + format!("event: {event_type}\ndata: {raw}\n\n").as_bytes() + ); +} + +#[tokio::test] +async fn connection_limit_payload_survives_exhausted_retry_budget() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let listener = TcpListener::bind("127.0.0.1:0").await.expect("listener"); + let base_url = format!("http://{}", listener.local_addr().expect("address")); + let raw = r#"{ "type":"error", "status":400, "headers":{"x-request-id":"req-limit-final"}, "error":{"code":"websocket_connection_limit_reached","type":"invalid_request_error","message":"synthetic expired connection"}, "retry_after_seconds":99, "future":9007199254740993 }"#; + let server = tokio::spawn(async move { + let (stream, _) = listener.accept().await.expect("accept WS"); + let mut ws = accept_codex_test_websocket(stream).await; + ws.next().await.expect("request").expect("valid frame"); + ws.send(Message::Text(raw.into())) + .await + .expect("limit frame"); + assert!( + timeout(Duration::from_millis(100), listener.accept()) + .await + .is_err(), + "Provider reports recovery intent, never sends a hidden retry" + ); + }); + let provider = provider_with_base_url_and_retry_budget(&store, base_url, 0); + let mut stream = provider + .execute( + planned_request("openai", generate_operation()), + context("req_limit_final", CancellationToken::new()), + ) + .await + .expect("prepare"); + let mut observation = None; + let error = timeout(Duration::from_secs(5), async { + loop { + match stream.next().await { + Some(Ok(event)) => { + assert!(!event.has_client_event(), "lifetime preflight stays atomic"); + if let Some(current) = event.response_observation() { + observation = Some(current.clone()); + } + } + Some(Err(error)) => break error, + None => panic!("expected limit rejection"), + } + } + }) + .await + .expect("bounded rejection"); + server.await.expect("server"); + assert_eq!(error.kind(), ProviderErrorKind::RateLimited); + assert_eq!(error.send_state(), UpstreamSendState::Sent); + assert!(error.replay_is_safe()); + assert!(matches!( + error.pre_delivery_retry(), + Some(PreDeliveryRetry::SameAccountTransportFallback) + )); + assert!( + error.retry_after().is_none(), + "raw retry hint must not alter existing lifetime recovery policy" + ); + assert_eq!(error.upstream_status(), Some(400)); + assert_eq!( + error.upstream_request_id().map(|id| id.as_str()), + Some("req-limit-final") + ); + assert_eq!(error.raw_upstream_error().expect("raw frame").as_str(), raw); + let visible = error + .client_visible_upstream_error() + .expect("original structured error"); + assert_eq!(visible.message(), "synthetic expired connection"); + assert_eq!(visible.error_type(), Some("invalid_request_error")); + assert!( + !error + .diagnostic() + .expect("safe diagnostic") + .as_str() + .contains("synthetic expired") + ); + assert!(error.client_visible_upstream_response().is_none()); + let observation = observation.expect("failure observation"); + assert_eq!(observation.status_code(), Some(400)); + assert_eq!( + observation.request_id().map(|id| id.as_str()), + Some("req-limit-final") + ); +} diff --git a/backend/crates/providers/openai/tests/transport/mod.rs b/backend/crates/providers/openai/tests/transport/mod.rs index ffcf09faf..dd0c260b8 100644 --- a/backend/crates/providers/openai/tests/transport/mod.rs +++ b/backend/crates/providers/openai/tests/transport/mod.rs @@ -103,7 +103,7 @@ pub(crate) async fn accept_codex_test_websocket( .await } -async fn accept_codex_test_websocket_with( +pub(crate) async fn accept_codex_test_websocket_with( stream: TcpStream, callback: F, ) -> tokio_tungstenite::WebSocketStream diff --git a/backend/crates/providers/openai/tests/transport/websocket/mod.rs b/backend/crates/providers/openai/tests/transport/websocket/mod.rs index 800f7c6ff..9e2577527 100644 --- a/backend/crates/providers/openai/tests/transport/websocket/mod.rs +++ b/backend/crates/providers/openai/tests/transport/websocket/mod.rs @@ -1539,7 +1539,7 @@ async fn websocket_connection_limit_should_be_reported_without_transport_retry() .expect_err("transport reports rejection"); assert!(matches!( error, - CodexClientError::WebSocket(CodexWebSocketExchangeError::ConnectionLimitReached) + CodexClientError::WebSocket(CodexWebSocketExchangeError::ConnectionLimitReached(_)) )); server.await.unwrap(); } @@ -2306,10 +2306,12 @@ async fn diagnostics_capture_metadata_and_unknown_events_before_normal_close() { metadata["data"]["metadata"]["headers"]["x-request-id"], "upstream-metadata-request" ); + // 未知事件名可能来自用户内容；仍记录事件，但按默认诊断合同仅保留摘要。 + let unknown_event = gateway_core::diagnostics::body_fingerprint(b"future.metadata"); assert!( events .iter() - .any(|event| event["data"]["eventType"] == "future.metadata") + .any(|event| event["data"]["eventType"] == unknown_event) ); let close = events .iter() diff --git a/deploy/README.md b/deploy/README.md index 61f574656..d43a68c79 100644 --- a/deploy/README.md +++ b/deploy/README.md @@ -282,6 +282,34 @@ OpenAI 主动额度重置卡及其消费结果由上游持有，不写入 Postgr 完整运行时、Provider、revision 与恢复边界见 [架构文档](../docs/architecture.md)。 +## 请求错误排查 + +1. 先记录故障时间和时区、网关版本/提交、客户端名称与版本，以及 HTTP/SSE/WebSocket 传输。 + 区分“上游返回”“网关实际响应”和“客户端终端展示”，不要只凭终端的统一文案推断根因。 +2. 收集响应中的 `x-gateway-request-id`、`x-request-id` / `x-oai-request-id`；配置了 + `api.request_id_header` 时也记录该入口头。WebSocket 合成错误的关联头位于本条错误的 `headers`。 + 在管理端错误列表按 ID 和时间搜索；检查平台条件并主动刷新，翻页不会推进查询时间。 + 已建立模型执行的 HTTP 响应在没有上游 ID 时以模型执行 ID 回传 `x-request-id`， + 不需要新增入口 ID 数据列。旧响应或执行前拒绝若只有入口 ID，改用入口日志和时间定位。 +3. 打开错误详情，核对上游/客户端状态、发送状态、attempt、失败阶段及后续恢复关联。 + 下载默认诊断包作为反馈材料，先看 `availability`、`attemptsComplete` 和淘汰计数； + 该包不含原始错误正文或完整 trace data。分享前仍应检查关联 ID 等内部信息。 +4. 需要更细上下文时，在 `codex-proxy-rs-application.*.log` 及 `.log.gz` 中按网关/上游 ID + 和时间检索，结合 `attempt.started`、`attempt.failed`、`request.finished` 与 transport 阶段判断。 + **开启 `host.logging.file.enabled` 时，stdout 仅保留 `gateway_startup` 通道**； + `docker logs` 看不到业务错误不代表没有错误。关闭普通文件日志且开启 `host.logging.stdout` + 时，普通日志才按级别输出到 stdout；专用 dump/OAuth 恢复通道不会混入。 +5. 查不到请求记录时，检查观测队列丢弃/写入失败告警及 `file_logging` 健康状态。首次合法 + ProviderStream 建立前的失败、鉴权/解析/准入拒绝不保证进入错误列表；应结合入口状态与日志， + 不能据此认定请求未发生。异步投影有延迟，刷新后仍需核对缺口，而不是反复重放可能已发送的请求。 +6. 仅在上述信息不足且能够控制访问范围时临时开启 `host.logging.request_dump` 复现一次。 + 它会记录完整凭据与用户正文，应限定访问、摘取最小片段并人工脱敏；复现后关闭开关， + 按配置的保留窗口及组织的数据处理要求管理已生成文件。不要为普通请求排查开启 OAuth 恢复记录， + 也不要直接上传整个日志目录或完整转储。 + +反馈入口见 [Issue 表单](../.github/ISSUE_TEMPLATE/bug_report.yml)；错误诊断与查询合同见 +[API 文档](../docs/api.md#10-dashboard用量与错误)。 + ## 密码语义 - `admin.default_password` 只在首次创建管理员时使用。 diff --git a/docs/api.md b/docs/api.md index 0063ff958..3e9746583 100644 --- a/docs/api.md +++ b/docs/api.md @@ -152,6 +152,19 @@ Responses WebSocket 仅接受文本 `response.create`，同一连接串行执行 客户端配置的 `supports_websockets` 只控制第一段连接，不是服务端传输策略开关。 上游在响应终态前发送 Close 1000 仍属于失败，不能按“正常关闭”计为成功。 +已建立模型执行的 Responses、Images 和 Search HTTP 响应使用现有 ID： +`x-gateway-request-id` 为模型执行 ID；`x-request-id` 保留有效上游值，只有上游 +`x-oai-request-id` 时复用其值，没有上游 ID 时使用模型执行 ID。`x-oai-request-id` 不是必需字段， +也不要求客户端识别它；OpenAI 与 xAI 路由使用相同规则。失败响应的关联 ID 不采用会话 opening ID， +错误正文读取失败时仍返回已知上游 ID；已采集的 turn state 等允许的会话头继续按原合同交付。 +尚未建立执行的入口拒绝继续使用 middleware 的入口关联。 + +WebSocket 在尚未交付上游业务事件时合成的错误保留已确认的失败状态，以及 Provider 提取的结构化 +message/type/code；没有结构化错误时使用稳定安全文案，不把原始 HTML 或截断正文当作 message。 +合成错误自身的 `headers` 携带允许下发的响应头：优先保留实际失败的上游 request ID，无上游 ID 时 +提供网关关联 ID，并用 `x-gateway-request-id` 独立标识网关请求。已经取得的原始上游错误帧不重写。 +客户端可能对特定状态另行统一展示；这不构成网关改写真实状态码的理由。 + `GET /v1/models` 默认返回 OpenAI 兼容列表 `{"object": "list", "data": [...]}`；请求携带非空 `client_version` query 参数（Codex 客户端）时改为返回 Codex 专用目录合同 `{"models": [...]}`。 @@ -698,6 +711,11 @@ errorCode, errorMessage, startedAt, completedAt, expiresAt, createdAt, updatedAt request/response/upstream ID、outcome 与搜索文本。诊断 `dimension` 可取 `model`、`account`、 `apiKey`、`provider`、`transport`、`failureClass`、`status`。 +请求 ID 和上游 ID 继续通过既有字段查询；不增加入口 ID 字段，也不扫描 trace 建立查询映射。 +旧响应中只有入口 ID 时仍需结合时间与入口日志定位，不能回填不存在的关联。 +管理端搜索完整 `sk_` Client Key 时仅提交其可见前缀，不将完整密钥放入 URL。 +错误列表的主动刷新、搜索和平台/时间条件变化会取得新的结束时间；翻页沿用该次查询快照。 + 汇总与洞察中的请求数与 outcome 分布覆盖筛选范围内全部请求；token、缓存、延迟与成本聚合仅统计 已完整交付客户端的成功响应。 @@ -705,6 +723,14 @@ request/response/upstream ID、outcome 与搜索文本。诊断 `dimension` 可 `relatedRequests[]`（`requestId / relation / outcome / completedAt`）；`relation` 为 `recovered_by` 或 `recovers`。`trace` 是执行终态时的有界脱敏时间线，包含 request、attempt 和 exchange 关联、阶段、 事件摘要及淘汰计数；普通用量列表不携带此字段。 +新采集的未知 JSON 键名与值只保留结构和摘要；事件摘要中的 `eventType` 为已知事件名称字符串、 +未知名称的 `{ bytes, sha256 }` 摘要，或缺失时的 `null`。旧 trace 不做清理或回填， +其中的 `sanitized` 标记不能作为可直接公开的保证。 + +管理端下载的诊断包 `schemaVersion: 2` 用于人工反馈，不是备份或导入格式。它包含关联 ID、错误分类摘要、 +请求与错误事件各自的状态、attempt、时间线阶段和计时；不自动导出 message/raw error、任意 metadata、 +trace event data、请求响应正文和头部。`availability` 与 `omitted` 明示未采集、不完整或主动省略的内容， +`null` 不代表没有发生错误。版本、环境及原始错误片段仍需操作者另行补充并审阅脱敏。 错误记录中的“已自动恢复”表示系统关联到了后续成功请求，不会把原来的失败记录改为成功。 `upstreamSendState = ambiguous` 表示无法确认该次上游执行结果，不代表后续恢复请求失败； diff --git a/docs/architecture.md b/docs/architecture.md index 272e8eada..6d8f9120d 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -376,6 +376,18 @@ credential 与 quota 是两组独立事实：credential refresh 不等于 quota 可恢复观测并累计指标，但不允许改变客户端响应。Usage 详情中的 attempt 因此是 best-effort，并通过 `attemptsComplete: false` 明示不完整性。 +模型执行沿用 `model_requests.id`，上游身份沿用 `upstream_request_id`，不新增入口 ID 列或查询索引。 +API 的 `x-gateway-request-id` 标识模型执行；HTTP `x-request-id` 优先保留上游值，只有上游别名 +`x-oai-request-id` 时复用其值，两者均缺失时回传模型执行 ID。该规则属于客户端协议，不依赖所选 +Provider 或客户端名称。入口 middleware 的 ID 仍可用于入口日志，但不作为模型请求的主键或检索映射。 +失败终态及中间失败的上游请求 ID 优先取对应 `ProviderError`，缺失时再取该 attempt 的响应 +observation、调用 metadata；最终失败的关联头不与 opening 身份混合。HTTP 错误仍按既有合同 +返回已采集的 turn state 等允许的会话头，不为修复 ID 关联丢弃续接状态。 + +请求行仍在首次合法 ProviderStream 建立后随 attempt 合并创建。建流前失败和未接纳为模型执行的 +入口拒绝不保证出现在错误列表；排障需结合入口日志，不能将“无记录”解释成没有发生错误。 +扩展此边界前必须先补齐真实 attempt 的账号、传输和发送事实，不能伪造零尝试或 `not_sent`。 + 只有完整交付客户端的成功响应进入 Token、延迟和成本聚合。实际 `service_tier` 只接受上游响应事件确认， 不能用请求期望值替代。 @@ -417,6 +429,9 @@ HTTP Client 构造失败也不会阻断网关启动。外部解析在已认证 - Provider 将安全诊断的阶段、原因码与消息独立于错误大类和发送状态传入 Core； `attempt.failed` 保存这些字段，最终错误记录优先持久化诊断消息。重试包装不能覆盖底层诊断， 也不能因为错误更详细而改变已有重试安全边界。 +- 默认诊断区分 transport 采集的真实头部与用户 JSON；未知 map 的键和值均不原样保留， + 嵌套的同名 header 不获得头部白名单权限。管理端反馈包只导出明确允许的关联、分类、阶段和计时， + 不复制 trace 的事件 data，也不因历史 `sanitized` 标记而信任旧正文。 - 真实 secret 不进入普通日志、Debug、fixture 或 audit details；明文只能通过账号导出、Key reveal、 备份设置等明确的敏感 Admin 合同返回。 - OAuth pending flow 使用有期限、带 owner 的一次性 claim；事务成功后才消费，失败释放 claim。 diff --git a/frontend/src/views/usage/components/OpsErrorDetailModal.vue b/frontend/src/views/usage/components/OpsErrorDetailModal.vue index 569e27b74..9d230b1a4 100644 --- a/frontend/src/views/usage/components/OpsErrorDetailModal.vue +++ b/frontend/src/views/usage/components/OpsErrorDetailModal.vue @@ -130,7 +130,7 @@ function visibleFields(items: DetailField[]) { size="xl" >