fix: improve error diagnostics and request correlation

Preserve confirmed upstream error details and correlate responses with existing request IDs without schema changes.
Harden diagnostic sanitization and export allowlisted feedback bundles.
Refresh usage and error queries with consistent filters and pagination snapshots.
Add protocol and privacy regressions and document troubleshooting boundaries.

Validated 2064 backend tests with dedicated PostgreSQL and Redis, strict Clippy, Rustfmt, frontend lint and build, and real OpenAI HTTP/WebSocket success and failure paths with 81 UI assertions.

Co-authored-by: Codex <noreply@openai.com>
This commit is contained in:
Zyy
2026-09-12 17:53:03 +08:00
co-authored by Codex
parent bfca754714
commit d3acb42d37
45 changed files with 2538 additions and 257 deletions
+11 -2
View File
@@ -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
@@ -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<dyn ExecutionSession>) -> Response {
let mut execution = PendingExecution::new(session);
let Some(session) = execution.session_mut() else {
return invalid_upstream_response();
};
@@ -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 共用。
@@ -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
}
@@ -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 后提交下游,再持续驱动同一执行会话。
@@ -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<str>,
) -> 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<str>,
) -> ForwardOutcome {
if connection
.send_text(
error_event(status, error_type, code, message, param, Some(request_id)),
message,
WriteContext::request(request_id, FramePhase::Error),
)
.await
@@ -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<String, Value> {
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<String, Value>) -> 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, Value>,
) -> 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(),
)
}
@@ -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",
@@ -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"))
@@ -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<String> {
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<Option<Box<dyn ExecutionSession>>>,
}
impl ExecutionService for SessionExecution {
fn authenticate(&self, _: &str) -> Result<AuthenticatedClient, ClientAuthenticationError> {
Ok(self.client.clone())
}
fn public_models(&self, _: &AuthenticatedClient) -> Vec<PublicModelId> {
Vec::new()
}
fn contains_public_model(&self, _: &AuthenticatedClient, _: &PublicModelId) -> bool {
true
}
fn start(
&self,
request: StartExecution,
) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
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<StartedExecution, GatewayError>> {
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<ProviderResponseHeader> {
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"
);
}
}
}
@@ -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<MaybeTlsStream<TcpStream>>;
// 全部使用合成响应事实和真实 API 路由,不读取账号、日志或请求转储。
struct TestServer(tokio::task::JoinHandle<()>);
impl Drop for TestServer {
fn drop(&mut self) {
self.0.abort();
}
}
async fn connect(
trace: Arc<AtomicFailureTrace>,
response_headers: Vec<ProviderResponseHeader>,
) -> (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<ProviderResponseHeader>) -> 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<ProviderResponseHeader>,
) -> 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"<html>synthetic-private-body-marker</html>".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();
}
}
@@ -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<Vec<String>>,
initial_errors: Mutex<std::collections::VecDeque<EngineError>>,
first_batch: Mutex<Option<CoordinatedEvent>>,
}
struct AtomicFailureSession {
@@ -319,6 +322,10 @@ impl ExecutionSession for AtomicFailureSession {
fn next_event(&mut self) -> BoxFuture<'_, Result<Option<CoordinatedEvent>, 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(
@@ -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"))
@@ -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::<Vec<_>>()
}),
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"
)
}
@@ -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,
);
@@ -230,11 +230,22 @@ struct CurrentAttempt {
response_observation: Option<ProviderResponseObservation>,
}
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<u16>,
upstream_request_id: Option<String>,
provider_error_code: Option<String>,
raw_upstream_error: Option<String>,
retry_after_ms: Option<u64>,
@@ -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()));
@@ -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("<redacted>"));
let headers = diagnostic_headers([(name, "PRIVATE_CREDENTIAL")]);
assert!(!headers.to_string().contains("PRIVATE_"));
assert_eq!(headers[name.to_ascii_lowercase()][0], "<redacted>");
}
}
@@ -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);
}
@@ -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<Mutex<Vec<String>>>);
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();
}
}
@@ -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())]
);
}
@@ -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);
}
@@ -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<ProviderDiagnostic> {
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<RawUpstreamError> {
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<RawUpstrea
pub(super) fn websocket_client_visible_error(
error: &CodexWebSocketExchangeError,
) -> Option<ClientVisibleUpstreamError> {
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 { .. }
@@ -88,15 +88,32 @@ impl OpenAiResponseObservationState {
}
}
pub(super) fn observation(&self) -> Option<ProviderResponseObservation> {
pub(super) fn observation(
&self,
failure: Option<&ProviderError>,
) -> Option<ProviderResponseObservation> {
// 失败快照的请求 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<ProviderResponseMetadata> {
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()));
}
@@ -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<CodexUpstreamDiagnostics>,
transport: CodexBackendTransport,
transport_metrics: Box<CodexTransportMetrics>,
},
/// 自定义 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(_)
@@ -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 {
@@ -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<ResponsesSseFailure>),
}
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
});
@@ -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<String> {
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<u16>, headers: &HeaderMap) -> Self {
Self {
status_code,
@@ -404,6 +404,8 @@ pub struct ResponsesSseFailure {
pub explicit_status_code: Option<u16>,
/// 上游显式重试间隔,或从官方限流消息中解析出的重试间隔。
pub retry_after_seconds: Option<u64>,
/// 当前错误事件自身携带的请求 ID;不是连接 opening ID。
pub(crate) request_id: Option<String>,
/// 上游错误事件的原始 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(),
}
}
@@ -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 = "<redacted>";
@@ -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<String> {
/// 若首个可投递 SSE 帧是连接寿命限制错误,保留原始失败事实而不改变恢复分类。
pub(crate) fn websocket_connection_limit_failure(frame: &[u8]) -> Option<ResponsesSseFailure> {
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<String>
let value = serde_json::from_str::<Value>(&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
})
}
@@ -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<CodexWebSocketUpstreamError>),
/// 上游要求结束当前 WebSocket 连接并在新连接上重试。
#[error("websocket connection limit reached")]
ConnectionLimitReached,
ConnectionLimitReached(Box<ResponsesSseFailure>),
/// 请求依赖的连接本地 previous response 无法在当前连接满足。
#[error("websocket continuation unavailable: {reason}")]
ContinuationUnavailable {
@@ -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<_>>(),
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")
);
}
@@ -103,7 +103,7 @@ pub(crate) async fn accept_codex_test_websocket(
.await
}
async fn accept_codex_test_websocket_with<F>(
pub(crate) async fn accept_codex_test_websocket_with<F>(
stream: TcpStream,
callback: F,
) -> tokio_tungstenite::WebSocketStream<TcpStream>
@@ -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()
+28
View File
@@ -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` 只在首次创建管理员时使用。
+26
View File
@@ -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` 表示无法确认该次上游执行结果,不代表后续恢复请求失败;
+15
View File
@@ -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。
@@ -130,7 +130,7 @@ function visibleFields(items: DetailField[]) {
size="xl"
>
<template v-if="record">
<RequestDiagnosticsPanel v-if="open && record.requestId" class="mb-3" :request-id="record.requestId" :metadata="record.metadata" />
<RequestDiagnosticsPanel v-if="open && record.requestId" class="mb-3" :request-id="record.requestId" :metadata="record.metadata" :error-record="record" />
<section :class="panelClass">
<h3 :class="panelTitleClass">
错误
@@ -18,10 +18,14 @@ import UsageClientIpCell from './UsageClientIpCell.vue'
const props = defineProps<{
timeRangeParams: UsageTimeRangeParams
latestTimeRangeParams: () => UsageTimeRangeParams
provider: string
active: boolean
}>()
const {
loading,
error,
refreshing,
records,
searchQuery,
@@ -29,7 +33,12 @@ const {
handlePageChange,
handlePageSizeChange,
refresh,
} = useOpsErrorsTable(toRef(props, 'timeRangeParams'))
} = useOpsErrorsTable({
timeRangeParams: toRef(props, 'timeRangeParams'),
latestTimeRangeParams: () => props.latestTimeRangeParams(),
provider: toRef(props, 'provider'),
active: toRef(props, 'active'),
})
const selectedRecord = shallowRef<OpsError | null>(null)
const detailOpen = shallowRef(false)
@@ -82,7 +91,8 @@ function upstreamSendStateText(value: string | null | undefined) {
<div class="min-w-0 flex-1">
<BaseInput
v-model="searchQuery"
placeholder="请求 ID、Key 或账号"
placeholder="请求 ID、Key ID / 可见前缀或账号"
aria-label="搜索错误:请求 ID、Key ID / 可见前缀或账号"
class="min-w-0 w-full lg:max-w-96"
>
<template #prefix>
@@ -109,7 +119,11 @@ function upstreamSendStateText(value: string | null | undefined) {
</div>
<div class="flex min-h-0 min-w-0 flex-col">
<p v-if="error && !loading" role="alert" class="text-cp-sm text-cp-error-text">
{{ error }}。请刷新重试。
</p>
<BaseTable
v-else
class="min-h-0 flex-1"
:columns="opsErrorColumns"
:rows="records"
@@ -1,5 +1,5 @@
<script setup lang="ts">
import type { OpsErrorMetadata } from '@/api'
import type { OpsError, OpsErrorMetadata } from '@/api'
import { Download, RefreshCw } from '@lucide/vue'
import { computed } from 'vue'
import BaseButton from '@/components/base/BaseButton.vue'
@@ -7,10 +7,11 @@ import BaseScrollbar from '@/components/base/BaseScrollbar.vue'
import { useAsyncAction } from '@/composables/useAsyncAction'
import { useDownload } from '@/composables/useDownload'
import { useRequestDiagnostics } from '../composables/useRequestDiagnostics'
import { requestDiagnosticsBundle } from '../utils/diagnosticsBundle'
import RequestTransportFailure from './RequestTransportFailure.vue'
import UsageDetailCodePanel from './UsageDetailCodePanel.vue'
const props = defineProps<{ requestId: string, metadata?: OpsErrorMetadata }>()
const props = defineProps<{ requestId: string, metadata?: OpsErrorMetadata, errorRecord?: OpsError }>()
const { downloadJson } = useDownload()
const { loading: exporting, run: runExport } = useAsyncAction()
const { selectedId, detail, loading, error, refresh } = useRequestDiagnostics(() => props.requestId)
@@ -19,35 +20,15 @@ const events = computed(() => (trace.value?.events ?? []).map(event => ({
...event,
content: JSON.stringify(event.data, null, 2),
})))
const selectedError = computed(() => props.errorRecord?.requestId === selectedId.value ? props.errorRecord : undefined)
const canExport = computed(() => !loading.value && (detail.value?.requestId === selectedId.value || !!selectedError.value))
function download() {
if (!detail.value || exporting.value)
if (!canExport.value || exporting.value)
return
const record = detail.value
const bundle = {
schemaVersion: 1,
exportedAt: new Date().toISOString(),
request: {
requestId: record.requestId,
route: record.route,
model: record.model,
provider: record.provider,
accountId: record.accountId,
clientTransport: record.clientTransport,
upstreamTransport: record.upstreamTransport,
upstreamRequestId: record.upstreamRequestId,
responseId: record.responseId,
statusCode: record.statusCode,
attemptCount: record.attemptCount,
createdAt: record.createdAt,
},
trace: record.trace,
attempts: record.attempts,
attemptsComplete: record.attemptsComplete,
relatedRequests: record.relatedRequests,
}
const bundle = requestDiagnosticsBundle(selectedId.value, detail.value, selectedError.value)
return runExport(
() => downloadJson(bundle, `diagnostics-${record.requestId.replace(/[^\w-]/g, '_')}.json`),
() => downloadJson(bundle, `diagnostics-${selectedId.value.replace(/[^\w-]/g, '_')}.json`),
{ errorText: '导出诊断包失败', minimumMs: 400 },
)
}
@@ -69,24 +50,34 @@ function download() {
</template>
刷新
</BaseButton>
<BaseButton variant="soft" size="sm" :loading="exporting" :disabled="!detail" @click="download">
<BaseButton variant="soft" size="sm" :loading="exporting" :disabled="!canExport" @click="download">
<template #icon>
<Download :size="14" />
</template>
导出诊断包
导出安全诊断包
</BaseButton>
</div>
</div>
<p class="mt-1 mb-3 break-all font-mono text-cp-xs leading-relaxed text-cp-text-secondary">
{{ selectedId }}
</p>
<p class="mb-3 text-cp-xs leading-relaxed text-cp-text-secondary">
诊断包 v2 仅含关联字段、错误分类、尝试与时间线阶段;不含错误原文、正文、用户内容或事件 data。版本与环境需另附,分享前请审阅关联 ID。
</p>
<p v-if="!loading && !selectedError" class="text-cp-xs leading-relaxed text-cp-text-secondary">
当前请求未提供错误详情摘要,导出不会沿用其他请求的错误;可结合尝试记录排查。
</p>
<p v-if="loading" role="status" class="text-cp-sm text-cp-text-secondary">
正在加载诊断记录…
</p>
<p v-else-if="error" role="alert" class="text-cp-sm text-cp-error-text">
{{ error }}
<span v-if="selectedError">仍可导出本次错误的分类摘要和关联字段,缺失诊断会在包内标明。</span>
</p>
<template v-else-if="detail">
<p v-if="!detail.attemptsComplete" role="status" class="text-cp-xs leading-relaxed text-cp-warning-text">
尝试记录可能不完整,诊断包会保留此缺口,不能仅按列表条数判断实际尝试次数。
</p>
<div v-if="detail.relatedRequests?.length" class="mb-3 flex flex-wrap gap-2">
<BaseButton v-for="related in detail.relatedRequests" :key="related.requestId" variant="soft" size="sm" class="max-w-full" @click="selectedId = related.requestId">
{{ related.relation === 'recovered_by' ? '查看恢复请求' : '查看先前失败' }} · {{ related.requestId }}
@@ -99,7 +90,7 @@ function download() {
<template v-else>
<div class="mb-3 grid gap-1 text-cp-xs leading-relaxed">
<p class="m-0 text-cp-text-secondary">
已观测 {{ trace.totalEvents }} 个阶段或事件,展示 {{ events.length }} 条摘要。连续增量合并计数;正文和敏感字段仅保留摘要。
已观测 {{ trace.totalEvents }} 个阶段或事件,展示 {{ events.length }} 条记录。连续增量合并计数;历史事件内容可能未充分脱敏,不会自动导出。
</p>
<p v-if="trace.droppedEvents" role="status" class="m-0 text-cp-warning-text">
达到保存上限,{{ trace.droppedEvents }} 个事件已被淘汰;保留请求开头与最近事件。
@@ -124,7 +115,7 @@ function download() {
</ol>
</BaseScrollbar>
<p class="mt-3 mb-0 text-cp-xs leading-relaxed text-cp-text-secondary">
时间线与导出诊断包均为摘要,保存至执行结束。完整报文如已启用,保存在服务器日志中,至少保留最近 24 小时;可用请求 ID 检索。
界面中的诊断事实不等于默认导出内容。原始错误或完整报文如确需反馈,请先单独审阅脱敏;可用请求 ID 在服务器日志中检索。
</p>
</template>
</template>
@@ -19,7 +19,7 @@ const search = defineModel<string>('search', { required: true })
<template>
<div class="flex w-full items-center gap-3" role="group" aria-label="使用记录筛选与操作">
<div class="min-w-0 flex-1 sm:w-96 sm:flex-none">
<BaseInput v-model="search" placeholder="请求、API Key、账号或模型 ID 前缀" class="w-full">
<BaseInput v-model="search" placeholder="请求、Key ID / 可见前缀、账号或模型" aria-label="搜索使用记录:请求、Key ID / 可见前缀、账号或模型" class="w-full">
<template #prefix>
<Search class="size-4.5 text-cp-text-tertiary" />
</template>
@@ -2,28 +2,45 @@ import type { Ref } from 'vue'
import type { UsageTimeRangeParams } from './useUsageTimeRange'
import { watchDebounced } from '@vueuse/core'
import { computed, onMounted, shallowRef, watch } from 'vue'
import { computed, onScopeDispose, shallowRef, watch } from 'vue'
import { getOpsErrors } from '@/api'
import { toast } from '@/components/base/BaseToast'
import { useStablePagedQuery } from '@/composables/useStablePagedQuery'
import { errorMessage, withMinimumDuration } from '@/utils/async'
import { usageSearchParam } from '../utils/search'
export function useOpsErrorsTable(timeRangeParams: Readonly<Ref<UsageTimeRangeParams>>) {
interface UseOpsErrorsTableOptions {
timeRangeParams: Readonly<Ref<UsageTimeRangeParams>>
latestTimeRangeParams: () => UsageTimeRangeParams
provider: Readonly<Ref<string>>
active: Readonly<Ref<boolean>>
}
export function useOpsErrorsTable(options: UseOpsErrorsTableOptions) {
const refreshing = shallowRef(false)
const searchQuery = shallowRef('')
const search = computed(() => usageSearchParam(searchQuery.value))
let disposed = false
// 时间、平台和搜索共同构成分页快照,避免翻页混入另一组筛选结果。
let tableParams = snapshot()
function snapshot() {
return {
...options.latestTimeRangeParams(),
provider: options.provider.value || undefined,
search: search.value,
}
}
const query = useStablePagedQuery({
initialPageSize: 10,
load: ({ currentPage, pageSize }) => getOpsErrors({
currentPage,
pageSize,
search: searchQuery.value.trim() || undefined,
...timeRangeParams.value,
...tableParams,
}),
onError: error => toast.error(errorMessage(error, '加载错误明细失败')),
})
// 首屏进入即处于加载态;首次 execute 完成后由 usePagedQuery 维护。
query.loading.value = true
const pagination = computed(() => ({
currentPage: query.currentPage.value,
pageSize: query.pageSize.value,
@@ -31,12 +48,26 @@ export function useOpsErrorsTable(timeRangeParams: Readonly<Ref<UsageTimeRangePa
}))
function handlePageChange(nextPage: number) {
if (tableParams.search !== search.value) {
void reloadLatest()
return
}
void query.execute(nextPage)
}
function handlePageSizeChange(nextPageSize: number) {
query.pageSize.value = nextPageSize
void query.reloadFromStart()
if (tableParams.search !== search.value)
void reloadLatest()
else
void query.reloadFromStart()
}
function reloadLatest() {
tableParams = snapshot()
// 筛选失败不能继续展示上一范围的数据,错误态由面板单独呈现。
query.items.value = []
return query.reloadFromStart()
}
async function refresh() {
@@ -44,7 +75,7 @@ export function useOpsErrorsTable(timeRangeParams: Readonly<Ref<UsageTimeRangePa
return
refreshing.value = true
try {
await withMinimumDuration(() => query.execute(query.currentPage.value))
await withMinimumDuration(reloadLatest)
}
finally {
refreshing.value = false
@@ -52,21 +83,26 @@ export function useOpsErrorsTable(timeRangeParams: Readonly<Ref<UsageTimeRangePa
}
watchDebounced(
searchQuery,
search,
() => {
void query.reloadFromStart()
if (!disposed && options.active.value && tableParams.search !== search.value)
void reloadLatest()
},
{ debounce: 250 },
)
watch(timeRangeParams, () => {
void query.reloadFromStart()
})
watch([options.timeRangeParams, options.provider, options.active], () => {
if (options.active.value)
void reloadLatest()
else
query.invalidate()
}, { immediate: true })
onMounted(() => void query.execute(1))
onScopeDispose(() => disposed = true)
return {
loading: query.loading,
error: query.error,
refreshing,
records: query.items,
searchQuery,
@@ -13,16 +13,16 @@ import {
import { toast } from '@/components/base/BaseToast'
import { errorMessage, withMinimumDuration } from '@/utils/async'
import { usageSearchParam } from '../utils/search'
interface UseUsageRecordsTableOptions {
timeRangeParams: Readonly<Ref<UsageTimeRangeParams>>
latestTimeRangeParams: () => UsageTimeRangeParams
active: Readonly<Ref<boolean>>
}
type UsageLoadScope = 'all' | 'table'
const CLIENT_API_KEY_VISIBLE_PREFIX_LENGTH = 10
interface UsageLoadOptions {
scope?: UsageLoadScope
background?: boolean
@@ -38,28 +38,33 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
const pageSize = shallowRef(10)
const totalRecords = shallowRef(0)
const searchQuery = shallowRef('')
const search = computed(() => usageSearchParam(searchQuery.value))
const providerQuery = shallowRef('')
const tableTimeRangeParams = shallowRef<UsageTimeRangeParams>({
...options.timeRangeParams.value,
})
let tableParams = snapshot()
const refreshingList = shallowRef(false)
const diagnosticDimension = shallowRef('model')
let tableRequestId = 0
let analyticsRequestId = 0
let diagnosticRequestId = 0
let disposed = false
const scopedParams = () => ({
...options.timeRangeParams.value,
...(providerQuery.value ? { provider: providerQuery.value } : {}),
})
const filterParams = () => ({
search: usageSearchParam(searchQuery.value),
})
const usagePagination = computed(() => ({
currentPage: currentPage.value,
pageSize: pageSize.value,
total: totalRecords.value,
}))
function snapshot() {
return {
...options.latestTimeRangeParams(),
provider: providerQuery.value || undefined,
search: search.value,
}
}
function resetPagination() {
currentPage.value = 1
totalRecords.value = 0
@@ -70,11 +75,11 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
const globalParams = scopedParams()
if (scope === 'all') {
resetPagination()
tableTimeRangeParams.value = { ...globalParams }
tableParams = snapshot()
}
await Promise.all([
loadUsagePage(background),
...(options.active.value ? [loadUsagePage(background)] : []),
...(scope === 'all' ? [loadUsageAnalytics(globalParams, background)] : []),
])
}
@@ -86,8 +91,7 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
const result = await getUsageRecords({
currentPage: currentPage.value,
pageSize: pageSize.value,
...tableTimeRangeParams.value,
...filterParams(),
...tableParams,
})
if (requestId !== tableRequestId)
return
@@ -175,22 +179,34 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
return
refreshingList.value = true
try {
tableTimeRangeParams.value = options.latestTimeRangeParams()
resetPagination()
await withMinimumDuration(() => loadUsageRecords({ scope: 'table' }))
await withMinimumDuration(reloadLatestTable)
}
finally {
refreshingList.value = false
}
}
function reloadLatestTable() {
tableParams = snapshot()
resetPagination()
return loadUsageRecords({ scope: 'table' })
}
function handlePageChange(nextPage: number) {
if (tableParams.search !== search.value) {
void reloadLatestTable()
return
}
currentPage.value = nextPage
void loadUsageRecords({ scope: 'table' })
}
function handlePageSizeChange(nextPageSize: number) {
pageSize.value = nextPageSize
if (tableParams.search !== search.value) {
void reloadLatestTable()
return
}
resetPagination()
void loadUsageRecords({ scope: 'table' })
}
@@ -207,16 +223,24 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
void loadUsageRecords({ background: true })
})
watch(options.active, (active) => {
if (active)
void reloadLatestTable()
else
tableRequestId += 1
})
watchDebounced(
searchQuery,
search,
() => {
resetPagination()
void loadUsageRecords({ scope: 'table' })
if (!disposed && options.active.value && tableParams.search !== search.value)
void reloadLatestTable()
},
{ debounce: 250 },
)
onScopeDispose(() => {
disposed = true
tableRequestId += 1
analyticsRequestId += 1
diagnosticRequestId += 1
@@ -242,15 +266,6 @@ export function useUsageRecordsTable(options: UseUsageRecordsTableOptions) {
}
}
function usageSearchParam(value: string) {
const search = value.trim()
if (!search)
return undefined
if (search.startsWith('sk_'))
return search.slice(0, CLIENT_API_KEY_VISIBLE_PREFIX_LENGTH)
return search
}
function emptySummary() {
const summary: Awaited<ReturnType<typeof getUsageRecordSummary>> = {
totalRequests: '0',
+8 -2
View File
@@ -1,6 +1,6 @@
<script setup lang="ts">
import { Eye } from '@lucide/vue'
import { shallowRef, watch } from 'vue'
import { computed, shallowRef, watch } from 'vue'
import BaseCard from '@/components/base/BaseCard.vue'
import BaseIconButton from '@/components/base/BaseIconButton.vue'
@@ -47,6 +47,7 @@ const {
} = useUsageRecordsTable({
timeRangeParams,
latestTimeRangeParams,
active: computed(() => recordView.value === 'success'),
})
const { showDetailModal, selectedUsageRecord, handleViewDetail } = useUsageRecordDetail()
@@ -141,7 +142,12 @@ watch(timeRange, () => {
</div>
<div v-show="recordView === 'errors'" class="min-h-130 min-w-0 flex-1">
<OpsErrorPanel :time-range-params="timeRangeParams" />
<OpsErrorPanel
:time-range-params="timeRangeParams"
:latest-time-range-params="latestTimeRangeParams"
:provider="providerQuery"
:active="recordView === 'errors'"
/>
</div>
</template>
</BaseCard>
@@ -0,0 +1,130 @@
import type { OpsError, UsageRecordDetail } from '@/api'
import { opsErrorSummary } from './opsErrorPresentation'
export function requestDiagnosticsBundle(
requestId: string,
detail: UsageRecordDetail | null,
errorRecord?: OpsError,
) {
// 关联请求切换后,不能把原弹窗的错误归到另一个请求。
const record = detail?.requestId === requestId ? detail : null
const error = errorRecord?.requestId === requestId ? errorRecord : null
const trace = record?.trace
// 请求终态与错误事件可能来自不同尝试,不用事件的值填补终态空字段。
const request = record ?? error
return {
schemaVersion: 2,
exportedAt: new Date().toISOString(),
source: 'usage.request_diagnostics',
exportPolicy: 'allowlisted_fields_without_payloads',
request: {
requestId,
source: record ? 'request_detail' : error ? 'error_record' : 'request_id_only',
route: request?.route ?? null,
provider: request?.provider ?? null,
accountId: request?.accountId ?? null,
clientTransport: request?.clientTransport ?? null,
upstreamTransport: record ? record.upstreamTransport : error?.transport ?? null,
upstreamRequestId: request?.upstreamRequestId ?? null,
responseId: request?.responseId ?? null,
statusCode: record?.statusCode ?? null,
clientStatusCode: request?.clientStatusCode ?? null,
upstreamStatusCode: request?.upstreamStatusCode ?? null,
logicalOutcome: record?.logicalOutcome ?? null,
attemptCount: record?.attemptCount ?? null,
createdAt: request?.createdAt ?? null,
},
error: error
? {
id: error.id,
requestId: error.requestId,
source: error.metadata.source,
component: error.metadata.component,
summary: opsErrorSummary(error),
summarySource: 'failure_class_or_provider_error_code',
failureClass: error.failureClass,
providerErrorCode: error.providerErrorCode,
upstreamSendState: error.upstreamSendState,
attemptIndex: error.attemptIndex,
clientStatusCode: error.clientStatusCode,
upstreamStatusCode: error.upstreamStatusCode,
upstreamRequestId: error.upstreamRequestId,
responseId: error.responseId,
latencyMs: error.latencyMs,
createdAt: error.createdAt,
}
: null,
// 旧 trace 的 sanitized 标记不是安全保证;不遍历或复制任何 event.data。
trace: trace
? {
schemaVersion: trace.schemaVersion,
startedAtUnixMs: trace.startedAtUnixMs,
totalEvents: trace.totalEvents,
droppedEvents: trace.droppedEvents,
events: trace.events.map(event => ({
sequence: event.sequence,
lastSequence: event.lastSequence,
elapsedMs: event.elapsedMs,
lastElapsedMs: event.lastElapsedMs,
attemptIndex: event.attemptIndex,
exchangeId: event.exchangeId,
stage: event.stage,
count: event.count,
})),
}
: null,
attempts: record?.attempts.map(attempt => ({
id: attempt.id,
attemptIndex: attempt.attemptIndex,
trigger: attempt.trigger,
provider: attempt.provider,
transport: attempt.transport,
sendState: attempt.sendState,
outcome: attempt.outcome,
downstreamCommitted: attempt.downstreamCommitted,
statusCode: attempt.statusCode,
providerErrorCode: attempt.providerErrorCode,
failureClass: attempt.failureClass,
accountId: attempt.accountId,
firstTokenMs: attempt.firstTokenMs,
latencyMs: attempt.latencyMs,
startedAt: attempt.startedAt,
completedAt: attempt.completedAt,
})) ?? [],
relatedRequests: record?.relatedRequests.map(related => ({
requestId: related.requestId,
relation: related.relation,
outcome: related.outcome,
completedAt: related.completedAt,
})) ?? [],
availability: {
requestDetail: record ? 'available' : 'unavailable',
errorSummary: error ? 'available' : 'not_available_for_selected_request',
trace: !record ? 'unknown' : trace ? 'stages_only' : 'not_recorded',
traceEventsDropped: trace?.droppedEvents ?? null,
attemptsComplete: record?.attemptsComplete ?? null,
relatedRequests: record ? 'available' : 'unknown',
environment: 'not_collected',
},
omitted: [
'error.message',
'error.rawUpstreamError',
'request.message',
'request.metadata',
'trace.events.data',
'request_and_response_headers_and_bodies',
'user_supplied_model_names',
'account_names_emails_and_credential_names',
'client_ip_user_agent_and_api_key',
],
notes: [
'schemaVersion 2 仅供人工排障;null 表示未知或未采集,不代表没有发生错误。',
'错误摘要复用列表的分类展示,不是上游错误原文;error 为 null 时请结合 attempts 的分类与状态。',
'时间线仅导出阶段、顺序和计时,不含事件 data;历史 sanitized 标记不作为安全依据。',
'attemptsComplete 不为 true 时,尝试列表不完整;没有时间线的旧记录无法补回未采集事件。',
'未自动采集网关版本、客户端版本、部署环境和故障发生时区,请另行补充。',
'关联 ID 仍可能属于内部信息,分享前请审阅;原始错误、正文和日志需另行审阅脱敏,勿直接公开。',
],
}
}
+10
View File
@@ -0,0 +1,10 @@
const CLIENT_API_KEY_VISIBLE_PREFIX_LENGTH = 10
export function usageSearchParam(value: string) {
const search = value.trim()
if (!search)
return undefined
if (search.startsWith('sk_'))
return search.slice(0, CLIENT_API_KEY_VISIBLE_PREFIX_LENGTH)
return search
}