chore: bump version to 3.3.23

Co-authored-by: Codex <noreply@openai.com>

backend/crates/gateway-api/src/admin/accounts/presenter.rs

@@ -230,6 +230,8 @@ pub(super) fn account_usage_view(
         output_tokens_display: display_optional_tokens(usage.output_tokens),
         cached_tokens: usage.cached_tokens,
         cached_tokens_display: display_optional_tokens(usage.cached_tokens),
+        reasoning_tokens: usage.reasoning_tokens,
+        reasoning_tokens_display: display_optional_tokens(usage.reasoning_tokens),
         image_input_tokens: usage.image_input_tokens,
         image_input_tokens_display: display_optional_tokens(usage.image_input_tokens),
         image_output_tokens: usage.image_output_tokens,
@@ -347,6 +349,8 @@ pub(super) fn empty_account_usage() -> AccountUsageView {
         output_tokens_display: "—".to_owned(),
         cached_tokens: None,
         cached_tokens_display: "—".to_owned(),
+        reasoning_tokens: None,
+        reasoning_tokens_display: "—".to_owned(),
         image_input_tokens: None,
         image_input_tokens_display: "—".to_owned(),
         image_output_tokens: None,

backend/crates/gateway-api/src/admin/accounts/wire.rs

@@ -276,6 +276,8 @@ pub struct AccountUsageView {
     pub output_tokens_display: String,
     pub cached_tokens: Option<u64>,
     pub cached_tokens_display: String,
+    pub reasoning_tokens: Option<u64>,
+    pub reasoning_tokens_display: String,
     pub image_input_tokens: Option<u64>,
     pub image_input_tokens_display: String,
     pub image_output_tokens: Option<u64>,

backend/crates/gateway-api/src/openai/responses/http.rs

@@ -14,8 +14,8 @@ use axum::{
     response::{IntoResponse, Response},
 };
 use futures::stream;
-use gateway_core::engine::CommitRequirement;
 use gateway_core::engine::execution::{ClientTransport, ExecutionSession, StartedExecution};
+use gateway_core::engine::{CommitRequirement, EngineError};
 use gateway_core::error::{GatewayError, GatewayErrorKind};
 use gateway_core::event::{ProviderEvent, ProviderResponseHeader};
 use gateway_core::lifecycle::ConnectionGuard;
@@ -83,6 +83,19 @@ pub(crate) async fn responses(
         Ok(started) => started,
         Err(error) => return gateway_error_response(&error),
     };
+    tracing::info!(
+        target: "request_dump",
+        provider = "openai",
+        request_id = %started.request_id,
+        request_direction = "downstream",
+        request_transport = if streaming { "http_sse" } else { "http_json" },
+        request_method = "POST",
+        request_path = "/v1/responses",
+        request_headers = ?headers,
+        request_body = %String::from_utf8_lossy(&body),
+        contains_sensitive_data = true,
+        "request dump (unredacted)"
+    );
     let StartedExecution {
         stream, session, ..
     } = started;
@@ -152,7 +165,8 @@ pub async fn collect_execution_response(session: Box<dyn ExecutionSession>) -> R
     let events = match session.collect_uncommitted().await {
         Ok(events) => events,
         Err(error) => {
-            let response = engine_error_response(&error);
+            let response_headers = session.response_headers().to_vec();
+            let response = engine_error_response_with_headers(&error, &response_headers);
             return execution.record_response_status(response).await;
         }
     };
@@ -178,7 +192,7 @@ pub async fn collect_execution_response(session: Box<dyn ExecutionSession>) -> R
         .await
     {
         execution.cancel_and_finalize().await;
-        let response = engine_error_response(&error);
+        let response = engine_error_response_with_headers(&error, &response_headers);
         return execution.record_response_status(response).await;
     }
     if !session.is_finalized() {
@@ -246,6 +260,13 @@ fn apply_response_headers(
     response
 }
 
+fn engine_error_response_with_headers(
+    error: &EngineError,
+    response_headers: &[ProviderResponseHeader],
+) -> Response {
+    apply_response_headers(engine_error_response(error), response_headers)
+}
+
 /// 编码首个 SSE frame 后提交下游，再持续驱动同一执行会话。
 pub async fn stream_execution_response(
     session: Box<dyn ExecutionSession>,
@@ -267,7 +288,8 @@ pub async fn stream_execution_response(
             return response;
         }
         Err(error) => {
-            let response = engine_error_response(&error);
+            let response_headers = session.response_headers().to_vec();
+            let response = engine_error_response_with_headers(&error, &response_headers);
             return execution.record_response_status(response).await;
         }
     };
@@ -299,7 +321,7 @@ pub async fn stream_execution_response(
         .await
     {
         execution.cancel_and_finalize().await;
-        let response = engine_error_response(&error);
+        let response = engine_error_response_with_headers(&error, &response_headers);
         return execution.record_response_status(response).await;
     }
     let Some(session) = execution.into_session() else {

backend/crates/gateway-api/src/openai/responses/request.rs

@@ -16,6 +16,7 @@ const OPENAI_PROTOCOL: &str = "openai";
 const OPENAI_SUBAGENT_KEY: &str = "x-openai-subagent";
 const CODEX_TURN_METADATA_KEY: &str = "x-codex-turn-metadata";
 const PASSTHROUGH_HEADERS_CONTEXT_KEY: &str = "opaque_request_headers";
+const DOWNSTREAM_WEBSOCKET_CONNECTION_ID_CONTEXT_KEY: &str = "downstream_websocket_connection_id";
 
 /// Responses 请求进入共享解码内核时的下游传输来源。
 #[derive(Clone, Copy)]
@@ -35,6 +36,7 @@ pub struct OpenAiRequestHeaders {
     include_timing_metrics: Option<String>,
     codex_window_id: Option<String>,
     parent_thread_id: Option<String>,
+    downstream_websocket_connection_id: Option<String>,
 
     conversation_id: Option<String>,
     session_id: Option<String>,
@@ -60,6 +62,7 @@ impl OpenAiRequestHeaders {
             include_timing_metrics: header_string(headers, "x-responsesapi-include-timing-metrics"),
             codex_window_id: header_string(headers, "x-codex-window-id"),
             parent_thread_id: header_string(headers, "x-codex-parent-thread-id"),
+            downstream_websocket_connection_id: None,
             conversation_id: header_string(headers, "conversation-id")
                 .or_else(|| header_string(headers, "conversation_id")),
             session_id: header_string(headers, "session-id")
@@ -74,6 +77,16 @@ impl OpenAiRequestHeaders {
         }
     }
 
+    /// 绑定代理为当前下游 WebSocket 分配的连接身份。
+    #[must_use]
+    pub(super) fn with_downstream_websocket_connection_id(
+        mut self,
+        connection_id: impl Into<String>,
+    ) -> Self {
+        self.downstream_websocket_connection_id = Some(connection_id.into());
+        self
+    }
+
     fn apply_subagent(&self, body: &mut Map<String, Value>) {
         if let Some(subagent) = self.subagent.as_deref() {
             inject_subagent_metadata(body, subagent);
@@ -109,6 +122,11 @@ impl OpenAiRequestHeaders {
             "parent_thread_id",
             self.parent_thread_id.as_ref(),
         );
+        insert_protocol_context(
+            &mut context,
+            DOWNSTREAM_WEBSOCKET_CONNECTION_ID_CONTEXT_KEY,
+            self.downstream_websocket_connection_id.as_ref(),
+        );
         insert_protocol_context(
             &mut context,
             "conversation_id",

backend/crates/gateway-api/src/openai/responses/websocket/forward.rs

@@ -77,8 +77,8 @@ pub(super) async fn forward_execution(
                 return send_gateway_error(connection, &error, &request_id).await;
             }
             ActiveInput::Event(Err(error)) => {
-                let error = gateway_error_from_engine(&error);
-                return send_gateway_error(connection, &error, &request_id).await;
+                return send_initial_engine_error(connection, &mut execution, &error, &request_id)
+                    .await;
             }
             ActiveInput::Control => continue,
             ActiveInput::Disconnect => return ForwardOutcome::Disconnect,
@@ -271,6 +271,30 @@ pub(super) async fn forward_execution(
     }
 }
 
+async fn send_initial_engine_error(
+    connection: &mut ResponsesWebSocketConnection,
+    execution: &mut PendingExecution,
+    error: &EngineError,
+    request_id: &Arc<str>,
+) -> ForwardOutcome {
+    let response_headers = execution
+        .session_mut()
+        .map(|session| session.response_headers().to_vec())
+        .unwrap_or_default();
+    if !response_headers.is_empty()
+        && !send_metadata(
+            connection,
+            request_id,
+            response_metadata_event(request_id, &response_headers),
+        )
+        .await
+    {
+        return ForwardOutcome::Disconnect;
+    }
+    let error = gateway_error_from_engine(error);
+    send_gateway_error(connection, &error, request_id).await
+}
+
 fn commit_connection_replay(
     replay: &mut ConnectionReplaySnapshot,
     encoder: &OpenAiResponsesEncoder,

backend/crates/gateway-api/src/openai/responses/websocket/mod.rs

@@ -60,12 +60,14 @@ pub(crate) async fn responses_websocket(
         &headers,
         connect_info.map(|Extension(ConnectInfo(address))| address),
     );
+    let request_headers = OpenAiRequestHeaders::from_headers(&headers);
     ResponsesWebSocketAdapter::new(service).upgrade_with_client_context(
         websocket,
         client,
         client_ip,
         user_agent,
-        OpenAiRequestHeaders::from_headers(&headers),
+        request_headers,
+        headers,
     )
 }
 
@@ -89,15 +91,29 @@ impl ResponsesWebSocketAdapter {
         client_ip: Option<IpAddr>,
         user_agent: Option<String>,
         request_headers: OpenAiRequestHeaders,
+        raw_headers: HeaderMap,
     ) -> Response {
         let connection_guard = match self.service.try_register_connection() {
             Ok(guard) => guard,
             Err(_) => return runtime_unavailable_response().into_response(),
         };
+        let connection_id = self.service.next_request_id().replacen("req_", "ws_", 1);
+        tracing::info!(
+            target: "request_dump",
+            provider = "openai",
+            websocket_connection_id = %connection_id,
+            request_direction = "downstream",
+            request_transport = "websocket_handshake",
+            request_method = "GET",
+            request_path = "/v1/responses",
+            request_headers = ?raw_headers,
+            contains_sensitive_data = true,
+            "request dump (unredacted)"
+        );
         let session = ResponsesWebSocketSession {
             service: self.service.clone(),
             client,
-            connection_id: self.service.next_request_id().replacen("req_", "ws_", 1),
+            connection_id,
             client_ip,
             user_agent,
             request_headers,
@@ -143,6 +159,8 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
     );
     let _connection_guard = connection_guard;
     let cancellation = lifecycle.cancellation();
+    let request_headers =
+        request_headers.with_downstream_websocket_connection_id(connection_id.clone());
     let mut connection = ResponsesWebSocketConnection::new(socket, connection_id, cancellation);
     let mut request_count = 0_u64;
     let mut replay = ConnectionReplaySnapshot::default();
@@ -169,6 +187,18 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
         }
         request_count = request_count.saturating_add(1);
         let correlation_id = Arc::<str>::from(service.next_request_id());
+        tracing::info!(
+            target: "request_dump",
+            provider = "openai",
+            websocket_connection_id = connection.id(),
+            correlation_id = %correlation_id,
+            request_index = request_count,
+            request_direction = "downstream",
+            request_transport = "websocket",
+            request_payload = %payload,
+            contains_sensitive_data = true,
+            "request dump (unredacted)"
+        );
         let decoded = match decode_response_create_with_context(&payload, &request_headers) {
             Ok(decoded) => decoded.with_client_context(client_ip, user_agent.clone()),
             Err(error) => {
@@ -217,6 +247,14 @@ async fn serve_responses_websocket(socket: WebSocket, session: ResponsesWebSocke
                 continue;
             }
         };
+        tracing::info!(
+            target: "request_dump",
+            provider = "openai",
+            websocket_connection_id = connection.id(),
+            correlation_id = %correlation_id,
+            request_id = %started.request_id,
+            "request dump correlation"
+        );
 
         if forward_execution(&mut connection, started, &mut replay).await
             == ForwardOutcome::Disconnect

backend/crates/gateway-api/tests/admin/accounts/mod.rs

@@ -348,6 +348,8 @@ mod response {
             output_tokens_display: "-".to_owned(),
             cached_tokens: None,
             cached_tokens_display: "-".to_owned(),
+            reasoning_tokens: None,
+            reasoning_tokens_display: "-".to_owned(),
             image_input_tokens: None,
             image_input_tokens_display: "-".to_owned(),
             image_output_tokens: None,
@@ -374,6 +376,8 @@ mod response {
         let value = serde_json::to_value(view).expect("serialize account usage");
         assert!(value["inputTokens"].is_null());
         assert!(value["totalTokens"].is_null());
+        assert!(value["reasoningTokens"].is_null());
+        assert_eq!(value["reasoningTokensDisplay"], "-");
         assert_eq!(value["createdTokensDisplay"], "-");
     }
 

backend/crates/gateway-api/tests/openai/responses/http.rs

@@ -893,6 +893,41 @@ async fn streaming_rate_limit_before_first_frame_should_persist_the_returned_429
     assert_eq!(trace.client_statuses(), vec![429]);
 }
 
+#[tokio::test]
+async fn streaming_failure_before_first_frame_should_return_safe_observed_headers() {
+    let trace = Arc::new(Trace::default());
+    let session = FakeSession::streaming(
+        Arc::clone(&trace),
+        vec![NextStep::Error(EngineError::Provider(ProviderError::new(
+            ProviderErrorKind::Unavailable,
+            UpstreamSendState::Ambiguous,
+        )))],
+    )
+    .with_response_headers(vec![
+        ProviderResponseHeader::new(
+            "x-codex-turn-state",
+            Bytes::from_static(b"turn-state-before-close"),
+        ),
+        ProviderResponseHeader::new(
+            "authorization",
+            Bytes::from_static(b"must-not-cross-boundary"),
+        ),
+    ]);
+
+    let response = stream_execution_response(Box::new(session), None).await;
+
+    assert_eq!(response.status(), StatusCode::BAD_GATEWAY);
+    assert_eq!(
+        response
+            .headers()
+            .get("x-codex-turn-state")
+            .and_then(|value| value.to_str().ok()),
+        Some("turn-state-before-close")
+    );
+    assert!(response.headers().get("authorization").is_none());
+    assert_eq!(trace.client_statuses(), vec![502]);
+}
+
 #[tokio::test]
 async fn streaming_upstream_http_failure_before_first_frame_should_preserve_raw_response() {
     let trace = Arc::new(Trace::default());
@@ -979,11 +1014,29 @@ async fn buffered_upstream_http_failure_should_preserve_raw_response() {
             raw_body.clone(),
         ));
     let session = FakeSession::buffered(Arc::clone(&trace), vec![started(), completed()])
-        .with_collect_error(EngineError::Provider(error));
+        .with_collect_error(EngineError::Provider(error))
+        .with_response_headers(vec![
+            ProviderResponseHeader::new(
+                "x-codex-turn-state",
+                Bytes::from_static(b"turn-state-before-close"),
+            ),
+            ProviderResponseHeader::new(
+                "authorization",
+                Bytes::from_static(b"must-not-cross-boundary"),
+            ),
+        ]);
 
     let response = collect_execution_response(Box::new(session)).await;
 
     assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+    assert_eq!(
+        response
+            .headers()
+            .get("x-codex-turn-state")
+            .and_then(|value| value.to_str().ok()),
+        Some("turn-state-before-close")
+    );
+    assert!(response.headers().get("authorization").is_none());
     assert_eq!(
         to_bytes(response.into_body(), usize::MAX)
             .await

backend/crates/gateway-api/tests/openai/responses/websocket/connection.rs

@@ -428,6 +428,7 @@ async fn idle_connection_reaches_the_official_limit_without_starting_an_executio
         client: authenticated_client("sk_ws_atomic"),
         trace: Arc::clone(&trace),
         response_headers: Vec::new(),
+        fail_before_first_event: false,
     });
     let app = api_router(execution).await;
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0")

backend/crates/gateway-api/tests/openai/responses/websocket/mod.rs

@@ -2,7 +2,7 @@ mod connection;
 mod protocol;
 
 use std::sync::{
-    Arc,
+    Arc, Mutex,
     atomic::{AtomicBool, AtomicUsize, Ordering},
 };
 use std::time::{Duration, SystemTime};
@@ -280,17 +280,26 @@ struct AtomicFailureTrace {
     next_calls: AtomicUsize,
     committed: AtomicBool,
     finalized: AtomicBool,
+    downstream_websocket_connection_ids: Mutex<Vec<String>>,
 }
 
 struct AtomicFailureSession {
     trace: Arc<AtomicFailureTrace>,
     response_headers: Vec<ProviderResponseHeader>,
+    fail_before_first_event: bool,
 }
 
 impl ExecutionSession for AtomicFailureSession {
     fn next_event(&mut self) -> BoxFuture<'_, Result<Option<CoordinatedEvent>, EngineError>> {
         Box::pin(async move {
-            match self.trace.next_calls.fetch_add(1, Ordering::AcqRel) {
+            let next_call = self.trace.next_calls.fetch_add(1, Ordering::AcqRel);
+            if self.fail_before_first_event && next_call == 0 {
+                return Err(EngineError::Provider(ProviderError::new(
+                    ProviderErrorKind::Unavailable,
+                    UpstreamSendState::Ambiguous,
+                )));
+            }
+            match next_call {
                 0 => Ok(Some(atomic_failure_batch())),
                 1 => {
                     self.trace.finalized.store(true, Ordering::Release);
@@ -340,6 +349,7 @@ struct AtomicFailureExecution {
     client: AuthenticatedClient,
     trace: Arc<AtomicFailureTrace>,
     response_headers: Vec<ProviderResponseHeader>,
+    fail_before_first_event: bool,
 }
 
 impl ExecutionService for AtomicFailureExecution {
@@ -365,6 +375,19 @@ impl ExecutionService for AtomicFailureExecution {
         request: StartExecution,
     ) -> BoxFuture<'_, Result<StartedExecution, GatewayError>> {
         self.trace.starts.fetch_add(1, Ordering::AcqRel);
+        if let Operation::Generate(generate) = &request.operation
+            && let Some(connection_id) = generate
+                .protocol_payload()
+                .context()
+                .get("downstream_websocket_connection_id")
+                .and_then(Value::as_str)
+        {
+            self.trace
+                .downstream_websocket_connection_ids
+                .lock()
+                .expect("downstream connection trace lock")
+                .push(connection_id.to_owned());
+        }
         Box::pin(async move {
             Ok(StartedExecution {
                 request_id: gateway_core::engine::ModelRequestId::new("req_ws_atomic")
@@ -374,6 +397,7 @@ impl ExecutionService for AtomicFailureExecution {
                 session: Box::new(AtomicFailureSession {
                     trace: Arc::clone(&self.trace),
                     response_headers: self.response_headers.clone(),
+                    fail_before_first_event: self.fail_before_first_event,
                 }),
             })
         })
@@ -439,6 +463,7 @@ async fn websocket_atomic_upstream_failure_batch_should_be_forwarded_once() {
         client: authenticated_client("sk_ws_atomic"),
         trace: Arc::clone(&trace),
         response_headers: Vec::new(),
+        fail_before_first_event: false,
     });
     let app = api_router(execution).await;
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
@@ -510,6 +535,14 @@ async fn websocket_atomic_upstream_failure_batch_should_be_forwarded_once() {
     assert!(trace.committed.load(Ordering::Acquire));
     assert_eq!(trace.next_calls.load(Ordering::Acquire), 2);
     assert!(trace.finalized.load(Ordering::Acquire));
+    {
+        let downstream_connection_ids = trace
+            .downstream_websocket_connection_ids
+            .lock()
+            .expect("downstream connection trace lock");
+        assert_eq!(downstream_connection_ids.len(), 1);
+        assert!(downstream_connection_ids[0].starts_with("ws_"));
+    }
 
     socket.close(None).await.expect("close WebSocket");
     server.abort();
@@ -540,6 +573,7 @@ async fn websocket_response_metadata_should_preserve_ordinary_headers_without_bl
             ProviderResponseHeader::new("content-type", Bytes::from_static(b"application/private")),
             ProviderResponseHeader::new("x-request-id", Bytes::from_static(b"req_upstream")),
         ],
+        fail_before_first_event: false,
     });
     let app = api_router(execution).await;
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
@@ -624,6 +658,91 @@ async fn websocket_response_metadata_should_preserve_ordinary_headers_without_bl
     server.abort();
 }
 
+#[tokio::test]
+async fn websocket_failure_before_first_event_should_return_observed_metadata_then_error() {
+    let trace = Arc::new(AtomicFailureTrace::default());
+    let execution = Arc::new(AtomicFailureExecution {
+        client: authenticated_client("sk_ws_atomic"),
+        trace: Arc::clone(&trace),
+        response_headers: vec![
+            ProviderResponseHeader::new(
+                "x-codex-turn-state",
+                Bytes::from_static(b"turn-state-before-close"),
+            ),
+            ProviderResponseHeader::new(
+                "authorization",
+                Bytes::from_static(b"must-not-cross-boundary"),
+            ),
+        ],
+        fail_before_first_event: true,
+    });
+    let app = api_router(execution).await;
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind WebSocket test server");
+    let address = listener.local_addr().expect("test server address");
+    let server = tokio::spawn(async move {
+        axum::serve(listener, app)
+            .await
+            .expect("serve WebSocket test router");
+    });
+    let mut request = format!("ws://{address}/v1/responses")
+        .into_client_request()
+        .expect("WebSocket request");
+    request.headers_mut().insert(
+        AUTHORIZATION,
+        "Bearer sk_ws_atomic".parse().expect("authorization"),
+    );
+    let (mut socket, response) = connect_async(request).await.expect("upgrade WebSocket");
+    assert_eq!(response.status(), StatusCode::SWITCHING_PROTOCOLS);
+    socket
+        .send(ClientMessage::Text(
+            json!({
+                "type": "response.create",
+                "model": "model-a",
+                "input": "hello"
+            })
+            .to_string()
+            .into(),
+        ))
+        .await
+        .expect("send response.create");
+
+    let mut messages = Vec::<Value>::new();
+    while messages.len() < 2 {
+        let message = tokio::time::timeout(Duration::from_secs(2), socket.next())
+            .await
+            .expect("WebSocket response timeout")
+            .expect("WebSocket remains open")
+            .expect("valid WebSocket frame");
+        let ClientMessage::Text(text) = message else {
+            continue;
+        };
+        messages.push(serde_json::from_str(&text).expect("JSON WebSocket event"));
+    }
+
+    assert_eq!(
+        messages
+            .iter()
+            .filter_map(|message| message.get("type").and_then(Value::as_str))
+            .collect::<Vec<_>>(),
+        vec!["response.metadata", "error"]
+    );
+    let headers = messages[0]
+        .get("headers")
+        .and_then(Value::as_object)
+        .expect("response metadata headers");
+    assert_eq!(
+        headers.get("x-codex-turn-state"),
+        Some(&json!("turn-state-before-close"))
+    );
+    assert!(!headers.contains_key("authorization"));
+    assert!(!trace.committed.load(Ordering::Acquire));
+
+    socket.close(None).await.expect("close WebSocket");
+    server.abort();
+}
+
 #[tokio::test]
 async fn websocket_response_create_should_not_have_a_private_16_mib_frame_limit() {
     let trace = Arc::new(AtomicFailureTrace::default());
@@ -631,6 +750,7 @@ async fn websocket_response_create_should_not_have_a_private_16_mib_frame_limit(
         client: authenticated_client("sk_ws_atomic"),
         trace,
         response_headers: Vec::new(),
+        fail_before_first_event: false,
     });
     let app = api_router(execution).await;
     let listener = tokio::net::TcpListener::bind("127.0.0.1:0")

backend/crates/gateway-core/src/engine/coordinator.rs

@@ -210,6 +210,8 @@ struct CurrentAttempt {
     index: NonZeroU32,
     started_at: SystemTime,
     send_observed: bool,
+    /// Provider 已为当前 attempt 交付可供同请求恢复的不透明状态检查点。
+    session_checkpoint_observed: bool,
     response_observation: Option<ProviderResponseObservation>,
 }
 
@@ -570,6 +572,9 @@ where
                     return Err(EngineError::Deadline);
                 }
                 PollBoundary::Item(Some(Ok(mut event))) => {
+                    if let Some(state) = event.session_update() {
+                        self.observe_session_checkpoint(state);
+                    }
                     if let Some(observation) = event.take_observation() {
                         self.observe_response(observation);
                     }
@@ -931,6 +936,7 @@ where
             index: next_attempt,
             started_at: SystemTime::now(),
             send_observed: false,
+            session_checkpoint_observed: false,
             response_observation: None,
         });
         Ok(None)
@@ -972,6 +978,17 @@ where
         self.mark_send_observed().await;
     }
 
+    fn observe_session_checkpoint(&mut self, state: &ProviderSessionState) {
+        let Some(current) = self.current.as_mut() else {
+            return;
+        };
+        if state.provider() != current.metadata.provider().as_str() {
+            return;
+        }
+        current.session_checkpoint_observed = true;
+        self.operation.set_provider_session_state(state.clone());
+    }
+
     async fn mark_send_observed(&mut self) {
         let Some(current) = self.current.as_mut() else {
             return;
@@ -1071,6 +1088,11 @@ where
             .await;
         }
         let provider_proved_replay_safe = provider_proved_replay_safe(&error);
+        // Ambiguous 仍是默认 fail-closed。唯一例外是 Provider 在同一 attempt 中先
+        // 交付了不透明恢复检查点，并明确证明从该检查点重放安全；该例外只用于
+        // 固定原账号的 transport recovery，不放开普通换号或 continuation 重放。
+        let checkpointed_transport_replay =
+            current.session_checkpoint_observed && error.replay_is_safe();
         let continuation_retry = self.prepare_continuation_retry(
             &current,
             &error,
@@ -1093,14 +1115,16 @@ where
                 delay,
             }) if self.downstream_committed_at.is_none()
                 && !self.delivery_pending
-                && attempt_send_state != UpstreamSendState::Ambiguous =>
+                && (attempt_send_state != UpstreamSendState::Ambiguous
+                    || checkpointed_transport_replay) =>
             {
                 Some((AttemptTransport::Retry(retry_index), delay))
             }
             Some(crate::error::PreDeliveryRetry::SameAccountTransportFallback)
                 if self.downstream_committed_at.is_none()
                     && !self.delivery_pending
-                    && attempt_send_state != UpstreamSendState::Ambiguous =>
+                    && (attempt_send_state != UpstreamSendState::Ambiguous
+                        || checkpointed_transport_replay) =>
             {
                 Some((AttemptTransport::Fallback, Duration::ZERO))
             }

backend/crates/gateway-core/src/error.rs

@@ -623,13 +623,21 @@ impl ProviderError {
             .get_or_insert_with(|| Box::new(ProviderErrorFailureObservation::default()))
     }
 
-    /// 标记 Provider 已证明本次拒绝没有执行生成，可在下游提交前重放。
+    /// 标记 Provider 已证明本次失败可在下游提交前安全重放。
+    ///
+    /// 证明可以来自明确的“未执行”拒绝，也可以来自同 attempt 已交付、可恢复
+    /// 同一业务请求的状态检查点；后者仍须由 Core 的同账号恢复门禁单独校验。
     #[must_use]
     pub const fn with_replay_safe(mut self) -> Self {
         self.replay_safe = true;
         self
     }
 
+    /// 原地标记 Provider 已证明本次失败可从已交付的状态检查点安全恢复。
+    pub const fn set_replay_safe(&mut self) {
+        self.replay_safe = true;
+    }
+
     /// 允许 Core 仅在客户端尚未收到任何事件时执行一次受预算约束的换号恢复。
     ///
     /// 该标记不证明上游未执行，也不等价于 [`Self::with_replay_safe`]；它只表达

backend/crates/gateway-core/src/event.rs

@@ -763,7 +763,8 @@ impl ProviderEvent {
         }
     }
 
-    /// 把本事件标记为 Provider 连接内状态的提交边界。
+    /// 附加 Provider 私有状态检查点；Core 可将其用于同请求恢复，协议连接可在
+    /// terminal 事件上持有它供下一轮使用。
     pub fn attach_session_update(&mut self, state: ProviderSessionState) {
         self.session_update = Some(Box::new(state));
     }

backend/crates/gateway-core/tests/engine/coordinator.rs

@@ -270,6 +270,7 @@ enum Script {
 struct ScriptedProvider {
     scripts: Mutex<VecDeque<Script>>,
     contexts: Mutex<Vec<AttemptContext>>,
+    operations: Mutex<Vec<Operation>>,
 }
 
 impl ScriptedProvider {
@@ -277,6 +278,7 @@ impl ScriptedProvider {
         Self {
             scripts: Mutex::new(scripts.into()),
             contexts: Mutex::new(Vec::new()),
+            operations: Mutex::new(Vec::new()),
         }
     }
 }
@@ -302,6 +304,10 @@ impl Provider for ScriptedProvider {
         request: ProviderRequest,
         context: AttemptContext,
     ) -> Result<ProviderStream, ProviderError> {
+        self.operations
+            .lock()
+            .expect("operations lock")
+            .push(request.operation().clone());
         self.contexts.lock().expect("contexts lock").push(context);
         let script = self
             .scripts
@@ -2838,14 +2844,25 @@ fn ambiguous_send_state_stops_retry() {
 fn ambiguous_pre_delivery_retry_marker_does_not_rotate_account() {
     let operation = generate_operation();
     let route_plan = plan(&operation);
+    let checkpoint = ProviderSessionState::new(
+        "openai",
+        Map::from_iter([("checkpoint".to_owned(), json!("opaque-state"))]),
+    )
+    .expect("provider checkpoint");
+    let mut checkpoint_event = ProviderEvent::observation(ProviderResponseObservation::new(
+        UpstreamTransport::new("websocket").expect("transport"),
+    ));
+    checkpoint_event.attach_session_update(checkpoint);
     let (coordinator, store, provider) = coordinator(vec![
-        Script::Stream {
+        Script::ObservedStream {
             account_id: "acct_first",
-            items: vec![Err(ProviderError::new(
-                ProviderErrorKind::Transport,
-                UpstreamSendState::Ambiguous,
-            )
-            .with_pre_delivery_retry())],
+            items: vec![
+                Ok(checkpoint_event),
+                Err(
+                    ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
+                        .with_pre_delivery_retry(),
+                ),
+            ],
         },
         Script::Stream {
             account_id: "acct_second",
@@ -2874,6 +2891,70 @@ fn ambiguous_pre_delivery_retry_marker_does_not_rotate_account() {
     assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed);
 }
 
+#[test]
+fn ambiguous_transport_retry_uses_provider_session_checkpoint_on_same_account() {
+    let operation = generate_operation();
+    let route_plan = plan(&operation);
+    let checkpoint = ProviderSessionState::new(
+        "openai",
+        Map::from_iter([("checkpoint".to_owned(), json!("opaque-state"))]),
+    )
+    .expect("provider checkpoint");
+    let mut checkpoint_event = ProviderEvent::observation(ProviderResponseObservation::new(
+        UpstreamTransport::new("websocket").expect("transport"),
+    ));
+    checkpoint_event.attach_session_update(checkpoint.clone());
+    let retry_index = NonZeroU32::new(1).expect("retry index");
+    let (coordinator, store, provider) = coordinator(vec![
+        Script::ObservedStream {
+            account_id: "acct_first",
+            items: vec![
+                Ok(checkpoint_event),
+                Err(
+                    ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
+                        .with_replay_safe()
+                        .with_pre_delivery_transport_retry(retry_index, Duration::ZERO),
+                ),
+            ],
+        },
+        Script::Stream {
+            account_id: "acct_first",
+            items: complete_stream(None),
+        },
+    ]);
+
+    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");
+    block_on(session.collect_uncommitted()).expect("checkpointed retry succeeds");
+    block_on(session.commit_downstream(Some(200))).expect("commit winning response");
+
+    let original = ProviderAccountId::new("acct_first").expect("account id");
+    let contexts = provider.contexts.lock().expect("contexts lock");
+    assert_eq!(contexts.len(), 2);
+    assert_eq!(
+        contexts[1].transport(),
+        AttemptTransport::Retry(retry_index)
+    );
+    assert_eq!(contexts[1].required_account(), Some(&original));
+    drop(contexts);
+    let operations = provider.operations.lock().expect("operations lock");
+    assert_eq!(
+        operations[1].provider_session_state("openai"),
+        Some(&checkpoint)
+    );
+    let state = store.state.lock().expect("store lock");
+    assert_eq!(state.intermediate_failures, 1);
+    assert_eq!(state.finalizations[0].attempt_count, 2);
+    assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded);
+}
+
 #[test]
 fn pre_delivery_transport_fallback_retries_the_same_account_with_fallback_transport() {
     let operation = generate_operation();

backend/crates/gateway-host/src/config.rs

@@ -118,6 +118,9 @@ pub struct LoggingConfig {
     pub level: String,
     pub stdout: bool,
     pub file: FileLoggingConfig,
+    /// 将完整请求原文写入独立诊断日志；默认关闭，因为内容包含凭据。
+    #[serde(default)]
+    pub request_dump: bool,
 }
 
 impl LoggingConfig {

backend/crates/gateway-host/src/logging.rs

@@ -6,16 +6,19 @@ use std::io::{self, Write};
 use std::path::{Path, PathBuf};
 
 use chrono::Utc;
-use tracing_appender::non_blocking::{NonBlockingBuilder, WorkerGuard};
+use tracing_appender::non_blocking::{NonBlocking, NonBlockingBuilder, WorkerGuard};
 use tracing_subscriber::EnvFilter;
 use tracing_subscriber::{Layer as _, layer::SubscriberExt as _, util::SubscriberInitExt as _};
 
-use crate::config::LoggingConfig;
+use crate::config::{FileLoggingConfig, LoggingConfig};
 
 const LOG_FILE_PREFIX: &str = "codex-proxy-rs";
 const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oath";
 /// OAuth 恢复记录含原始 AT/RT，文件日志开启时必须落到独立文件。
 const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery";
+const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump";
+/// 请求转储含完整凭据与正文，只允许在显式开启时写入独立文件。
+const REQUEST_DUMP_LOG_TARGET: &str = "request_dump";
 
 /// non-blocking 日志 writer 的进程级守卫。
 pub struct LogGuard {
@@ -39,6 +42,7 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result<LogGuard, LogError>
     let directive = env::var("RUST_LOG").unwrap_or_else(|_| config.level.clone());
     let file_filter = application_file_filter(&directive)?;
     let recovery_file_filter = oauth_recovery_file_filter();
+    let request_dump_file_filter = request_dump_file_filter();
     let stdout_filter =
         EnvFilter::try_new(stdout_filter_directive(&directive, config.file.enabled))
             .map_err(|_| LogError::InvalidFilter)?;
@@ -52,37 +56,30 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result<LogGuard, LogError>
         writer
     });
     let (file_writer, recovery_file_writer) = if config.file.enabled {
-        let maximum_bytes = config
-            .file
-            .max_file_size_mb
-            .checked_mul(1024 * 1024)
-            .ok_or(LogError::SizeOverflow)?;
-        let file_writer = RotatingLogWriter::open(
-            config.file.directory.clone(),
-            LOG_FILE_PREFIX,
-            maximum_bytes,
-            config.file.retention_days,
-            config.file.max_files,
-        )?;
-        let recovery_file_writer = RotatingLogWriter::open(
-            config.file.directory.clone(),
+        let (file_writer, file_guard) =
+            rotating_file_writer(&config.file, LOG_FILE_PREFIX, "gateway-log-file")?;
+        let (recovery_file_writer, recovery_file_guard) = rotating_file_writer(
+            &config.file,
             OAUTH_RECOVERY_LOG_FILE_PREFIX,
-            maximum_bytes,
-            config.file.retention_days,
-            config.file.max_files,
+            "gateway-log-oauth-recovery-file",
         )?;
-        let (file_writer, file_guard) = NonBlockingBuilder::default()
-            .thread_name("gateway-log-file")
-            .finish(file_writer);
-        let (recovery_file_writer, recovery_file_guard) = NonBlockingBuilder::default()
-            .thread_name("gateway-log-oauth-recovery-file")
-            .finish(recovery_file_writer);
         guards.push(file_guard);
         guards.push(recovery_file_guard);
         (Some(file_writer), Some(recovery_file_writer))
     } else {
         (None, None)
     };
+    let request_dump_file_writer = if config.request_dump {
+        let (writer, guard) = rotating_file_writer(
+            &config.file,
+            REQUEST_DUMP_LOG_FILE_PREFIX,
+            "gateway-log-request-dump-file",
+        )?;
+        guards.push(guard);
+        Some(writer)
+    } else {
+        None
+    };
 
     let stdout_layer = stdout_writer.map(|writer| {
         tracing_subscriber::fmt::layer()
@@ -118,10 +115,24 @@ pub fn initialize_logging(config: &LoggingConfig) -> Result<LogGuard, LogError>
             .with_span_list(true)
             .with_filter(recovery_file_filter)
     });
+    let request_dump_file_layer = request_dump_file_writer.map(|writer| {
+        tracing_subscriber::fmt::layer()
+            .json()
+            .with_writer(writer)
+            .with_target(true)
+            .with_file(true)
+            .with_line_number(true)
+            .with_thread_ids(true)
+            .with_thread_names(true)
+            .with_current_span(true)
+            .with_span_list(true)
+            .with_filter(request_dump_file_filter)
+    });
     tracing_subscriber::registry()
         .with(stdout_layer)
         .with(file_layer)
         .with(recovery_file_layer)
+        .with(request_dump_file_layer)
         .try_init()
         .map_err(|_| LogError::AlreadyInitialized)?;
     Ok(LogGuard { _writers: guards })
@@ -131,8 +142,12 @@ fn application_file_filter(directive: &str) -> Result<EnvFilter, LogError> {
     let recovery_directive = format!("{OAUTH_RECOVERY_LOG_TARGET}=off")
         .parse()
         .expect("static OAuth recovery log directive is valid");
+    let request_dump_directive = format!("{REQUEST_DUMP_LOG_TARGET}=off")
+        .parse()
+        .expect("static request dump log directive is valid");
     EnvFilter::try_new(directive)
         .map(|filter| filter.add_directive(recovery_directive))
+        .map(|filter| filter.add_directive(request_dump_directive))
         .map_err(|_| LogError::InvalidFilter)
 }
 
@@ -142,14 +157,41 @@ fn oauth_recovery_file_filter() -> EnvFilter {
         .expect("static OAuth recovery log filter is valid")
 }
 
+fn request_dump_file_filter() -> EnvFilter {
+    format!("off,{REQUEST_DUMP_LOG_TARGET}=info")
+        .parse()
+        .expect("static request dump log filter is valid")
+}
+
 fn stdout_filter_directive(directive: &str, persistent_log_enabled: bool) -> String {
     if !persistent_log_enabled {
-        directive.to_owned()
+        format!("{directive},{REQUEST_DUMP_LOG_TARGET}=off")
     } else {
         "off,gateway_startup=info".to_owned()
     }
 }
 
+fn rotating_file_writer(
+    config: &FileLoggingConfig,
+    file_prefix: &'static str,
+    thread_name: &'static str,
+) -> Result<(NonBlocking, WorkerGuard), LogError> {
+    let maximum_bytes = config
+        .max_file_size_mb
+        .checked_mul(1024 * 1024)
+        .ok_or(LogError::SizeOverflow)?;
+    let writer = RotatingLogWriter::open(
+        config.directory.clone(),
+        file_prefix,
+        maximum_bytes,
+        config.retention_days,
+        config.max_files,
+    )?;
+    Ok(NonBlockingBuilder::default()
+        .thread_name(thread_name)
+        .finish(writer))
+}
+
 struct RotatingLogWriter {
     directory: PathBuf,
     file_prefix: &'static str,

backend/crates/gateway-host/tests/config.rs

@@ -103,6 +103,7 @@ fn valid_config() -> HostConfig {
                 max_file_size_mb: 100,
                 max_files: 30,
             },
+            request_dump: false,
         },
         system_update,
         drain_timeout_seconds: 30,

backend/crates/gateway-host/tests/logging.rs

@@ -8,13 +8,18 @@ use gateway_host::system_update::SystemUpdateConfig;
 
 const LOG_DIRECTORY_ENV: &str = "CPR_LOGGING_TEST_DIRECTORY";
 const CHILD_PROCESS_ENV: &str = "CPR_LOGGING_TEST_CHILD";
+const REQUEST_DUMP_ENABLED_ENV: &str = "CPR_LOGGING_TEST_REQUEST_DUMP_ENABLED";
 const APPLICATION_LOG_FILE_PREFIX: &str = "codex-proxy-rs.";
 const OAUTH_RECOVERY_LOG_FILE_PREFIX: &str = "codex-proxy-rs-oath.";
+const REQUEST_DUMP_LOG_FILE_PREFIX: &str = "codex-proxy-rs-request-dump.";
 const APPLICATION_LOG_TARGET: &str = "logging_test_application";
 const APPLICATION_LOG_MARKER: &str = "application-file-filter-test";
 const OAUTH_RECOVERY_LOG_TARGET: &str = "oauth_recovery";
 const OAUTH_RECOVERY_LOG_MARKER: &str = "oauth-recovery-file-filter-test";
 const OAUTH_RECOVERY_PROVIDER: &str = "openai";
+const REQUEST_DUMP_LOG_TARGET: &str = "request_dump";
+const REQUEST_DUMP_LOG_MARKER: &str = "request-dump-file-filter-test";
+const REQUEST_DUMP_SECRET: &str = "unredacted-authorization-value";
 
 #[test]
 fn logging_requires_at_least_one_sink() {
@@ -34,6 +39,7 @@ fn logging_requires_at_least_one_sink() {
                 max_file_size_mb: 100,
                 max_files: 30,
             },
+            request_dump: false,
         },
         system_update: SystemUpdateConfig::default(),
         drain_timeout_seconds: 30,
@@ -48,11 +54,12 @@ fn logging_requires_at_least_one_sink() {
 }
 
 #[test]
-fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() {
+fn sensitive_file_logging_is_separate_and_overrides_global_log_level() {
     if env::var_os(CHILD_PROCESS_ENV).is_some() {
-        write_oauth_recovery_log(PathBuf::from(
-            env::var_os(LOG_DIRECTORY_ENV).expect("child log directory"),
-        ));
+        write_sensitive_logs(
+            PathBuf::from(env::var_os(LOG_DIRECTORY_ENV).expect("child log directory")),
+            env::var(REQUEST_DUMP_ENABLED_ENV).is_ok_and(|value| value == "true"),
+        );
         return;
     }
 
@@ -60,14 +67,15 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() {
     let output = Command::new(env::current_exe().expect("current test executable"))
         .args([
             "--exact",
-            "logging::oauth_recovery_file_logging_is_separate_and_overrides_global_log_level",
+            "logging::sensitive_file_logging_is_separate_and_overrides_global_log_level",
             "--nocapture",
         ])
         .env(CHILD_PROCESS_ENV, "1")
         .env(LOG_DIRECTORY_ENV, directory.path())
+        .env(REQUEST_DUMP_ENABLED_ENV, "true")
         .env(
             "RUST_LOG",
-            "off,logging_test_application=info,oauth_recovery=off",
+            "off,logging_test_application=info,oauth_recovery=off,request_dump=trace",
         )
         .output()
         .expect("run logging child process");
@@ -84,6 +92,8 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() {
     assert!(application_log.contains(APPLICATION_LOG_MARKER));
     assert!(!application_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET)));
     assert!(!application_log.contains(OAUTH_RECOVERY_LOG_MARKER));
+    assert!(!application_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET)));
+    assert!(!application_log.contains(REQUEST_DUMP_SECRET));
 
     let recovery_log = read_log_file_set(directory.path(), OAUTH_RECOVERY_LOG_FILE_PREFIX);
     assert!(recovery_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET)));
@@ -91,9 +101,49 @@ fn oauth_recovery_file_logging_is_separate_and_overrides_global_log_level() {
     assert!(!recovery_log.contains(&json_target_field(APPLICATION_LOG_TARGET)));
     assert!(!recovery_log.contains(APPLICATION_LOG_MARKER));
     assert!(recovery_log.contains(&format!(r#""provider":"{OAUTH_RECOVERY_PROVIDER}""#)));
+    assert!(!recovery_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET)));
+    assert!(!recovery_log.contains(REQUEST_DUMP_SECRET));
+
+    let request_dump_log = read_log_file_set(directory.path(), REQUEST_DUMP_LOG_FILE_PREFIX);
+    assert!(request_dump_log.contains(&json_target_field(REQUEST_DUMP_LOG_TARGET)));
+    assert!(request_dump_log.contains(REQUEST_DUMP_LOG_MARKER));
+    assert!(request_dump_log.contains(REQUEST_DUMP_SECRET));
+    assert!(!request_dump_log.contains(&json_target_field(APPLICATION_LOG_TARGET)));
+    assert!(!request_dump_log.contains(APPLICATION_LOG_MARKER));
+    assert!(!request_dump_log.contains(&json_target_field(OAUTH_RECOVERY_LOG_TARGET)));
+    assert!(!request_dump_log.contains(OAUTH_RECOVERY_LOG_MARKER));
+
+    let disabled_directory = tempfile::tempdir().expect("create disabled log directory");
+    let output = Command::new(env::current_exe().expect("current test executable"))
+        .args([
+            "--exact",
+            "logging::sensitive_file_logging_is_separate_and_overrides_global_log_level",
+            "--nocapture",
+        ])
+        .env(CHILD_PROCESS_ENV, "1")
+        .env(LOG_DIRECTORY_ENV, disabled_directory.path())
+        .env(REQUEST_DUMP_ENABLED_ENV, "false")
+        .env(
+            "RUST_LOG",
+            "off,logging_test_application=info,oauth_recovery=off,request_dump=trace",
+        )
+        .output()
+        .expect("run logging child process with request dump disabled");
+    assert!(
+        output.status.success(),
+        "logging child process failed:\nstdout:\n{}\nstderr:\n{}",
+        String::from_utf8_lossy(&output.stdout),
+        String::from_utf8_lossy(&output.stderr),
+    );
+    assert!(!log_file_set_exists(
+        disabled_directory.path(),
+        REQUEST_DUMP_LOG_FILE_PREFIX
+    ));
+    let application_log = read_log_file_set(disabled_directory.path(), APPLICATION_LOG_FILE_PREFIX);
+    assert!(!application_log.contains(REQUEST_DUMP_SECRET));
 }
 
-fn write_oauth_recovery_log(directory: PathBuf) {
+fn write_sensitive_logs(directory: PathBuf, request_dump: bool) {
     let config = HostConfig {
         listen: ListenConfig {
             host: "127.0.0.1".to_owned(),
@@ -110,6 +160,7 @@ fn write_oauth_recovery_log(directory: PathBuf) {
                 max_file_size_mb: 1,
                 max_files: 1,
             },
+            request_dump,
         },
         system_update: SystemUpdateConfig::default(),
         drain_timeout_seconds: 30,
@@ -134,6 +185,12 @@ fn write_oauth_recovery_log(directory: PathBuf) {
         marker = OAUTH_RECOVERY_LOG_MARKER,
         "OAuth recovery test record"
     );
+    tracing::info!(
+        target: REQUEST_DUMP_LOG_TARGET,
+        marker = REQUEST_DUMP_LOG_MARKER,
+        authorization = REQUEST_DUMP_SECRET,
+        "request dump test record"
+    );
     drop(bundle);
 }
 
@@ -156,6 +213,18 @@ fn read_log_file_set(directory: &Path, file_prefix: &str) -> String {
         .collect()
 }
 
+fn log_file_set_exists(directory: &Path, file_prefix: &str) -> bool {
+    fs::read_dir(directory)
+        .expect("read log directory")
+        .filter_map(Result::ok)
+        .any(|entry| {
+            entry
+                .file_name()
+                .to_str()
+                .is_some_and(|name| name.starts_with(file_prefix))
+        })
+}
+
 fn json_target_field(target: &str) -> String {
     format!(r#""target":"{target}""#)
 }

backend/crates/gateway-store/src/postgres/execution.rs

@@ -728,7 +728,8 @@ impl ModelRequestRepository for PgExecutionStore {
                  upstream_connection_idle_ms = $47
              where id = $1 and outcome = 'running'
              returning id, client_api_key_ref, continuation_affinity_hash,
-                       continuation_requested, outcome, started_at, completed_at
+                       continuation_requested, provider_kind, upstream_transport,
+                       outcome, started_at, completed_at
            ), recovery_target as (
              select prior.id
              from model_requests prior
@@ -775,9 +776,47 @@ impl ModelRequestRepository for PgExecutionStore {
              where prior.id = target.id
                and prior.recovered_at is null
              returning prior.id
+           ), session_transport_recovery_target as (
+             select prior.id
+             from model_requests prior
+             cross join finalized current
+             where current.outcome = 'succeeded'
+               and current.provider_kind = 'openai'
+               and current.upstream_transport = 'http_sse'
+               and current.continuation_affinity_hash is not null
+               and prior.client_api_key_ref = current.client_api_key_ref
+               and prior.continuation_affinity_hash = current.continuation_affinity_hash
+               and prior.provider_kind = current.provider_kind
+               and prior.outcome = 'failed'
+               and prior.error_kind = 'upstream_unavailable'
+               and prior.upstream_transport = 'websocket'
+               and prior.upstream_send_state = 'ambiguous'
+               and prior.downstream_committed_at is null
+               and prior.recovered_at is null
+               and current.started_at >= prior.completed_at
+               and current.started_at <= prior.completed_at + interval '30 seconds'
+           ), session_transport_recovery_update as (
+             update model_requests prior
+             set recovery_attempt_count = prior.recovery_attempt_count + 1,
+                 recovery_request_id = current.id,
+                 recovered_at = current.completed_at,
+                 recovery_retry_delay_ms = greatest(
+                   0,
+                   floor(extract(epoch from (current.started_at - prior.completed_at)) * 1000)
+                 )::bigint,
+                 recovery_total_latency_ms = greatest(
+                   0,
+                   floor(extract(epoch from (current.completed_at - prior.completed_at)) * 1000)
+                 )::bigint
+             from finalized current
+             join session_transport_recovery_target target on true
+             where prior.id = target.id
+               and prior.recovered_at is null
+             returning prior.id
            )
            select (select count(*) from finalized)::bigint
-                + (select count(*) * 0 from recovery_update)::bigint",
+                + (select count(*) * 0 from recovery_update)::bigint
+                + (select count(*) * 0 from session_transport_recovery_update)::bigint",
         )
         .bind(&finalization.model_request_id)
         .bind(finalization.outcome.as_str())

backend/crates/gateway-store/src/postgres/observability/queries/ops.rs

@@ -128,7 +128,6 @@ fn push_request_error_predicates(
     range: ObservabilityRange,
     filter: &OpsErrorFilter,
 ) {
-    statement.push(" and mr.recovered_at is null");
     push_range(statement, "mr.completed_at", range);
     for (column, value) in [
         ("mr.client_api_key_ref", &filter.client_api_key_ref),
@@ -171,7 +170,6 @@ fn push_ops_event_predicates(
     range: ObservabilityRange,
     filter: &OpsErrorFilter,
 ) {
-    statement.push(" and (mr.id is null or mr.recovered_at is null)");
     push_range(statement, "oe.created_at", range);
     for (column, value) in [
         ("mr.client_api_key_ref", &filter.client_api_key_ref),

backend/crates/gateway-store/tests/postgres/execution.rs

@@ -719,6 +719,165 @@ async fn successful_new_chain_should_mark_recent_continuation_failure_recovered(
     database.close().await;
 }
 
+#[tokio::test]
+async fn successful_http_downgrade_should_mark_pending_websocket_failures_recovered() {
+    let Some(database) = TestDatabase::create("execution_transport_downgrade_recovered").await
+    else {
+        return;
+    };
+    let affinity_hash = "d".repeat(64);
+    let first_failure_at = Utc::now() - Duration::seconds(8);
+    let second_failure_at = first_failure_at + Duration::seconds(1);
+    seed_websocket_transport_failure(
+        &database.pool,
+        "req_websocket_failed_first",
+        &affinity_hash,
+        first_failure_at,
+    )
+    .await;
+    seed_websocket_transport_failure(
+        &database.pool,
+        "req_websocket_failed_second",
+        &affinity_hash,
+        second_failure_at,
+    )
+    .await;
+    seed_websocket_transport_failure(
+        &database.pool,
+        "req_websocket_failed_after_delivery",
+        &affinity_hash,
+        second_failure_at,
+    )
+    .await;
+    sqlx::query(
+        "update model_requests
+            set downstream_committed_at = completed_at
+          where id = 'req_websocket_failed_after_delivery'",
+    )
+    .execute(&database.pool)
+    .await
+    .expect("mark control failure delivered downstream");
+
+    let recovery_started_at = first_failure_at + Duration::seconds(4);
+    seed_transport_recovery_request(
+        &database.pool,
+        "req_http_fallback_succeeded",
+        &affinity_hash,
+        recovery_started_at,
+    )
+    .await;
+    let store = PgExecutionStore::new(database.pool.clone());
+    let mut recovery = successful_core_finalization("req_http_fallback_succeeded");
+    recovery.upstream_transport = Some("http_sse".to_owned());
+    recovery.downstream_committed_at = Some(std::time::SystemTime::now());
+    ExecutionStore::finalize_model_request(&store, recovery)
+        .await
+        .expect("finalize successful HTTP fallback");
+
+    type PersistedRecovery = (
+        String,
+        Option<String>,
+        Option<chrono::DateTime<Utc>>,
+        i32,
+        Option<i64>,
+        Option<i64>,
+    );
+    let recovered: Vec<PersistedRecovery> = sqlx::query_as(
+        "select id, recovery_request_id, recovered_at, recovery_attempt_count,
+                recovery_retry_delay_ms, recovery_total_latency_ms
+           from model_requests
+          where id like 'req_websocket_failed_%'
+          order by id",
+    )
+    .fetch_all(&database.pool)
+    .await
+    .expect("load WebSocket recovery observations");
+
+    assert_eq!(recovered.len(), 3);
+    let delivered = &recovered[0];
+    assert_eq!(delivered.0, "req_websocket_failed_after_delivery");
+    assert_eq!((&delivered.1, delivered.2, delivered.3), (&None, None, 0));
+    for (row, expected_delay_ms) in recovered[1..].iter().zip([4_000, 3_000]) {
+        assert_eq!(row.1.as_deref(), Some("req_http_fallback_succeeded"));
+        assert!(row.2.is_some());
+        assert_eq!(row.3, 1);
+        assert_eq!(row.4, Some(expected_delay_ms));
+        assert!(row.5.is_some_and(|latency| latency >= expected_delay_ms));
+    }
+    database.close().await;
+}
+
+#[tokio::test]
+async fn websocket_success_or_late_http_success_should_not_hide_transport_failure() {
+    let Some(database) = TestDatabase::create("execution_transport_downgrade_unresolved").await
+    else {
+        return;
+    };
+    let failure_at = Utc::now() - Duration::seconds(60);
+    let store = PgExecutionStore::new(database.pool.clone());
+
+    let websocket_affinity = "e".repeat(64);
+    seed_websocket_transport_failure(
+        &database.pool,
+        "req_websocket_control_failure",
+        &websocket_affinity,
+        failure_at,
+    )
+    .await;
+    seed_transport_recovery_request(
+        &database.pool,
+        "req_websocket_control_success",
+        &websocket_affinity,
+        failure_at + Duration::seconds(4),
+    )
+    .await;
+    ExecutionStore::finalize_model_request(
+        &store,
+        successful_core_finalization("req_websocket_control_success"),
+    )
+    .await
+    .expect("finalize successful WebSocket control request");
+
+    let late_affinity = "f".repeat(64);
+    seed_websocket_transport_failure(
+        &database.pool,
+        "req_websocket_late_failure",
+        &late_affinity,
+        failure_at,
+    )
+    .await;
+    seed_transport_recovery_request(
+        &database.pool,
+        "req_http_late_success",
+        &late_affinity,
+        failure_at + Duration::seconds(31),
+    )
+    .await;
+    let mut late_recovery = successful_core_finalization("req_http_late_success");
+    late_recovery.upstream_transport = Some("http_sse".to_owned());
+    ExecutionStore::finalize_model_request(&store, late_recovery)
+        .await
+        .expect("finalize late HTTP request");
+
+    type UnresolvedTransportFailure = (String, Option<String>, Option<chrono::DateTime<Utc>>, i32);
+    let unresolved: Vec<UnresolvedTransportFailure> = sqlx::query_as(
+        "select id, recovery_request_id, recovered_at, recovery_attempt_count
+               from model_requests
+              where id in ('req_websocket_control_failure', 'req_websocket_late_failure')
+              order by id",
+    )
+    .fetch_all(&database.pool)
+    .await
+    .expect("load unresolved transport failures");
+    assert_eq!(unresolved.len(), 2);
+    assert!(
+        unresolved
+            .iter()
+            .all(|row| row.1.is_none() && row.2.is_none() && row.3 == 0)
+    );
+    database.close().await;
+}
+
 #[tokio::test]
 async fn failed_or_late_new_chain_should_leave_continuation_failure_unresolved() {
     let Some(database) = TestDatabase::create("execution_continuation_unresolved").await else {
@@ -842,6 +1001,61 @@ async fn seed_recovery_request(
     .expect("prepare seeded recovery request");
 }
 
+async fn seed_websocket_transport_failure(
+    pool: &sqlx::PgPool,
+    id: &str,
+    affinity_hash: &str,
+    completed_at: chrono::DateTime<Utc>,
+) {
+    seed_running_request(pool, id)
+        .await
+        .expect("seed WebSocket transport failure request");
+    sqlx::query(
+        "update model_requests
+            set client_api_key_ref = 'key_transport_recovery',
+                client_transport = 'http_sse',
+                continuation_affinity_hash = $2,
+                attempt_count = 1, upstream_send_state = 'ambiguous',
+                upstream_transport = 'websocket', outcome = 'failed',
+                error_kind = 'upstream_unavailable',
+                provider_error_code = 'websocket_close_1000',
+                started_at = $3 - interval '1 second', deadline_at = $3,
+                completed_at = $3
+          where id = $1",
+    )
+    .bind(id)
+    .bind(affinity_hash)
+    .bind(completed_at)
+    .execute(pool)
+    .await
+    .expect("finalize seeded WebSocket transport failure");
+}
+
+async fn seed_transport_recovery_request(
+    pool: &sqlx::PgPool,
+    id: &str,
+    affinity_hash: &str,
+    started_at: chrono::DateTime<Utc>,
+) {
+    seed_running_request(pool, id)
+        .await
+        .expect("seed transport recovery request");
+    sqlx::query(
+        "update model_requests
+            set client_api_key_ref = 'key_transport_recovery',
+                client_transport = 'http_sse',
+                continuation_affinity_hash = $2,
+                started_at = $3, deadline_at = $3 + interval '1 minute'
+          where id = $1",
+    )
+    .bind(id)
+    .bind(affinity_hash)
+    .bind(started_at)
+    .execute(pool)
+    .await
+    .expect("prepare seeded transport recovery request");
+}
+
 async fn seed_running_request(pool: &sqlx::PgPool, id: &str) -> Result<(), sqlx::Error> {
     sqlx::query(
         "insert into model_requests (

backend/crates/gateway-store/tests/postgres/observability/mod.rs

@@ -258,7 +258,8 @@ async fn ops_search_should_treat_sql_wildcards_as_literals() {
 }
 
 #[tokio::test]
-async fn recovered_continuation_failure_should_be_hidden_from_default_business_metrics() {
+async fn recovered_continuation_failure_should_be_visible_in_ops_but_hidden_from_business_metrics()
+{
     let Some(database) = TestDatabase::create("observability_recovered_continuation").await else {
         return;
     };
@@ -318,8 +319,13 @@ async fn recovered_continuation_failure_should_be_hidden_from_default_business_m
             page_size: ObservabilityPageSize::new(10).expect("page size"),
         })
         .await
-        .expect("ops errors without recovered intermediates");
-    assert_eq!(errors.total, 0);
+        .expect("ops errors retain recovered failures");
+    assert_eq!(errors.total, 2);
+    assert!(errors.items.iter().all(|error| {
+        error.recovery_request_id.as_deref() == Some("req_observe_success")
+            && error.recovered_at.is_some()
+            && error.recovery_attempt_count == 1
+    }));
 
     let retained: (String, Option<String>) = sqlx::query_as(
         "select outcome, recovery_request_id

backend/crates/providers/openai/src/provider/execution.rs

@@ -186,6 +186,13 @@ pub(super) struct OpenAiSessionState {
     #[serde(default)]
     pub(super) client_turn_id: Option<String>,
     pub(super) continuation_scope: OpenAiContinuationScope,
+    /// 该状态来自同一业务请求的 metadata 检查点，只允许被紧随其后的恢复 attempt 消费。
+    #[serde(default, skip_serializing_if = "is_false")]
+    pub(super) retry_checkpoint: bool,
+}
+
+const fn is_false(value: &bool) -> bool {
+    !*value
 }
 
 #[derive(Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -232,6 +239,26 @@ pub(super) fn encode_openai_session_state(
         .map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent))
 }
 
+fn encode_openai_session_capture(
+    capture: &OpenAiSessionCapture,
+    retry_checkpoint: bool,
+) -> Result<ProviderSessionState, ProviderError> {
+    let Some(continuation_scope) = capture.continuation_scope else {
+        return Err(provider_error(
+            ProviderErrorKind::Protocol,
+            UpstreamSendState::Sent,
+        ));
+    };
+    encode_openai_session_state(OpenAiSessionState {
+        account_id: capture.account_id.clone(),
+        conversation_id: capture.conversation_id.clone(),
+        turn_state: capture.turn_state.clone(),
+        client_turn_id: capture.client_turn_id.clone(),
+        continuation_scope,
+        retry_checkpoint,
+    })
+}
+
 pub(super) fn attach_openai_session_update(
     events: &mut [ProviderEvent],
     capture: &mut Option<OpenAiSessionCapture>,
@@ -245,16 +272,7 @@ pub(super) fn attach_openai_session_update(
     let Some(capture) = capture.take() else {
         return;
     };
-    let Some(continuation_scope) = capture.continuation_scope else {
-        return;
-    };
-    let Ok(update) = encode_openai_session_state(OpenAiSessionState {
-        account_id: capture.account_id,
-        conversation_id: capture.conversation_id,
-        turn_state: capture.turn_state,
-        client_turn_id: capture.client_turn_id,
-        continuation_scope,
-    }) else {
+    let Ok(update) = encode_openai_session_capture(&capture, false) else {
         return;
     };
     events[terminal_index].attach_session_update(update);
@@ -563,6 +581,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                         WebSocketRecoveryContext {
                             policy,
                             requirement: request_transport_requirement,
+                            replay_checkpoint_available: false,
                             retry_count: websocket_retry_count,
                             max_retries: stream_max_retries,
                             request_id: context.request_id().as_str(),
@@ -594,6 +613,9 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             Err(failure.error)?;
             return;
         }
+        let mut turn_state_checkpoint_observed = response.transport
+            == CodexBackendTransport::WebSocket
+            && response.turn_state.is_some();
         if let Some(capture) = session_capture.as_mut() {
             capture.continuation_scope = Some(if capture.response_store {
                 OpenAiContinuationScope::Persisted
@@ -707,6 +729,49 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                     continue;
                 }
                 Err(mut failure) => {
+                    let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
+                    let rate_limits_changed = if updates.is_empty() {
+                        false
+                    } else {
+                        passive_quota_observation.observe(&updates);
+                        let update_headers = rate_limit_update_headers(&updates);
+                        observation_state.merge_rate_limit_headers(&update_headers)
+                    };
+                    let turn_state_merge = merge_turn_state_update(
+                        turn_state_updates.as_ref(),
+                        &mut session_capture,
+                        &mut observation_state,
+                    )
+                    .await;
+                    turn_state_checkpoint_observed |= turn_state_merge.is_some();
+                    let retry_checkpoint_update = if turn_state_checkpoint_observed
+                        && failure.websocket_checkpoint_retryable
+                    {
+                        session_capture
+                            .as_ref()
+                            .and_then(|capture| encode_openai_session_capture(capture, true).ok())
+                    } else {
+                        None
+                    };
+                    let mut observation_event = if rate_limits_changed
+                        || turn_state_merge.is_some()
+                        || retry_checkpoint_update.is_some()
+                    {
+                        observation_state
+                            .observation()
+                            .map(ProviderEvent::observation)
+                    } else {
+                        None
+                    };
+                    if let (Some(event), Some(update)) =
+                        (observation_event.as_mut(), retry_checkpoint_update)
+                    {
+                        event.attach_session_update(update);
+                    }
+                    let replay_checkpoint_available = observation_event
+                        .as_ref()
+                        .and_then(ProviderEvent::session_update)
+                        .is_some();
                     if failure.websocket_transport_retryable
                         && response_transport == CodexBackendTransport::WebSocket
                         && (matches!(
@@ -719,6 +784,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                             WebSocketRecoveryContext {
                                 policy: WebSocketFailurePolicy::Budgeted,
                                 requirement: request_transport_requirement,
+                                replay_checkpoint_available,
                                 retry_count: websocket_retry_count,
                                 max_retries: stream_max_retries,
                                 request_id: context.request_id().as_str(),
@@ -731,15 +797,8 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                             },
                         );
                     }
-                    let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
-                    if !updates.is_empty() {
-                        passive_quota_observation.observe(&updates);
-                        let update_headers = rate_limit_update_headers(&updates);
-                        if observation_state.merge_rate_limit_headers(&update_headers)
-                            && let Some(observation) = observation_state.observation()
-                        {
-                            yield ProviderEvent::observation(observation);
-                        }
+                    if let Some(event) = observation_event {
+                        yield event;
                     }
                     if allows_account_state_mutation {
                         synchronize_passive_quota(
@@ -763,16 +822,14 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                 passive_quota_observation.observe(&updates);
                 observation_state.merge_rate_limit_headers(&rate_limit_update_headers(&updates))
             };
-            let turn_state_changed = if let Some(updates) = turn_state_updates.as_ref()
-                && let Some(turn_state) = updates.lock().await.take()
-            {
-                if let Some(capture) = session_capture.as_mut() {
-                    capture.turn_state = Some(turn_state.clone());
-                }
-                observation_state.merge_client_header("x-codex-turn-state", &turn_state)
-            } else {
-                false
-            };
+            let turn_state_merge = merge_turn_state_update(
+                turn_state_updates.as_ref(),
+                &mut session_capture,
+                &mut observation_state,
+            )
+            .await;
+            turn_state_checkpoint_observed |= turn_state_merge.is_some();
+            let turn_state_changed = turn_state_merge.unwrap_or(false);
             let first_event_changed =
                 observation_state.observe_stream_chunk(&chunk, output_started_at);
             let chunk_len = chunk.len();
@@ -949,16 +1006,13 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             apply_failure(&failure_context, &active_account, failure)
             .await;
         }
-        let turn_state_changed = if let Some(updates) = turn_state_updates.as_ref()
-            && let Some(turn_state) = updates.lock().await.take()
-        {
-            if let Some(capture) = session_capture.as_mut() {
-                capture.turn_state = Some(turn_state.clone());
-            }
-            observation_state.merge_client_header("x-codex-turn-state", &turn_state)
-        } else {
-            false
-        };
+        let turn_state_changed = merge_turn_state_update(
+            turn_state_updates.as_ref(),
+            &mut session_capture,
+            &mut observation_state,
+        )
+        .await
+        .unwrap_or(false);
         attach_openai_session_update(&mut events, &mut session_capture);
         let completed = events
             .iter()
@@ -1024,3 +1078,16 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
         }
     })
 }
+
+async fn merge_turn_state_update(
+    updates: Option<&CodexTurnStateUpdate>,
+    session_capture: &mut Option<OpenAiSessionCapture>,
+    observation_state: &mut OpenAiResponseObservationState,
+) -> Option<bool> {
+    let updates = updates?;
+    let turn_state = updates.lock().await.take()?;
+    if let Some(capture) = session_capture.as_mut() {
+        capture.turn_state = Some(turn_state.clone());
+    }
+    Some(observation_state.merge_client_header("x-codex-turn-state", &turn_state))
+}

backend/crates/providers/openai/src/provider/failure.rs

@@ -39,6 +39,8 @@ pub fn openai_failure_affects_account_score(error: &ProviderError) -> bool {
 pub(super) struct MappedProviderFailure {
     pub(super) error: ProviderError,
     pub(super) websocket_transport_retryable: bool,
+    /// 兼容少数链路在 metadata 后、终态前关闭；仅在捕获 turn-state 时允许安全恢复。
+    pub(super) websocket_checkpoint_retryable: bool,
     pub(super) account_failure: Option<CodexAccountFailure>,
     /// 原始上游错误描述，仅在凭据错误状态下持久化。
     pub(super) error_message: Option<String>,
@@ -54,6 +56,7 @@ impl MappedProviderFailure {
         Self {
             error,
             websocket_transport_retryable: false,
+            websocket_checkpoint_retryable: false,
             account_failure: None,
             error_message: None,
             cyber_policy_failure: false,
@@ -451,6 +454,8 @@ pub(super) enum WebSocketFailurePolicy {
 pub(super) struct WebSocketRecoveryContext<'a> {
     pub(super) policy: WebSocketFailurePolicy,
     pub(super) requirement: TransportRequirement,
+    /// 本 attempt 已从 metadata 捕获可恢复同一请求的 turn-state 检查点。
+    pub(super) replay_checkpoint_available: bool,
     pub(super) retry_count: u32,
     pub(super) max_retries: u32,
     pub(super) request_id: &'a str,
@@ -505,10 +510,11 @@ pub(super) fn apply_websocket_recovery_policy(
     context: WebSocketRecoveryContext<'_>,
 ) {
     let send_state = failure.error.send_state();
-    if matches!(
+    let post_send = matches!(
         send_state,
         UpstreamSendState::Sent | UpstreamSendState::Ambiguous
-    ) {
+    );
+    if post_send && !context.replay_checkpoint_available {
         // payload 已经发送或发送结果不确定时，代理不能用同一业务输入自动重放。
         // 当前请求保持终态失败；恢复状态只裁决客户端主动发起的下一次请求。
         let session_recovery_transition = match context.session_recovery_probe.take() {
@@ -555,6 +561,11 @@ pub(super) fn apply_websocket_recovery_policy(
     // 传输恢复必须保持原账号可调度；最终 HTTP attempt 若仍失败，再按真实 HTTP
     // 结果更新账号健康度，避免中间 WS 错误把同账号钉选提前冷却掉。
     failure.account_failure = None;
+    if post_send {
+        // Core 对 Ambiguous 默认仍然 fail-closed；只有同一 attempt 的 session
+        // checkpoint 与这个 Provider proof 同时存在时，才允许同账号传输恢复。
+        failure.error.set_replay_safe();
+    }
     let fallback_now = context.policy == WebSocketFailurePolicy::ImmediateFallback
         || context.retry_count >= context.max_retries;
     if !fallback_now {
@@ -583,6 +594,7 @@ pub(super) fn apply_websocket_recovery_policy(
             websocket_max_retries = context.max_retries,
             websocket_retry_delay_ms = u64::try_from(delay.as_millis()).unwrap_or(u64::MAX),
             transport_requirement = context.requirement.as_str(),
+            replay_checkpoint_available = context.replay_checkpoint_available,
             "OpenAI upstream WebSocket failed; retrying the same account"
         );
         return;
@@ -621,6 +633,7 @@ pub(super) fn apply_websocket_recovery_policy(
             WebSocketFailurePolicy::Budgeted => "retry_budget_exhausted",
             WebSocketFailurePolicy::ImmediateFallback => "upgrade_required",
         },
+        replay_checkpoint_available = context.replay_checkpoint_available,
         session_affinity_present = context.session_affinity_key.is_some(),
         session_affinity_key_hash = context.session_affinity_key_hash.unwrap_or(""),
         session_transport_action =
@@ -673,14 +686,31 @@ pub(super) fn continuation_replay_required_error(reason: &'static str) -> Provid
 pub(super) fn map_stream_error(error: CodexClientError) -> MappedProviderFailure {
     let allows_pre_delivery_retry = stream_transport_allows_pre_delivery_retry(&error);
     let websocket_failure = error.transport() == Some(CodexBackendTransport::WebSocket);
+    let websocket_checkpoint_retryable = websocket_metadata_close_can_resume(&error);
     let mut failure = map_client_error(error, UpstreamSendState::Sent, false);
     failure.websocket_transport_retryable = allows_pre_delivery_retry && websocket_failure;
+    failure.websocket_checkpoint_retryable =
+        allows_pre_delivery_retry && websocket_checkpoint_retryable;
     if allows_pre_delivery_retry && !websocket_failure {
         failure.error = failure.error.with_pre_delivery_retry();
     }
     failure
 }
 
+fn websocket_metadata_close_can_resume(error: &CodexClientError) -> bool {
+    let CodexClientError::WebSocket(error) = error else {
+        return false;
+    };
+    let Some(close) = error.close_before_terminal() else {
+        return false;
+    };
+    close.code() == Some(1000)
+        && matches!(
+            close.last_event_type(),
+            Some("codex.response.metadata" | "response.metadata")
+        )
+}
+
 pub(super) fn stream_transport_allows_pre_delivery_retry(error: &CodexClientError) -> bool {
     match error {
         CodexClientError::Http(_)
@@ -1068,6 +1098,7 @@ pub(super) fn map_upstream_failure(
     MappedProviderFailure {
         error,
         websocket_transport_retryable: false,
+        websocket_checkpoint_retryable: false,
         account_failure: account_failure(
             category,
             failure.retry_after_seconds,

backend/crates/providers/openai/src/provider/mod.rs

@@ -93,7 +93,7 @@ use crate::transport::{
     CODEX_RESPONSES_PATH, CodexAccountSelectionTelemetry, CodexBackendClient,
     CodexBackendJsonResponse, CodexBackendStreamingResponse, CodexBackendTransport,
     CodexClientError, CodexRateLimitUpdates, CodexRequestContext, CodexResponseMetadata,
-    CodexTransportMetrics, CodexUpstreamDiagnostics, CodexWebSocketPool,
+    CodexTransportMetrics, CodexTurnStateUpdate, CodexUpstreamDiagnostics, CodexWebSocketPool,
     WebSocketConnectionPreference, endpoint_url,
 };
 
@@ -320,15 +320,22 @@ impl Provider for CodexProvider {
         if let Some(identity) = &self.session_identity {
             identity.prepare_local_conversation(&mut upstream_request);
         }
-        if previous_session.as_ref().is_some_and(|state| {
-            same_client_turn(
-                state.client_turn_id.as_deref(),
-                upstream_request.client_turn_id.as_deref(),
-            )
-        }) {
-            upstream_request.turn_state = previous_session
-                .as_ref()
-                .and_then(|state| state.turn_state.clone());
+        if let Some(previous_session) = previous_session.as_ref() {
+            upstream_request.turn_state = if previous_session.retry_checkpoint
+                || same_client_turn(
+                    previous_session.client_turn_id.as_deref(),
+                    upstream_request.client_turn_id.as_deref(),
+                ) {
+                // 客户端可能携带 metadata-close 失败后才拿到的新状态；同一
+                // turn 内显式回传值比最后一次成功响应保存的状态更新。同请求
+                // 内部恢复由一次性 retry checkpoint 明确证明，不依赖客户端 turn ID。
+                upstream_request
+                    .turn_state
+                    .take()
+                    .or_else(|| previous_session.turn_state.clone())
+            } else {
+                None
+            };
         }
         let session_affinity =
             derive_codex_session_affinity(&upstream_request, context.client_api_key_ref());

backend/crates/providers/openai/src/transport/client.rs

@@ -24,7 +24,9 @@ use serde_json::{Value, map::Map};
 use thiserror::Error;
 use uuid::Uuid;
 
-use crate::transport::protocol::responses::{CodexResponsesRequest, TransportRequirement};
+use crate::transport::protocol::responses::{
+    CodexResponsesRequest, TransportRequirement, X_CODEX_TURN_STATE_CLIENT_METADATA_KEY,
+};
 
 use super::diagnostics::{CodexUpstreamDiagnostics, CodexUpstreamFailure, CodexUpstreamSendPhase};
 use super::response_meta::CodexResponseMetadata;
@@ -546,7 +548,7 @@ pub struct CodexBackendStreamingResponse {
     pub transport: CodexBackendTransport,
     /// WebSocket 响应所绑定的连接；HTTP transport 为 `None`。
     pub websocket_connection_id: Option<Uuid>,
-    /// 响应头里的最新 turn state。
+    /// 响应头或 metadata 事件里最先确认的 turn state。
     pub turn_state: Option<String>,
     /// 上游透传的 `set-cookie` 列表。
     pub set_cookie_headers: Vec<String>,
@@ -760,32 +762,33 @@ pub(super) async fn read_error_response_body(
 
 pub(super) fn websocket_upstream_request(request: &CodexResponsesRequest) -> CodexResponsesRequest {
     let mut request = request.clone();
-    project_responses_lite_to_ws_metadata(&mut request);
-    stamp_ws_stream_request_start_ms(&mut request);
+    project_websocket_client_metadata(&mut request);
     request
 }
 
-fn project_responses_lite_to_ws_metadata(request: &mut CodexResponsesRequest) {
-    let Some(responses_lite) = request.responses_lite.clone() else {
-        return;
-    };
-    let mut metadata = match request.client_metadata() {
-        Some(Value::Object(metadata)) => metadata.clone(),
-        None => Map::new(),
-        Some(_) => return,
-    };
-    metadata
-        .entry(WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY.to_string())
-        .or_insert(Value::String(responses_lite));
-    request.set_client_metadata(Some(Value::Object(metadata)));
-}
-
-fn stamp_ws_stream_request_start_ms(request: &mut CodexResponsesRequest) {
+fn project_websocket_client_metadata(request: &mut CodexResponsesRequest) {
     let mut metadata = match request.client_metadata() {
         Some(Value::Object(metadata)) => metadata.clone(),
         None => Map::new(),
         Some(_) => return,
     };
+    if let Some(responses_lite) = request.responses_lite.clone() {
+        metadata
+            .entry(WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY.to_owned())
+            .or_insert(Value::String(responses_lite));
+    }
+    let turn_state = request.turn_state.clone();
+    match turn_state {
+        Some(turn_state) => {
+            metadata.insert(
+                X_CODEX_TURN_STATE_CLIENT_METADATA_KEY.to_owned(),
+                Value::String(turn_state),
+            );
+        }
+        None => {
+            metadata.remove(X_CODEX_TURN_STATE_CLIENT_METADATA_KEY);
+        }
+    }
     metadata.insert(
         X_CODEX_WS_STREAM_REQUEST_START_MS_CLIENT_METADATA_KEY.to_string(),
         Value::String(now_unix_timestamp_millis().to_string()),

backend/crates/providers/openai/src/transport/client_sse.rs

@@ -37,10 +37,11 @@ use crate::transport::{
     response_meta,
     websocket::{
         CodexWebSocketConnection, CodexWebSocketExchangeError, CodexWebSocketPool,
-        CodexWebSocketPoolKey, CodexWebSocketStreamingExchange, DEFAULT_INITIAL_EVENT_TIMEOUT,
-        WEBSOCKET_FAST_PATH_BUDGET, WebSocketFastPath, WebSocketOriginBreaker,
-        WebSocketPoolDecision, execute_prepared_response_create_request_stream,
-        post_send_ambiguous, prepare_response_create_request_with_pool, websocket_audit_dir,
+        CodexWebSocketPoolKey, CodexWebSocketRequest, CodexWebSocketStreamingExchange,
+        DEFAULT_INITIAL_EVENT_TIMEOUT, WEBSOCKET_FAST_PATH_BUDGET, WebSocketFastPath,
+        WebSocketOriginBreaker, WebSocketPoolDecision,
+        execute_prepared_response_create_request_stream, post_send_ambiguous,
+        prepare_response_create_request_with_pool, websocket_audit_dir,
         write_websocket_audit_artifact_from_env,
     },
 };
@@ -104,11 +105,26 @@ impl CodexBackendClient {
         upstream_body.insert("stream".to_owned(), serde_json::Value::Bool(true));
         let body =
             serde_json::to_vec(&upstream_body).map_err(CodexClientError::RequestBodyEncode)?;
+        let endpoint = endpoint_url(&self.base_url, CODEX_RESPONSES_PATH);
+        tracing::info!(
+            target: "request_dump",
+            provider = "openai",
+            request_id = %context.request_id,
+            account_id = context.account_id.unwrap_or_default(),
+            request_direction = "upstream",
+            request_transport = "http_sse",
+            request_method = "POST",
+            request_url = %endpoint,
+            request_headers = ?headers,
+            request_body = %String::from_utf8_lossy(&body),
+            contains_sensitive_data = true,
+            "request dump (unredacted)"
+        );
         let body = zstd::stream::encode_all(std::io::Cursor::new(body), 3)
             .map_err(CodexClientError::RequestCompression)?;
         let response = self
             .client
-            .post(endpoint_url(&self.base_url, CODEX_RESPONSES_PATH))
+            .post(endpoint)
             .headers(headers)
             .header(CONTENT_ENCODING, HeaderValue::from_static("zstd"))
             .body(body)
@@ -425,6 +441,7 @@ impl CodexBackendClient {
                     prepared,
                 } = *route;
                 let delivery_wait_started_at = Instant::now();
+                log_unredacted_websocket_request(&websocket_request, context);
                 let mut exchange = match execute_prepared_response_create_request_stream(
                     &websocket_request,
                     prepared,
@@ -495,6 +512,7 @@ impl CodexBackendClient {
                                         route: PreparedResponseRoute::WebSocket(route),
                                         ..
                                     }) => {
+                                        log_unredacted_websocket_request(&route.request, context);
                                         match execute_prepared_response_create_request_stream(
                                             &route.request,
                                             route.prepared,
@@ -682,10 +700,12 @@ impl CodexBackendClient {
             .local_conversation_id
             .as_deref()
             .or(request.previous_response_id())?;
-        Some(
-            CodexWebSocketPoolKey::new(&self.base_url, account_id, conversation_id)
-                .with_connection_profile(connection_profile),
-        )
+        let mut key = CodexWebSocketPoolKey::new(&self.base_url, account_id, conversation_id)
+            .with_connection_profile(connection_profile);
+        if let Some(connection_id) = request.downstream_websocket_connection_id.as_deref() {
+            key = key.with_downstream_connection_id(connection_id);
+        }
+        Some(key)
     }
 
     /// 获取后端模型目录条目。
@@ -889,6 +909,26 @@ impl CodexBackendClient {
     }
 }
 
+fn log_unredacted_websocket_request(
+    request: &CodexWebSocketRequest,
+    context: CodexRequestContext<'_>,
+) {
+    tracing::info!(
+        target: "request_dump",
+        provider = "openai",
+        request_id = %context.request_id,
+        account_id = context.account_id.unwrap_or_default(),
+        request_direction = "upstream",
+        request_transport = "websocket",
+        request_method = "GET + response.create",
+        request_url = request.connection().endpoint(),
+        request_headers = ?request.connection().headers(),
+        request_payload = request.payload_text(),
+        contains_sensitive_data = true,
+        "request dump (unredacted)"
+    );
+}
+
 /// 首个可投递帧前的交付边界结果。
 enum DeliveryBoundary {
     /// 已越过边界，可开始向下游投递。

backend/crates/providers/openai/src/transport/protocol/responses.rs

@@ -10,6 +10,8 @@ use serde_json::{Map, Value};
 
 /// 官方 Codex 客户端据此触发完整历史重放的稳定错误码。
 pub(crate) const PREVIOUS_RESPONSE_NOT_FOUND_CODE: &str = "previous_response_not_found";
+/// Responses WebSocket 用于回传同一 turn 不透明状态的官方 client metadata 键。
+pub(crate) const X_CODEX_TURN_STATE_CLIENT_METADATA_KEY: &str = "x-codex-turn-state";
 /// 本地生成 history unavailable 错误时使用的官方提示文本。
 pub(crate) const PREVIOUS_RESPONSE_NOT_FOUND_MESSAGE: &str =
     "Previous response was not found. Retrying the full request.";
@@ -82,6 +84,8 @@ pub struct CodexResponsesRequest {
     pub memgen_request: Option<String>,
     /// Codex window ID。
     pub codex_window_id: Option<String>,
+    /// 代理分配的下游 WebSocket 连接 ID，仅用于隔离本地连接池通道。
+    pub downstream_websocket_connection_id: Option<String>,
     /// 父线程 ID。
     pub parent_thread_id: Option<String>,
     /// 已知 previous response 的持久化范围，仅用于本地 transport 校验。
@@ -519,6 +523,7 @@ impl CodexResponsesRequest {
             responses_lite: None,
             memgen_request: None,
             codex_window_id: None,
+            downstream_websocket_connection_id: None,
             parent_thread_id: None,
             previous_response_scope: None,
         }

backend/crates/providers/openai/src/transport/protocol/websocket.rs

@@ -110,9 +110,9 @@ fn is_internal_websocket_event(event: &str) -> bool {
     event == "codex.rate_limits"
 }
 
-/// 提取 `response.metadata` 帧中的字符串响应头。
+/// 提取 Responses WebSocket metadata 帧中的字符串响应头。
 pub fn websocket_metadata_headers(value: &Value) -> Vec<(String, String)> {
-    if value.get("type").and_then(Value::as_str) != Some("response.metadata") {
+    if !is_websocket_metadata_event(websocket_event_type(value)) {
         return Vec::new();
     }
     value
@@ -124,24 +124,18 @@ pub fn websocket_metadata_headers(value: &Value) -> Vec<(String, String)> {
         .collect()
 }
 
-/// 从 `response.metadata` 帧中提取 `x-codex-turn-state`。
+/// 从 Responses WebSocket metadata 帧中提取 `x-codex-turn-state`。
 pub fn websocket_metadata_turn_state(value: &Value) -> Option<String> {
-    if value.get("type").and_then(Value::as_str) != Some("response.metadata") {
-        return None;
-    }
     websocket_metadata_headers(value)
         .into_iter()
         .find_map(|(name, value)| {
             name.eq_ignore_ascii_case("x-codex-turn-state")
                 .then_some(value)
         })
-        .or_else(|| {
-            value
-                .pointer("/metadata/turn_state")
-                .or_else(|| value.get("turn_state"))
-                .and_then(Value::as_str)
-                .map(ToString::to_string)
-        })
+}
+
+fn is_websocket_metadata_event(event: Option<&str>) -> bool {
+    matches!(event, Some("response.metadata" | "codex.response.metadata"))
 }
 
 /// 上游 WebSocket 连接寿命限制错误码。

backend/crates/providers/openai/src/transport/request.rs

@@ -7,9 +7,12 @@ use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
 use serde_json::{Map, Value};
 use sha2::{Digest, Sha256};
 
-use crate::transport::protocol::responses::CodexResponsesRequest;
+use crate::transport::protocol::responses::{
+    CodexResponsesRequest, X_CODEX_TURN_STATE_CLIENT_METADATA_KEY,
+};
 
 const PASSTHROUGH_HEADERS_CONTEXT_KEY: &str = "opaque_request_headers";
+const TURN_ID_CLIENT_METADATA_KEY: &str = "turn_id";
 const THREAD_SPAWN_SUBAGENT_KIND: &str = "thread_spawn";
 const THREAD_SPAWN_CONVERSATION_PREFIX: &str = "thread-spawn:";
 const UNSUPPORTED_CODEX_RESPONSES_FIELDS: &[&str] = &["max_output_tokens", "temperature"];
@@ -144,8 +147,16 @@ struct ExtractedRequestContext {
 
 impl ExtractedRequestContext {
     fn from_body(body: &Map<String, Value>) -> Self {
+        let client_metadata = body.get("client_metadata").and_then(Value::as_object);
         Self {
-            turn_state: body_string(body, "turnState"),
+            // 官方 downstream WebSocket 无法逐帧更新 HTTP header，因此把
+            // response.metadata 返回的 turn state 放回下一帧 client_metadata。
+            // Provider 仍会在账号/turn 归属确定后决定是否允许复用该状态。
+            turn_state: body_string(body, "turnState").or_else(|| {
+                client_metadata.and_then(|metadata| {
+                    string_value(metadata.get(X_CODEX_TURN_STATE_CLIENT_METADATA_KEY))
+                })
+            }),
             turn_metadata: body_string(body, "turnMetadata"),
             beta_features: body_string(body, "betaFeatures"),
             version: body_string(body, "version"),
@@ -156,15 +167,15 @@ impl ExtractedRequestContext {
             session_id: body_string(body, "session_id"),
             thread_id: body_string(body, "thread_id"),
             client_request_id: body_string(body, "x-client-request-id"),
-            turn_id: body_string(body, "turn_id"),
+            turn_id: body_string(body, "turn_id").or_else(|| {
+                client_metadata
+                    .and_then(|metadata| string_value(metadata.get(TURN_ID_CLIENT_METADATA_KEY)))
+            }),
             // Responses Lite 在 WebSocket body 中的这个键是官方 header 投影，
             // 不是普通上下文字段的 metadata 回退。
-            responses_lite: body
-                .get("client_metadata")
-                .and_then(Value::as_object)
-                .and_then(|metadata| {
-                    string_value(metadata.get(WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY))
-                }),
+            responses_lite: client_metadata.and_then(|metadata| {
+                string_value(metadata.get(WS_REQUEST_HEADER_RESPONSES_LITE_CLIENT_METADATA_KEY))
+            }),
             // Memory consolidation 只由官方请求头提供；body 中同名字段不参与
             // transport 事实提取。
             memgen_request: None,
@@ -511,6 +522,9 @@ fn apply_protocol_context(request: &mut CodexResponsesRequest, context: &Map<Str
         .or_else(|| request.include_timing_metrics.take());
     request.codex_window_id =
         context_string(context, "codex_window_id").or_else(|| request.codex_window_id.take());
+    request.downstream_websocket_connection_id =
+        context_string(context, "downstream_websocket_connection_id")
+            .or_else(|| request.downstream_websocket_connection_id.take());
     request.parent_thread_id =
         context_string(context, "parent_thread_id").or_else(|| request.parent_thread_id.take());
     let prompt_cache_key = request.prompt_cache_key().map(ToOwned::to_owned);

backend/crates/providers/openai/src/transport/websocket/exchange/mod.rs

@@ -31,7 +31,7 @@ pub struct CodexWebSocketStreamingExchange {
     pub(crate) websocket_connection_id: Uuid,
     /// 由 WebSocket 事件转换出的 live SSE 字节流。
     pub body: CodexWebSocketSseStream,
-    /// 上游 metadata 帧中的最新 turn state。
+    /// 上游为本次响应返回的首个 turn state。
     pub turn_state: Option<String>,
     /// 上游握手响应里的 `set-cookie` 列表。
     pub set_cookie_headers: Vec<String>,
@@ -39,7 +39,7 @@ pub struct CodexWebSocketStreamingExchange {
     pub rate_limit_headers: Vec<(String, String)>,
     /// 上游内部 `codex.rate_limits` 事件里的结构化动态更新。
     pub rate_limit_updates: CodexWebSocketRateLimitUpdates,
-    /// 上游内部 metadata 事件里的动态 turn state。
+    /// 上游内部 metadata 事件里的首个动态 turn state。
     pub turn_state_update: CodexWebSocketTurnStateUpdate,
     /// WebSocket 连接池决策。
     pub pool_decision: Option<WebSocketPoolDecision>,
@@ -63,5 +63,10 @@ pub(super) fn reusable_websocket_metadata(
     mut metadata: CodexWebSocketConnectionMetadata,
 ) -> CodexWebSocketConnectionMetadata {
     metadata.rate_limit_headers.clear();
+    metadata.turn_state = None;
+    metadata
+        .response_metadata
+        .client_headers
+        .retain(|(name, _)| !name.eq_ignore_ascii_case("x-codex-turn-state"));
     metadata
 }

backend/crates/providers/openai/src/transport/websocket/exchange/reducer.rs

@@ -21,7 +21,6 @@ pub(in crate::transport::websocket) enum WebSocketTerminalKind {
 
 pub(super) enum ExchangeAction {
     RateLimits(events::ParsedRateLimits),
-    TurnState(String),
     Forward {
         frame: String,
         terminal: Option<WebSocketTerminalKind>,
@@ -32,6 +31,7 @@ pub(super) enum ExchangeAction {
 pub(super) struct ReducedWebSocketEvent {
     pub(super) action: ExchangeAction,
     pub(super) diagnostic_event_type: Option<String>,
+    pub(super) turn_state_update: Option<String>,
 }
 
 pub(super) fn reduce_websocket_event(
@@ -45,6 +45,7 @@ pub(super) fn reduce_websocket_event(
         return Ok(ReducedWebSocketEvent {
             action: ExchangeAction::Ignore,
             diagnostic_event_type: None,
+            turn_state_update: None,
         });
     };
     let diagnostic_event_type = diagnostic_event_type(websocket_event_type(&value));
@@ -54,6 +55,7 @@ pub(super) fn reduce_websocket_event(
         return Ok(ReducedWebSocketEvent {
             action: ExchangeAction::RateLimits(parsed),
             diagnostic_event_type,
+            turn_state_update: None,
         });
     }
 
@@ -61,13 +63,13 @@ pub(super) fn reduce_websocket_event(
         &mut metadata.response_metadata,
         websocket_metadata_headers(&value),
     );
-    if let Some(turn_state) = websocket_metadata_turn_state(&value) {
+    let turn_state_update = websocket_metadata_turn_state(&value).and_then(|turn_state| {
+        if metadata.turn_state.is_some() {
+            return None;
+        }
         metadata.turn_state = Some(turn_state.clone());
-        return Ok(ReducedWebSocketEvent {
-            action: ExchangeAction::TurnState(turn_state),
-            diagnostic_event_type,
-        });
-    }
+        Some(turn_state)
+    });
 
     let event = websocket_event_type(&value);
     if event == Some("response.completed")
@@ -89,6 +91,7 @@ pub(super) fn reduce_websocket_event(
     Ok(ReducedWebSocketEvent {
         action,
         diagnostic_event_type,
+        turn_state_update,
     })
 }
 

backend/crates/providers/openai/src/transport/websocket/exchange/stream.rs

@@ -271,15 +271,17 @@ async fn forward_websocket_response_stream(state: WebSocketStreamForwardState) {
         if let Some(event_type) = reduced.diagnostic_event_type {
             last_event_type = Some(event_type);
         }
+        if let Some(turn_state) = reduced.turn_state_update {
+            let mut pending = turn_state_update.lock().await;
+            if pending.is_none() {
+                *pending = Some(turn_state);
+            }
+        }
         let (frame, terminal) = match reduced.action {
             ExchangeAction::RateLimits(rate_limits) => {
                 rate_limit_updates.lock().await.push(rate_limits);
                 continue;
             }
-            ExchangeAction::TurnState(turn_state) => {
-                *turn_state_update.lock().await = Some(turn_state);
-                continue;
-            }
             ExchangeAction::Forward { frame, terminal } => (frame, terminal),
             ExchangeAction::Ignore => continue,
         };

backend/crates/providers/openai/src/transport/websocket/pool/state.rs

@@ -23,6 +23,7 @@ pub struct CodexWebSocketPoolKey {
     account_id: String,
     conversation_id: String,
     connection_profile: String,
+    downstream_connection_id: String,
 }
 
 impl CodexWebSocketPoolKey {
@@ -37,6 +38,7 @@ impl CodexWebSocketPoolKey {
             account_id: account_id.into(),
             conversation_id: conversation_id.into(),
             connection_profile: String::new(),
+            downstream_connection_id: String::new(),
         }
     }
 
@@ -46,6 +48,15 @@ impl CodexWebSocketPoolKey {
         self
     }
 
+    /// 隔离同一逻辑会话中由不同下游 WebSocket 驱动的并发响应链。
+    pub(crate) fn with_downstream_connection_id(
+        mut self,
+        connection_id: impl Into<String>,
+    ) -> Self {
+        self.downstream_connection_id = connection_id.into();
+        self
+    }
+
     pub(crate) fn account_id(&self) -> &str {
         &self.account_id
     }
@@ -60,6 +71,7 @@ impl CodexWebSocketPoolKey {
             self.account_id.as_str(),
             self.conversation_id.as_str(),
             self.connection_profile.as_str(),
+            self.downstream_connection_id.as_str(),
         ])
     }
 

backend/crates/providers/openai/tests/provider/contract.rs

@@ -461,6 +461,7 @@ fn contract_account_scope() -> Arc<FrozenAccountScope> {
         "acct_websocket_cooldown",
         "acct_websocket_fast_path",
         "acct_websocket_busy_replay",
+        "acct_websocket_metadata_close",
         "acct_websocket_turn_state",
     ]
     .into_iter()
@@ -664,6 +665,7 @@ async fn capture_turn_state_request(
     previous_turn_id: Option<&str>,
     current_turn_id: Option<&str>,
     client_turn_state: Option<&str>,
+    retry_checkpoint: bool,
 ) -> wiremock::Request {
     let account_id = "acct_session_affinity";
     let store = Arc::new(MemoryAccountStore::default());
@@ -688,6 +690,9 @@ async fn capture_turn_state_request(
     if let Some(previous_turn_id) = previous_turn_id {
         session_state.insert("client_turn_id".to_owned(), json!(previous_turn_id));
     }
+    if retry_checkpoint {
+        session_state.insert("retry_checkpoint".to_owned(), json!(true));
+    }
     let mut protocol_context = Map::from_iter([("use_websocket".to_owned(), json!(false))]);
     if let Some(current_turn_id) = current_turn_id {
         protocol_context.insert("turn_id".to_owned(), json!(current_turn_id));
@@ -2519,6 +2524,114 @@ async fn websocket_turn_state_metadata_is_exposed_through_response_observation()
     assert!(observed_turn_state);
 }
 
+#[tokio::test]
+async fn websocket_turn_state_metadata_close_requests_checkpointed_retry() {
+    const ACCOUNT_ID: &str = "acct_websocket_metadata_close";
+    const TURN_STATE: &str = "turn-state-before-close";
+
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, ACCOUNT_ID).await;
+    let listener = TcpListener::bind("127.0.0.1:0")
+        .await
+        .expect("bind WebSocket listener");
+    let base_url = format!(
+        "http://{}",
+        listener.local_addr().expect("listener address")
+    );
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener
+            .accept()
+            .await
+            .expect("accept WebSocket connection");
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        let _request = websocket
+            .next()
+            .await
+            .expect("WebSocket request")
+            .expect("valid WebSocket request");
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "codex.response.metadata",
+                    "headers": {"x-codex-turn-state": TURN_STATE}
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .expect("send response metadata");
+        websocket
+            .close(Some(CloseFrame {
+                code: CloseCode::Normal,
+                reason: "".into(),
+            }))
+            .await
+            .expect("close before terminal response");
+    });
+
+    let operation = Operation::Generate(generate_with_persisted_session_context(
+        ACCOUNT_ID,
+        "conversation-metadata-close",
+        "session-metadata-close",
+        "turn-metadata-close",
+    ));
+    let mut stream = provider_with_base_url(&store, base_url)
+        .execute(
+            planned_request("openai", operation),
+            context("req_websocket_metadata_close", CancellationToken::new()),
+        )
+        .await
+        .expect("prepare WebSocket provider stream");
+    let mut observed_turn_state = false;
+    let mut replay_checkpoint = None;
+    let error = loop {
+        match stream.next().await {
+            Some(Ok(event)) => {
+                if let Some(observation) = event.response_observation() {
+                    observed_turn_state |= observation.client_headers().iter().any(|header| {
+                        header.name().eq_ignore_ascii_case("x-codex-turn-state")
+                            && header.value().as_ref() == TURN_STATE.as_bytes()
+                    });
+                }
+                if let Some(update) = event.session_update() {
+                    replay_checkpoint = Some(update.clone());
+                }
+            }
+            Some(Err(error)) => break error,
+            None => panic!("metadata close must surface a provider error"),
+        }
+    };
+    server.await.expect("WebSocket server");
+
+    assert!(observed_turn_state);
+    let replay_checkpoint = replay_checkpoint.expect("metadata close checkpoint");
+    assert_eq!(
+        replay_checkpoint.payload().get("turn_state"),
+        Some(&json!(TURN_STATE))
+    );
+    assert_eq!(
+        replay_checkpoint.payload().get("retry_checkpoint"),
+        Some(&json!(true))
+    );
+    assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
+    assert!(error.replay_is_safe());
+    assert!(matches!(
+        error.pre_delivery_retry(),
+        Some(PreDeliveryRetry::SameAccountTransportRetry { retry_index, .. })
+            if retry_index.get() == 1
+    ));
+    assert_eq!(
+        error.upstream_code().map(|code| code.as_str()),
+        Some("websocket_close_1000")
+    );
+    assert_eq!(
+        error.diagnostic().map(|diagnostic| diagnostic.as_str()),
+        Some(
+            "OpenAI WebSocket closed before a terminal response (close code 1000); last event type: codex.response.metadata"
+        )
+    );
+}
+
 #[tokio::test]
 async fn disabled_account_is_excluded_from_normal_scheduling() {
     let store = Arc::new(MemoryAccountStore::default());
@@ -2945,6 +3058,7 @@ async fn matching_turn_id_should_restore_previous_turn_state() {
         Some("turn-same"),
         Some("turn-same"),
         None,
+        false,
     )
     .await;
 
@@ -2955,30 +3069,58 @@ async fn matching_turn_id_should_restore_previous_turn_state() {
 }
 
 #[tokio::test]
-async fn matching_turn_id_should_prefer_the_latest_provider_state_over_a_stale_client_echo() {
+async fn retry_checkpoint_should_restore_turn_state_without_client_turn_id() {
+    let request =
+        capture_turn_state_request("req_restore_retry_checkpoint", None, None, None, true).await;
+
+    assert_eq!(
+        captured_header_values(&request, "x-codex-turn-state"),
+        vec![b"previous-turn-state".to_vec()]
+    );
+}
+
+#[tokio::test]
+async fn matching_turn_id_should_prefer_an_explicit_client_echo_over_saved_provider_state() {
     let request = capture_turn_state_request(
-        "req_replace_stale_client_turn_state",
+        "req_use_client_turn_state",
         Some("turn-same"),
         Some("turn-same"),
-        Some("stale-client-turn-state"),
+        Some("client-turn-state"),
+        false,
     )
     .await;
 
     assert_eq!(
         captured_header_values(&request, "x-codex-turn-state"),
-        vec![b"previous-turn-state".to_vec()]
+        vec![b"client-turn-state".to_vec()]
     );
 }
 
 #[tokio::test]
 async fn new_or_unidentified_turn_should_not_restore_previous_turn_state() {
-    for (request_id, previous_turn_id, current_turn_id) in [
-        ("req_new_turn_state", Some("turn-old"), Some("turn-new")),
-        ("req_unidentified_turn_state", None, Some("turn-new")),
-        ("req_missing_current_turn_state", Some("turn-old"), None),
+    for (request_id, previous_turn_id, current_turn_id, client_turn_state) in [
+        (
+            "req_new_turn_state",
+            Some("turn-old"),
+            Some("turn-new"),
+            Some("stale-client-turn-state"),
+        ),
+        ("req_unidentified_turn_state", None, Some("turn-new"), None),
+        (
+            "req_missing_current_turn_state",
+            Some("turn-old"),
+            None,
+            None,
+        ),
     ] {
-        let request =
-            capture_turn_state_request(request_id, previous_turn_id, current_turn_id, None).await;
+        let request = capture_turn_state_request(
+            request_id,
+            previous_turn_id,
+            current_turn_id,
+            client_turn_state,
+            false,
+        )
+        .await;
         assert!(captured_header_values(&request, "x-codex-turn-state").is_empty());
     }
 }

backend/crates/providers/openai/tests/transport/protocol/websocket.rs

@@ -155,6 +155,7 @@ fn websocket_event_to_sse_should_forward_public_events_and_strip_internal_events
     );
     assert!(websocket_event_to_sse_frame(r#"{"type":"codex.rate_limits"}"#).is_none());
     assert!(websocket_event_to_sse_frame(r#"{"type":"response.metadata"}"#).is_none());
+    assert!(websocket_event_to_sse_frame(r#"{"type":"codex.response.metadata"}"#).is_some());
     assert!(websocket_event_to_sse_frame(r#"{"response":{}}"#).is_none());
     assert!(websocket_event_to_sse_frame("not-json").is_none());
 }
@@ -178,15 +179,48 @@ fn websocket_event_to_sse_should_add_missing_previous_response_recovery_code() {
 
 #[test]
 fn websocket_metadata_turn_state_should_accept_case_insensitive_header() {
-    let event = json!({
-        "type": "response.metadata",
-        "headers": {"X-Codex-Turn-State": ["turn-from-metadata"]}
-    });
+    for (event, expected) in [
+        (
+            json!({
+                "type": "response.metadata",
+                "headers": {"X-Codex-Turn-State": ["turn-from-legacy-metadata"]}
+            }),
+            "turn-from-legacy-metadata",
+        ),
+        (
+            json!({
+                "type": "codex.response.metadata",
+                "headers": {"X-Codex-Turn-State": "turn-from-codex-metadata"}
+            }),
+            "turn-from-codex-metadata",
+        ),
+    ] {
+        assert_eq!(
+            websocket_metadata_turn_state(&event).as_deref(),
+            Some(expected)
+        );
+    }
+}
 
-    assert_eq!(
-        websocket_metadata_turn_state(&event).as_deref(),
-        Some("turn-from-metadata")
-    );
+#[test]
+fn websocket_metadata_turn_state_should_ignore_unverified_shapes() {
+    for event in [
+        json!({
+            "type": "codex.response.metadata",
+            "headers": {"x-codex-turn-state": {"future": "shape"}}
+        }),
+        json!({
+            "type": "codex.response.metadata",
+            "metadata": {"turn_state": "nested-value"},
+            "turn_state": "top-level-value"
+        }),
+        json!({
+            "type": "future.response.metadata",
+            "headers": {"x-codex-turn-state": "unverified-event"}
+        }),
+    ] {
+        assert_eq!(websocket_metadata_turn_state(&event), None);
+    }
 }
 
 #[test]

backend/crates/providers/openai/tests/transport/request.rs

@@ -227,6 +227,26 @@ fn encoder_should_ignore_legacy_context_aliases_and_metadata_fallbacks() {
     );
 }
 
+#[test]
+fn encoder_should_extract_official_websocket_context_projection() {
+    let request = request(Map::from_iter([
+        ("model".to_owned(), json!("client-model")),
+        ("input".to_owned(), json!("prompt")),
+        (
+            "client_metadata".to_owned(),
+            json!({
+                "x-codex-turn-state": "metadata-turn-state",
+                "turn_id": "metadata-turn-id"
+            }),
+        ),
+    ]));
+
+    let encoded = encode_generate_request(&request, "gpt-test").expect("encode");
+
+    assert_eq!(encoded.turn_state.as_deref(), Some("metadata-turn-state"));
+    assert_eq!(encoded.client_turn_id.as_deref(), Some("metadata-turn-id"));
+}
+
 #[test]
 fn header_context_should_win_over_body_topline_aliases() {
     let payload = ProtocolPayload::json_object(
@@ -236,6 +256,13 @@ fn header_context_should_win_over_body_topline_aliases() {
             ("input".to_owned(), json!("prompt")),
             ("turnState".to_owned(), json!("body-turn-state")),
             ("turnMetadata".to_owned(), json!("body-turn-metadata")),
+            (
+                "client_metadata".to_owned(),
+                json!({
+                    "x-codex-turn-state": "metadata-turn-state",
+                    "turn_id": "metadata-turn-id"
+                }),
+            ),
         ]),
     )
     .expect("OpenAI payload")
@@ -248,6 +275,10 @@ fn header_context_should_win_over_body_topline_aliases() {
             "turn_metadata".to_owned(),
             Value::String("header-turn-metadata".to_owned()),
         ),
+        (
+            "turn_id".to_owned(),
+            Value::String("header-turn-id".to_owned()),
+        ),
     ]));
     let request = GenerateRequest::from_protocol_payload(payload);
 
@@ -258,6 +289,36 @@ fn header_context_should_win_over_body_topline_aliases() {
         encoded.turn_metadata.as_deref(),
         Some("header-turn-metadata")
     );
+    assert_eq!(encoded.client_turn_id.as_deref(), Some("header-turn-id"));
+}
+
+#[test]
+fn encoder_should_preserve_downstream_websocket_connection_identity_outside_wire_body() {
+    let payload = ProtocolPayload::json_object(
+        "openai",
+        Map::from_iter([
+            ("model".to_owned(), json!("client-model")),
+            ("input".to_owned(), json!("prompt")),
+        ]),
+    )
+    .expect("OpenAI payload")
+    .with_context(Map::from_iter([(
+        "downstream_websocket_connection_id".to_owned(),
+        Value::String("ws_downstream_a".to_owned()),
+    )]));
+    let request = GenerateRequest::from_protocol_payload(payload);
+
+    let encoded = encode_generate_request(&request, "gpt-test").expect("encode");
+
+    assert_eq!(
+        encoded.downstream_websocket_connection_id.as_deref(),
+        Some("ws_downstream_a")
+    );
+    assert!(
+        !encoded
+            .body()
+            .contains_key("downstream_websocket_connection_id")
+    );
 }
 
 #[test]

backend/crates/providers/openai/tests/transport/websocket/mod.rs

@@ -914,6 +914,191 @@ async fn websocket_execute_response_create_request_should_capture_internal_metad
     );
 }
 
+#[tokio::test]
+async fn websocket_codex_metadata_should_capture_first_turn_state_without_changing_forwarding() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        let _message = websocket.next().await.unwrap().unwrap();
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "codex.response.metadata",
+                    "headers": {"x-codex-turn-state": "turn-first"}
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .unwrap();
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "response.metadata",
+                    "headers": {"x-codex-turn-state": ["turn-second"]}
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .unwrap();
+        websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_metadata_alias", 1, 1).into(),
+            ))
+            .await
+            .unwrap();
+    });
+    let prepared = prepared_websocket_request(&format!("http://{addr}"));
+
+    let response = execute_response_create_request(&prepared)
+        .await
+        .expect("metadata aliases should preserve a completed response");
+    server.await.unwrap();
+
+    assert_eq!(response.turn_state.as_deref(), Some("turn-first"));
+    assert!(response.body.contains("event: codex.response.metadata"));
+    assert!(!response.body.contains("event: response.metadata\n"));
+    assert!(response.body.contains("event: response.completed"));
+}
+
+#[tokio::test]
+async fn reused_websocket_should_not_leak_turn_state_into_the_next_response() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let server = tokio::spawn(async move {
+        let (stream, _) = listener.accept().await.unwrap();
+        let mut websocket = accept_codex_test_websocket(stream).await;
+        let _first = websocket.next().await.unwrap().unwrap();
+        websocket
+            .send(Message::Text(
+                json!({
+                    "type": "codex.response.metadata",
+                    "headers": {"x-codex-turn-state": "turn-from-first-response"}
+                })
+                .to_string()
+                .into(),
+            ))
+            .await
+            .unwrap();
+        websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_turn_state_first", 1, 1).into(),
+            ))
+            .await
+            .unwrap();
+
+        let _second = websocket.next().await.unwrap().unwrap();
+        websocket
+            .send(Message::Text(
+                completed_websocket_response("resp_turn_state_second", 1, 1).into(),
+            ))
+            .await
+            .unwrap();
+    });
+    let pool = Arc::new(CodexWebSocketPool::new(Duration::from_mins(1)));
+    let backend = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        format!("http://{addr}"),
+        test_wire_profile(),
+    )
+    .with_websocket_pool(pool);
+    let request = pooled_websocket_request("conversation-turn-state-reset");
+
+    let first = backend
+        .create_response(
+            &request,
+            request_context("req_turn_state_first", Some("chatgpt-account")),
+        )
+        .await
+        .expect("first response should complete");
+    let second = backend
+        .create_response(
+            &request,
+            request_context("req_turn_state_second", Some("chatgpt-account")),
+        )
+        .await
+        .expect("second response should reuse the WebSocket");
+    server.await.unwrap();
+
+    assert_eq!(
+        first.turn_state.as_deref(),
+        Some("turn-from-first-response")
+    );
+    assert!(
+        second
+            .websocket_pool_decision
+            .is_some_and(WebSocketPoolDecision::is_reuse)
+    );
+    assert_eq!(second.turn_state, None);
+    assert!(
+        second
+            .response_metadata
+            .client_headers
+            .iter()
+            .all(|(name, _)| { !name.eq_ignore_ascii_case("x-codex-turn-state") })
+    );
+}
+
+#[tokio::test]
+async fn websocket_request_should_project_trusted_turn_state_and_remove_stale_metadata() {
+    for (case, turn_state) in [("trusted", Some("trusted-turn-state")), ("cleared", None)] {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let server = tokio::spawn(async move {
+            let (stream, _) = listener.accept().await.unwrap();
+            let mut websocket = accept_codex_test_websocket(stream).await;
+            let message = websocket.next().await.unwrap().unwrap();
+            let payload = serde_json::from_str::<Value>(&message.into_text().unwrap())
+                .expect("response.create payload");
+            websocket
+                .send(Message::Text(
+                    completed_websocket_response("resp_turn_state_projection", 1, 1).into(),
+                ))
+                .await
+                .unwrap();
+            payload
+        });
+        let mut body = codex_request_body("gpt-5.5", "be brief", Vec::new());
+        body.insert("generate".to_owned(), json!(false));
+        body.insert("store".to_owned(), json!(false));
+        let mut request = CodexResponsesRequest::from_body(body);
+        request.turn_state = turn_state.map(str::to_owned);
+        request.local_conversation_id = Some(format!("turn-state-projection-{case}"));
+        request.set_client_metadata(Some(json!({
+            "x-codex-turn-state": "stale-client-value",
+            "keep": "preserved"
+        })));
+        let backend = CodexBackendClient::new(
+            reqwest::Client::builder().no_proxy().build().unwrap(),
+            format!("http://{addr}"),
+            test_wire_profile(),
+        )
+        .with_websocket_pool(Arc::new(CodexWebSocketPool::new(Duration::from_mins(1))));
+        let mut context = request_context(case, Some("chatgpt-account"));
+        context.turn_state = turn_state;
+
+        let response = backend
+            .create_response(&request, context)
+            .await
+            .unwrap_or_else(|error| panic!("required WebSocket request should complete: {error}"));
+        let payload = server.await.unwrap();
+
+        assert_eq!(response.transport, CodexBackendTransport::WebSocket);
+        assert_eq!(
+            payload.pointer("/client_metadata/keep"),
+            Some(&json!("preserved"))
+        );
+        assert_eq!(
+            payload.pointer("/client_metadata/x-codex-turn-state"),
+            turn_state.map(|turn_state| json!(turn_state)).as_ref(),
+            "case {case}"
+        );
+    }
+}
+
 #[tokio::test]
 async fn codex_backend_client_should_not_reuse_websocket_rate_limit_headers() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

backend/crates/providers/openai/tests/transport/websocket_pool.rs

@@ -71,6 +71,138 @@ async fn codex_backend_client_should_reuse_pooled_websocket_for_same_account_and
     assert_eq!(accepted_connections.load(Ordering::SeqCst), 1);
 }
 
+#[tokio::test]
+async fn websocket_pool_should_isolate_concurrent_downstream_lanes_for_one_conversation() {
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let accepted_connections = Arc::new(AtomicUsize::new(0));
+    let accepted_connections_for_server = Arc::clone(&accepted_connections);
+    let (first_accepted_tx, first_accepted_rx) = tokio::sync::oneshot::channel();
+    let server = tokio::spawn(async move {
+        let (first_stream, _) = listener.accept().await.unwrap();
+        accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
+        let mut first = accept_codex_test_websocket(first_stream).await;
+        first_accepted_tx.send(()).unwrap();
+
+        let (second_stream, _) = timeout(Duration::from_secs(2), listener.accept())
+            .await
+            .expect("the second downstream lane should open its own upstream websocket")
+            .unwrap();
+        accepted_connections_for_server.fetch_add(1, Ordering::SeqCst);
+        let mut second = accept_codex_test_websocket(second_stream).await;
+
+        let (first_warmup, second_warmup) = tokio::join!(first.next(), second.next());
+        first_warmup.unwrap().unwrap();
+        second_warmup.unwrap().unwrap();
+        first
+            .send(Message::Text(
+                completed_websocket_response("resp_lane_a", 1, 0).into(),
+            ))
+            .await
+            .unwrap();
+        second
+            .send(Message::Text(
+                completed_websocket_response("resp_lane_b", 1, 0).into(),
+            ))
+            .await
+            .unwrap();
+
+        let _first_continuation = first.next().await.unwrap().unwrap();
+        first
+            .send(Message::Text(
+                completed_websocket_response("resp_lane_a_next", 2, 1).into(),
+            ))
+            .await
+            .unwrap();
+        let _second_continuation = second.next().await.unwrap().unwrap();
+        second
+            .send(Message::Text(
+                completed_websocket_response("resp_lane_b_next", 2, 1).into(),
+            ))
+            .await
+            .unwrap();
+        first.close(None).await.unwrap();
+        second.close(None).await.unwrap();
+    });
+    let pool = Arc::new(CodexWebSocketPool::new(Duration::from_mins(1)));
+    let backend = CodexBackendClient::new(
+        reqwest::Client::builder().no_proxy().build().unwrap(),
+        format!("http://{addr}"),
+        test_wire_profile(),
+    )
+    .with_websocket_pool(Arc::clone(&pool));
+    let warmup = |connection_id: &str| {
+        let mut body = Map::new();
+        body.insert("model".to_owned(), json!("gpt-5.5"));
+        body.insert("input".to_owned(), json!([]));
+        body.insert("generate".to_owned(), json!(false));
+        body.insert("store".to_owned(), json!(false));
+        let mut request = CodexResponsesRequest::from_body(body);
+        request.use_websocket = true;
+        request.local_conversation_id = Some("conversation-lanes".to_owned());
+        request.downstream_websocket_connection_id = Some(connection_id.to_owned());
+        request
+    };
+
+    let first_backend = backend.clone();
+    let first_request = warmup("ws_downstream_a");
+    let first = tokio::spawn(async move {
+        first_backend
+            .create_response(
+                &first_request,
+                request_context("req_lane_a_warmup", Some("chatgpt-account")),
+            )
+            .await
+    });
+    first_accepted_rx.await.unwrap();
+    let second = backend
+        .create_response(
+            &warmup("ws_downstream_b"),
+            request_context("req_lane_b_warmup", Some("chatgpt-account")),
+        )
+        .await
+        .expect("the second downstream lane should not contend with the first warmup");
+    let first = first.await.unwrap().unwrap();
+
+    let mut first_continuation = pooled_websocket_request("conversation-lanes");
+    first_continuation.downstream_websocket_connection_id = Some("ws_downstream_a".to_owned());
+    first_continuation.set_previous_response_id(Some("resp_lane_a".to_owned()));
+    first_continuation.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
+    let first_continuation = backend
+        .create_response(
+            &first_continuation,
+            request_context("req_lane_a_next", Some("chatgpt-account")),
+        )
+        .await
+        .expect("the first lane should retain its connection-local response owner");
+
+    let mut second_continuation = pooled_websocket_request("conversation-lanes");
+    second_continuation.downstream_websocket_connection_id = Some("ws_downstream_b".to_owned());
+    second_continuation.set_previous_response_id(Some("resp_lane_b".to_owned()));
+    second_continuation.previous_response_scope = Some(PreviousResponseScope::ConnectionLocal);
+    let second_continuation = backend
+        .create_response(
+            &second_continuation,
+            request_context("req_lane_b_next", Some("chatgpt-account")),
+        )
+        .await
+        .expect("the second lane should retain its connection-local response owner");
+    server.await.unwrap();
+
+    assert_eq!(first.websocket_pool_decision.unwrap().kind(), "new");
+    assert_eq!(second.websocket_pool_decision.unwrap().kind(), "new");
+    assert_eq!(
+        first_continuation.websocket_pool_decision.unwrap().kind(),
+        "reuse"
+    );
+    assert_eq!(
+        second_continuation.websocket_pool_decision.unwrap().kind(),
+        "reuse"
+    );
+    assert_eq!(accepted_connections.load(Ordering::SeqCst), 2);
+    pool.shutdown().await;
+}
+
 #[tokio::test]
 async fn exact_continuation_reuses_owning_socket_after_connection_profile_updates() {
     let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();

backend/migrations/.frozen-sha256

@@ -3,3 +3,4 @@
 1b6ffab434c754814aebb0eee3b650bd9770010059472bccf5bc856b319880ca  0003_client_min_versions.sql
 22f912ab3e196de7983ef50a1dba5aee340bbea82bdd15d05863e87012000c83  0004_ops_error_send_state.sql
 194d7b4bedd83f0d76428a21699b0e946449bf2ba61a0fdf8710ebfd421af8b2  0005_openai_continuation_recovery_observability.sql
+3b4bfd5e3f5135fa2fad1088b0c2a287861e1250d66af31dbd40dc890751b567  0006_openai_session_transport_recovery.sql

backend/migrations/0006_openai_session_transport_recovery.sql

@@ -0,0 +1,61 @@
+create index model_requests_pending_ws_transport_recovery_idx
+  on model_requests (
+    client_api_key_ref,
+    continuation_affinity_hash,
+    completed_at desc,
+    id desc
+  )
+  where provider_kind = 'openai'
+    and outcome = 'failed'
+    and error_kind = 'upstream_unavailable'
+    and upstream_transport = 'websocket'
+    and upstream_send_state = 'ambiguous'
+    and downstream_committed_at is null
+    and recovered_at is null
+    and continuation_affinity_hash is not null;
+
+with session_transport_recoveries as (
+  select
+    prior.id as failed_request_id,
+    recovered.id as recovery_request_id,
+    recovered.started_at as recovery_started_at,
+    recovered.completed_at as recovered_at
+  from model_requests prior
+  cross join lateral (
+    select current.id, current.started_at, current.completed_at
+    from model_requests current
+    where current.client_api_key_ref = prior.client_api_key_ref
+      and current.continuation_affinity_hash = prior.continuation_affinity_hash
+      and current.provider_kind = prior.provider_kind
+      and current.outcome = 'succeeded'
+      and current.upstream_transport = 'http_sse'
+      and current.completed_at is not null
+      and current.started_at >= prior.completed_at
+      and current.started_at <= prior.completed_at + interval '30 seconds'
+    order by current.started_at, current.id
+    limit 1
+  ) recovered
+  where prior.provider_kind = 'openai'
+    and prior.outcome = 'failed'
+    and prior.error_kind = 'upstream_unavailable'
+    and prior.upstream_transport = 'websocket'
+    and prior.upstream_send_state = 'ambiguous'
+    and prior.downstream_committed_at is null
+    and prior.recovered_at is null
+    and prior.continuation_affinity_hash is not null
+)
+update model_requests prior
+set recovery_attempt_count = prior.recovery_attempt_count + 1,
+    recovery_request_id = recovery.recovery_request_id,
+    recovered_at = recovery.recovered_at,
+    recovery_retry_delay_ms = greatest(
+      0,
+      floor(extract(epoch from (recovery.recovery_started_at - prior.completed_at)) * 1000)
+    )::bigint,
+    recovery_total_latency_ms = greatest(
+      0,
+      floor(extract(epoch from (recovery.recovered_at - prior.completed_at)) * 1000)
+    )::bigint
+from session_transport_recoveries recovery
+where prior.id = recovery.failed_request_id
+  and prior.recovered_at is null;

deploy/config.example.yaml

@@ -20,6 +20,8 @@ host:
   logging:
     level: 'info'
     stdout: true
+    # 将完整请求头和正文写入独立文件；包含凭据，仅排障时开启。
+    request_dump: false
     file:
       enabled: true
       # 相对路径以 deploy/config.yaml 所在目录为基准。

frontend/src/api/modules/accounts.ts

@@ -83,6 +83,8 @@ export interface AccountUsage {
   outputTokensDisplay: string
   cachedTokens: number | null
   cachedTokensDisplay: string
+  reasoningTokens: number | null
+  reasoningTokensDisplay: string
   imageInputTokens: number | null
   imageInputTokensDisplay: string
   imageOutputTokens: number | null

frontend/src/components/base/BaseSwitch.vue

@@ -45,7 +45,8 @@ const thumbClasses = computed(() => [
   model.value ? 'left-[calc(100%-1.375rem)]' : 'left-0.5',
 ])
 const inlineTextClasses = computed(() => [
-  'pointer-events-none absolute inset-0 flex min-w-0 select-none items-center justify-center overflow-hidden text-[10px] leading-none font-heavy text-cp-white transition-[padding] duration-180 motion-reduce:transition-none',
+  'pointer-events-none absolute inset-0 flex min-w-0 select-none items-center justify-center overflow-hidden text-[10px] leading-none font-heavy transition-[padding,color] duration-180 motion-reduce:transition-none',
+  model.value ? 'text-cp-white' : 'text-cp-text-secondary',
   model.value ? 'pr-6 pl-1' : 'pr-1 pl-6',
 ])
 const inactiveTextClasses = computed(() => [

frontend/src/views/accounts/components/AccountUsagePanel.vue

@@ -35,33 +35,33 @@ const modelUsageColumns = defineTableColumns<AccountModelUsage>([
     <div class="xl:flex xl:min-h-0 xl:flex-col">
       <div class="mb-3 flex shrink-0 items-baseline justify-between gap-3">
         <h3 class="m-0 text-cp-lg font-heavy text-cp-text">
-          Token 结构
+          Tokens 结构
         </h3>
         <span class="text-cp-xs font-emphasis text-cp-text-quaternary">当前额度窗口</span>
       </div>
       <div class="grid gap-2 xl:min-h-0 xl:flex-1 xl:grid-rows-5">
         <div class="flex items-center justify-between rounded-lg bg-cp-green-bg px-3 py-2">
-          <span class="text-cp-sm font-bold text-cp-green-text-on-bg">输入 Tokens</span>
+          <span class="text-cp-sm font-bold text-cp-green-text-on-bg">输入</span>
           <strong class="font-mono text-cp text-cp-text">
             {{ account.usage.inputTokensDisplay }}
           </strong>
         </div>
         <div class="flex items-center justify-between rounded-lg bg-cp-orange-bg px-3 py-2">
-          <span class="text-cp-sm font-bold text-cp-orange-text-on-bg">输出 Tokens</span>
+          <span class="text-cp-sm font-bold text-cp-orange-text-on-bg">输出</span>
           <strong class="font-mono text-cp text-cp-text">
             {{ account.usage.outputTokensDisplay }}
           </strong>
         </div>
         <div class="flex items-center justify-between rounded-lg bg-cp-cyan-bg px-3 py-2">
-          <span class="text-cp-sm font-bold text-cp-cyan-text-on-bg">缓存 Tokens</span>
+          <span class="text-cp-sm font-bold text-cp-cyan-text-on-bg">缓存</span>
           <strong class="font-mono text-cp text-cp-text">
             {{ account.usage.cachedTokensDisplay }}
           </strong>
         </div>
         <div class="flex items-center justify-between rounded-lg bg-cp-blue-bg px-3 py-2">
-          <span class="text-cp-sm font-bold text-cp-blue-text-on-bg">创建</span>
+          <span class="text-cp-sm font-bold text-cp-blue-text-on-bg">推理</span>
           <strong class="font-mono text-cp text-cp-text">
-            {{ account.usage.createdTokensDisplay }}
+            {{ account.usage.reasoningTokensDisplay }}
           </strong>
         </div>
         <div class="flex items-center justify-between rounded-lg bg-cp-blue-bg px-3 py-2">

frontend/src/views/api-keys/utils/codexConfig.ts

@@ -1,5 +1,5 @@
 export const CODEX_DEFAULT_MODEL = 'gpt-5.6-terra'
-export const CODEX_WEBSOCKET_ENABLED_BY_DEFAULT = true
+export const CODEX_WEBSOCKET_ENABLED_BY_DEFAULT = false
 
 export interface CodexConfigInput {
   apiKey: string

frontend/src/views/usage/components/OpsErrorDetailModal.vue

@@ -31,6 +31,10 @@ const accountLabel = computed(() => props.record?.accountEmail
   || props.record?.accountId)
 
 const errorFields = computed(() => visibleFields([
+  {
+    label: '处理状态',
+    value: props.record?.metadata.recoveredAt ? '已自动恢复' : null,
+  },
   { label: 'Provider 错误码', value: props.record?.providerErrorCode, mono: true },
   { label: '失败分类', value: failureClassText(props.record?.failureClass) },
   { label: '分类代码', value: props.record?.failureClass, mono: true },

frontend/src/views/usage/components/OpsErrorPanel.vue

@@ -124,9 +124,17 @@ function upstreamSendStateText(value: string | null | undefined) {
         </template>
         <template #message="{ row }">
           <div class="min-w-0 py-0.5" :title="row.message || opsErrorSummary(row)">
-            <code class="block max-w-full truncate font-mono text-cp-sm font-bold text-cp-error-text">
-              {{ opsErrorSummary(row) }}
-            </code>
+            <div class="flex min-w-0 items-center gap-2">
+              <code class="block min-w-0 flex-1 truncate font-mono text-cp-sm font-bold text-cp-error-text">
+                {{ opsErrorSummary(row) }}
+              </code>
+              <span
+                v-if="row.metadata.recoveredAt"
+                class="inline-flex h-5 shrink-0 items-center rounded-full bg-cp-success-bg px-2 text-cp-xs leading-none font-heavy text-cp-success-text"
+              >
+                已自动恢复
+              </span>
+            </div>
             <p
               v-if="row.message"
               class="mt-1 mb-0 line-clamp-1 text-cp-xs leading-[1.45] font-emphasis text-cp-text-secondary"

release/version.yaml

@@ -1 +1 @@
-version: "3.3.20"
+version: "3.3.23"