chore: bump version to 3.3.22

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;
@@ -152,7 +152,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 +179,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 +247,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 +275,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 +308,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

@@ -143,6 +143,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();

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-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

@@ -732,14 +732,23 @@ 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() {
+                    let rate_limits_changed = if updates.is_empty() {
+                        false
+                    } else {
                         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);
-                        }
+                        observation_state.merge_rate_limit_headers(&update_headers)
+                    };
+                    let turn_state_changed = merge_turn_state_update(
+                        turn_state_updates.as_ref(),
+                        &mut session_capture,
+                        &mut observation_state,
+                    )
+                    .await;
+                    if (rate_limits_changed || turn_state_changed)
+                        && let Some(observation) = observation_state.observation()
+                    {
+                        yield ProviderEvent::observation(observation);
                     }
                     if allows_account_state_mutation {
                         synchronize_passive_quota(
@@ -763,16 +772,12 @@ 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_changed = merge_turn_state_update(
+                turn_state_updates.as_ref(),
+                &mut session_capture,
+                &mut observation_state,
+            )
+            .await;
             let first_event_changed =
                 observation_state.observe_stream_chunk(&chunk, output_started_at);
             let chunk_len = chunk.len();
@@ -949,16 +954,12 @@ 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;
         attach_openai_session_update(&mut events, &mut session_capture);
         let completed = events
             .iter()
@@ -1024,3 +1025,20 @@ 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,
+) -> bool {
+    let Some(updates) = updates else {
+        return false;
+    };
+    let Some(turn_state) = updates.lock().await.take() else {
+        return false;
+    };
+    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)
+}

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,20 @@ 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(),
+        if let Some(previous_session) = previous_session.as_ref() {
+            upstream_request.turn_state = if same_client_turn(
+                previous_session.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());
+            ) {
+                // 客户端可能携带 metadata-close 失败后才拿到的新状态；同一
+                // turn 内显式回传值比最后一次成功响应保存的状态更新。
+                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

@@ -682,10 +682,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)
     }
 
     /// 获取后端模型目录条目。

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()
@@ -2519,6 +2520,96 @@ async fn websocket_turn_state_metadata_is_exposed_through_response_observation()
     assert!(observed_turn_state);
 }
 
+#[tokio::test]
+async fn websocket_turn_state_metadata_survives_close_1000_without_changing_failure_semantics() {
+    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 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()
+                    });
+                }
+            }
+            Some(Err(error)) => break error,
+            None => panic!("metadata close must surface a provider error"),
+        }
+    };
+    server.await.expect("WebSocket server");
+
+    assert!(observed_turn_state);
+    assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
+    assert_eq!(error.pre_delivery_retry(), None);
+    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());
@@ -2955,30 +3046,45 @@ 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 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"),
     )
     .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,
+        )
+        .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;

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.22"