chore(release): v3.26.1

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

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

@@ -560,8 +560,6 @@ where
             match self.pull().await? {
                 PullOutcome::Events(events) => {
                     self.replay_barrier_reached |= self.middleware.is_some()
-                        || self.client_transport != super::execution::ClientTransport::WebSocket
-                        || self.response_control.is_none()
                         || events.iter().any(|event| !event.is_replayable_prelude());
                     let requirement = if self.delivery_committed {
                         CommitRequirement::AlreadyCommitted

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

@@ -559,12 +559,10 @@ impl AttemptContext {
         self.request.client_transport
     }
 
-    /// 原生双向通道可提前交付可替换前导事件；中间件可能改变其输出语义
+    /// 原生响应可交付可替换前导事件；中间件可能改变其输出语义
     #[must_use]
     pub fn allows_replayable_prelude(&self) -> bool {
-        self.client_transport() == execution::ClientTransport::WebSocket
-            && self.request.middleware.is_none()
-            && self.response_control().is_some()
+        self.request.middleware.is_none()
     }
 
     #[must_use]

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

@@ -4286,6 +4286,85 @@ fn structural_event_before_replay_safe_failure_should_switch_account_before_comm
     assert_eq!(state.finalizations[0].client_status_code, Some(200));
 }
 
+#[test]
+fn delivered_prelude_allows_retry_without_control_until_non_replayable_delivery() {
+    for close_window in [false, true] {
+        let operation = generate_operation();
+        let route_plan = plan(&operation);
+        let prelude = || {
+            ProviderEvent::wire(
+                ProtocolWireEvent::json(
+                    "openai",
+                    Some("response.created".to_owned()),
+                    json!({"type":"response.created","response":{"id":"response-1","output":[]}}),
+                )
+                .unwrap(),
+            )
+            .with_replayable_prelude()
+        };
+        let mut items = vec![Ok(prelude())];
+        if close_window {
+            items.push(Ok(ProviderEvent::wire(
+                ProtocolWireEvent::json(
+                    "openai",
+                    Some("response.future".to_owned()),
+                    json!({"type":"response.future"}),
+                )
+                .unwrap(),
+            )));
+            // 后续前导不能撤销已经交付未知事件的事实
+            items.push(Ok(prelude()));
+        }
+        items.push(Err(ProviderError::new(
+            ProviderErrorKind::Transport,
+            UpstreamSendState::Sent,
+        )
+        .with_replay_safe()));
+        let (coordinator, store, provider) = coordinator(vec![
+            Script::ObservedStream {
+                account_id: "acct_first",
+                items,
+            },
+            Script::Stream {
+                account_id: "acct_second",
+                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(),
+        ))
+        .unwrap();
+        let first = block_on(session.next_event()).unwrap().unwrap();
+        assert!(first.into_provider_events()[0].is_replayable_prelude());
+        block_on(session.commit_downstream(Some(200))).unwrap();
+        let result = loop {
+            match block_on(session.next_event()) {
+                Ok(Some(_)) => {}
+                terminal => break terminal,
+            }
+        };
+        assert_eq!(result.is_err(), close_window);
+        let attempts = if close_window { 1 } else { 2 };
+        assert_eq!(provider.contexts.lock().unwrap().len(), attempts);
+        let state = store.state.lock().unwrap();
+        assert_eq!(state.commits, 1);
+        assert_eq!(state.intermediate_failures, attempts - 1);
+        assert_eq!(
+            state.finalizations[0].outcome,
+            if close_window {
+                ExecutionOutcome::Incomplete
+            } else {
+                ExecutionOutcome::Succeeded
+            }
+        );
+    }
+}
+
 #[test]
 fn cancellation_before_pending_delivery_commit_reaches_terminal_state() {
     let operation = generate_operation();

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

@@ -954,11 +954,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             .with_requested_service_tier(request.service_tier())
             .with_request_tool_pricing(upstream_model.as_str(), request.tools())
             .with_raw_sse_passthrough();
-        let mut pre_commit_events = PreCommitClientEvents::new(trace).with_replayable_prelude(
-            websocket_transport
-                && request.responses_lite.as_deref() == Some("true")
-                && context.allows_replayable_prelude(),
-        );
+        let mut client_events = ClientEventDelivery::new(context.allows_replayable_prelude());
         loop {
             if context.deadline().is_elapsed() {
                 if allows_account_state_mutation {
@@ -1016,8 +1012,10 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                         &mut decoder,
                     )
                     .await;
-                    let observation_event = if rate_limits_changed || metadata_merge.is_some() {
-                        observation_state.observation(None).map(ProviderEvent::observation)
+                    let observation_event = if rate_limits_changed || metadata_merge.is_some()
+                        || response_transport == CodexBackendTransport::WebSocket
+                    {
+                        observation_state.observation(Some(&failure.error)).map(ProviderEvent::observation)
                     } else {
                         None
                     };
@@ -1028,7 +1026,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                             &mut failure,
                             WebSocketRecoveryContext {
                                 policy: WebSocketFailurePolicy::Budgeted,
-                                replay_boundary: ReplayBoundary::from_semantic_output(pre_commit_events.is_committed()),
+                                replay_boundary: ReplayBoundary::from_semantic_output(client_events.replay_closed()),
                                 requirement: request_transport_requirement,
                                 retry_count: websocket_retry_count,
                                 max_retries: stream_max_retries,
@@ -1057,7 +1055,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                     Err(quota_continuation_replay_error(
                         failure.error,
                         &request,
-                        ReplayBoundary::from_semantic_output(pre_commit_events.is_committed()),
+                        ReplayBoundary::from_semantic_output(client_events.replay_closed()),
                     ))?;
                     return;
                 }
@@ -1072,7 +1070,6 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             };
             let first_event_changed =
                 observation_state.observe_stream_chunk(&chunk, context.timing_started_at());
-            let chunk_len = chunk.len();
             let decoded = if websocket_transport {
                 let text = std::str::from_utf8(&chunk).map_err(|_| provider_error(ProviderErrorKind::Protocol, UpstreamSendState::Sent))?;
                 decoder.push_websocket(text)
@@ -1094,7 +1091,6 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             )
             .await;
             let metadata_changed = metadata_merge.unwrap_or(false);
-            pre_commit_events.observe_chunk(chunk_len);
             let response_model_changed = observation_state
                 .observe_upstream_response_model(decoder.response_model());
             let service_tier_changed = observation_state
@@ -1117,7 +1113,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                         &failure_set_cookie_headers,
                         &failure_rate_limit_headers,
                         ReplayBoundary::from_semantic_output(
-                            semantic_output_seen || pre_commit_events.is_committed(),
+                            semantic_output_seen || client_events.replay_closed(),
                         ),
                     ),
                     atomic_upstream_failure,
@@ -1183,15 +1179,15 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             }
             if let Some((mut failure, atomic_upstream_failure)) = terminal_failure {
                 let failure_after_commit =
-                    timing_signals.semantic_output || pre_commit_events.is_committed();
+                    timing_signals.semantic_output || client_events.replay_closed();
                 if failure_after_commit {
-                    for event in pre_commit_events.commit(events) {
+                    for event in client_events.commit(events) {
                         yield event;
                     }
                 } else if atomic_upstream_failure {
                     failure.error = failure
                         .error
-                        .with_atomic_client_events(pre_commit_events.take_for_failure(events));
+                        .with_atomic_client_events(events);
                 }
                 Err(quota_continuation_replay_error(
                     failure.error,
@@ -1200,7 +1196,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                 ))?;
                 return;
             }
-            let events = pre_commit_events.stage(events, timing_signals, completed);
+            let events = client_events.forward(events, timing_signals, completed);
             for event in events {
                 yield event;
             }
@@ -1233,7 +1229,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                     &failure_set_cookie_headers,
                     &failure_rate_limit_headers,
                     ReplayBoundary::from_semantic_output(
-                        semantic_output_seen || pre_commit_events.is_committed(),
+                        semantic_output_seen || client_events.replay_closed(),
                     ),
                 ),
                 atomic_upstream_failure,
@@ -1317,15 +1313,15 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
         }
         if let Some((mut failure, atomic_upstream_failure)) = terminal_failure {
             let failure_after_commit =
-                timing_signals.semantic_output || pre_commit_events.is_committed();
+                timing_signals.semantic_output || client_events.replay_closed();
             if failure_after_commit {
-                for event in pre_commit_events.commit(events) {
+                for event in client_events.commit(events) {
                     yield event;
                 }
             } else if atomic_upstream_failure {
                 failure.error = failure
                     .error
-                    .with_atomic_client_events(pre_commit_events.take_for_failure(events));
+                    .with_atomic_client_events(events);
             }
             Err(quota_continuation_replay_error(
                 failure.error,
@@ -1334,7 +1330,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
             ))?;
             return;
         }
-        let events = pre_commit_events.finish(events, timing_signals, completed);
+        let events = client_events.forward(events, timing_signals, completed);
         for event in events {
             yield event;
         }

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

@@ -1,7 +1,7 @@
 //! OpenAI 上游失败分类、恢复决策与稳定错误投影
 
 use super::*;
-use gateway_core::diagnostics::TraceContext;
+use crate::transport::protocol::responses::is_replaceable_response_prelude;
 
 /// OpenAI 失败对 Smart 账号分数的结构化 reason 闭集
 ///
@@ -69,128 +69,50 @@ impl MappedProviderFailure {
     }
 }
 
-/// 提交边界前的上游事件预取
-///
-/// 原生 Lite 双向通道立即交付正常事件，其他通道保留到语义输出、终态或 EOF
-/// committed 只表示重放窗口已结束，不等同于首次 wire 已交付
-pub(super) struct PreCommitClientEvents {
-    pending: Vec<ProviderEvent>,
-    prefetched_bytes: usize,
-    started_at: tokio::time::Instant,
-    committed: bool,
+/// 正常事件立即交付，只记录后续失败能否请求重放
+/// 可替换前导不关闭窗口；实际输出、终态和未知事件使边界单向推进
+pub(super) struct ClientEventDelivery {
+    replay_closed: bool,
     replayable_prelude: bool,
-    trace: TraceContext,
 }
 
-#[derive(Clone, Copy, Serialize)]
-#[serde(rename_all = "snake_case")]
-pub(super) enum PreCommitReleaseReason {
-    SemanticOutput,
-    NonReplayableEvent,
-    Terminal,
-    Eof,
-}
-
-impl PreCommitClientEvents {
-    pub(super) fn new(trace: TraceContext) -> Self {
+impl ClientEventDelivery {
+    pub(super) fn new(replayable_prelude: bool) -> Self {
         Self {
-            pending: Vec::new(),
-            prefetched_bytes: 0,
-            started_at: tokio::time::Instant::now(),
-            committed: false,
-            replayable_prelude: false,
-            trace,
+            replay_closed: false,
+            replayable_prelude,
         }
     }
 
-    pub(super) fn with_replayable_prelude(mut self, enabled: bool) -> Self {
-        self.replayable_prelude = enabled;
-        self
-    }
-
-    pub(super) fn observe_chunk(&mut self, bytes: usize) {
-        if !self.committed {
-            self.prefetched_bytes = self.prefetched_bytes.saturating_add(bytes);
-        }
-    }
-
-    pub(super) fn stage(
+    pub(super) fn forward(
         &mut self,
         incoming: Vec<ProviderEvent>,
         timing_signals: ResponseEventSignals,
         completed: bool,
     ) -> Vec<ProviderEvent> {
-        if self.committed {
+        if self.replay_closed || incoming.is_empty() {
             return incoming;
         }
-        self.pending.extend(incoming);
-        if timing_signals.semantic_output {
-            return self.commit_pending(PreCommitReleaseReason::SemanticOutput);
+        if timing_signals.semantic_output
+            || completed
+            || !self.replayable_prelude
+            || !incoming.iter().all(is_replaceable_prelude)
+        {
+            return self.commit(incoming);
         }
-        if completed {
-            return self.commit_pending(PreCommitReleaseReason::Terminal);
-        }
-        if self.replayable_prelude {
-            // 未确认可重放的事件也立即交付，但后续前导不能重新开启重试窗口
-            if !self.pending.iter().all(is_replaceable_prelude) {
-                return self.commit_pending(PreCommitReleaseReason::NonReplayableEvent);
-            }
-            self.prefetched_bytes = 0;
-            return std::mem::take(&mut self.pending)
-                .into_iter()
-                .map(ProviderEvent::with_replayable_prelude)
-                .collect();
-        }
-        Vec::new()
-    }
-
-    pub(super) fn finish(
-        &mut self,
-        incoming: Vec<ProviderEvent>,
-        timing_signals: ResponseEventSignals,
-        completed: bool,
-    ) -> Vec<ProviderEvent> {
-        let mut events = self.stage(incoming, timing_signals, completed);
-        if !self.committed {
-            events.extend(self.commit_pending(PreCommitReleaseReason::Eof));
-        }
-        events
+        incoming
+            .into_iter()
+            .map(ProviderEvent::with_replayable_prelude)
+            .collect()
     }
 
     pub(super) fn commit(&mut self, incoming: Vec<ProviderEvent>) -> Vec<ProviderEvent> {
-        if self.committed {
-            return incoming;
-        }
-        self.pending.extend(incoming);
-        // 此入口只在同批次已出现语义输出、随后失败时释放原始事件
-        self.commit_pending(PreCommitReleaseReason::SemanticOutput)
+        self.replay_closed = true;
+        incoming
     }
 
-    pub(super) fn take_for_failure(&mut self, incoming: Vec<ProviderEvent>) -> Vec<ProviderEvent> {
-        self.pending.extend(incoming);
-        self.prefetched_bytes = 0;
-        std::mem::take(&mut self.pending)
-    }
-
-    pub(super) const fn is_committed(&self) -> bool {
-        self.committed
-    }
-
-    pub(super) fn commit_pending(&mut self, reason: PreCommitReleaseReason) -> Vec<ProviderEvent> {
-        // Provider 只记录释放缓存的原因；实际下游提交仍由 Core 记录和裁决
-        if self.trace.is_enabled() {
-            self.trace.record(
-                "provider.precommit.released",
-                json!({
-                    "reason": reason,
-                    "prefetchedBytes": self.prefetched_bytes,
-                    "waitMs": self.started_at.elapsed().as_millis(),
-                }),
-            );
-        }
-        self.committed = true;
-        self.prefetched_bytes = 0;
-        std::mem::take(&mut self.pending)
+    pub(super) const fn replay_closed(&self) -> bool {
+        self.replay_closed
     }
 }
 
@@ -198,12 +120,14 @@ fn is_replaceable_prelude(event: &ProviderEvent) -> bool {
     event
         .wire_event()
         .is_some_and(|wire| match wire.event_type() {
-            Some("codex.response.metadata" | "codex.rate_limits") => true,
-            Some("response.created" | "response.in_progress") => wire
-                .data()
-                .pointer("/response/output")
-                .is_none_or(|output| output.as_array().is_some_and(Vec::is_empty)),
-            _ => false,
+            Some(event_type) => is_replaceable_response_prelude(event_type, wire.data()),
+            // 上游 SSE 注释也即时透传，但不能把传输心跳当作业务交付
+            None => wire.raw_sse_frame().is_some_and(|frame| {
+                std::str::from_utf8(frame).is_ok_and(|text| {
+                    text.lines()
+                        .all(|line| line.is_empty() || line.starts_with(':'))
+                })
+            }),
         })
 }
 

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

@@ -27,7 +27,8 @@ use crate::transport::{
     profile::CodexWireProfileState,
     protocol::{
         responses::{
-            CodexResponsesRequest, ResponsesSseFailure, TransportRequirement, transport_requirement,
+            CodexResponsesRequest, TransportRequirement, is_replaceable_response_prelude,
+            transport_requirement,
         },
         websocket::{
             websocket_audit_artifact_from_attempt, websocket_connection_limit_failure,
@@ -37,10 +38,9 @@ use crate::transport::{
     response_meta,
     websocket::{
         CodexWebSocketConnection, CodexWebSocketExchangeError, CodexWebSocketPool,
-        CodexWebSocketPoolKey, CodexWebSocketStreamingExchange, DEFAULT_STREAM_IDLE_TIMEOUT,
-        WEBSOCKET_FAST_PATH_BUDGET, WebSocketFastPath, WebSocketOriginBreaker,
-        execute_prepared_response_create_request_stream, post_send_ambiguous,
-        prepare_response_create_request_with_pool, websocket_audit_dir,
+        CodexWebSocketPoolKey, DEFAULT_STREAM_IDLE_TIMEOUT, WEBSOCKET_FAST_PATH_BUDGET,
+        WebSocketFastPath, WebSocketOriginBreaker, execute_prepared_response_create_request_stream,
+        post_send_ambiguous, prepare_response_create_request_with_pool, websocket_audit_dir,
         write_websocket_audit_artifact_from_env,
     },
 };
@@ -488,15 +488,12 @@ impl CodexBackendClient {
                 )
                 .await
                 .map_err(websocket_exchange_error_to_client_error)?;
-                if requirement.allows_connection_restart()
-                    && request.responses_lite.as_deref() == Some("true")
-                    && request.downstream_websocket_connection_id.is_some()
-                {
-                    // Lite 客户端需先拿到响应 ID 才能中断；连接级拒绝仍沿用原有分类
+                if requirement.allows_connection_restart() {
+                    // 前导帧即时交付；连接寿命拒绝仍沿用原有分类与恢复预算
                     let mut boundary_reached = false;
                     exchange.body = Box::pin(exchange.body.map(move |frame| {
                         let frame = frame?;
-                        if !boundary_reached && !is_websocket_lifecycle_prelude(&frame) {
+                        if !boundary_reached && !is_websocket_replayable_prelude(&frame) {
                             boundary_reached = true;
                             if let Some(failure) = websocket_connection_limit_failure(&frame) {
                                 return Err(CodexWebSocketExchangeError::ConnectionLimitReached(
@@ -506,20 +503,6 @@ impl CodexBackendClient {
                         }
                         Ok(frame)
                     }));
-                } else if requirement.allows_connection_restart() {
-                    match await_websocket_delivery_boundary(&mut exchange).await {
-                        Ok(DeliveryBoundary::Ready) => {}
-                        Ok(DeliveryBoundary::ConnectionLimitReached(failure)) => {
-                            return Err(CodexClientError::WebSocket(
-                                CodexWebSocketExchangeError::ConnectionLimitReached(failure),
-                            ));
-                        }
-                        Err(error) => {
-                            return Err(websocket_exchange_error_to_client_error(
-                                post_send_ambiguous(error),
-                            ));
-                        }
-                    }
                 }
                 Ok(CodexBackendStreamingResponse {
                     body: Box::pin(
@@ -630,54 +613,14 @@ impl CodexBackendClient {
     }
 }
 
-/// 首个可投递帧前的交付边界结果
-enum DeliveryBoundary {
-    /// 已越过边界，可开始向下游投递
-    Ready,
-    /// 首个可投递帧是上游连接寿命限制错误
-    ConnectionLimitReached(Box<ResponsesSseFailure>),
-}
-
-async fn await_websocket_delivery_boundary(
-    exchange: &mut CodexWebSocketStreamingExchange,
-) -> Result<DeliveryBoundary, CodexWebSocketExchangeError> {
-    let mut prelude = Vec::new();
-    loop {
-        match exchange.body.next().await {
-            Some(Ok(frame)) if is_websocket_lifecycle_prelude(&frame) => prelude.push(frame),
-            Some(Ok(frame)) => {
-                let connection_limit_failure = websocket_connection_limit_failure(&frame);
-                prelude.push(frame);
-                let remaining =
-                    std::mem::replace(&mut exchange.body, Box::pin(futures::stream::empty()));
-                exchange.body =
-                    Box::pin(futures::stream::iter(prelude.into_iter().map(Ok)).chain(remaining));
-                return Ok(if let Some(failure) = connection_limit_failure {
-                    DeliveryBoundary::ConnectionLimitReached(Box::new(failure))
-                } else {
-                    DeliveryBoundary::Ready
-                });
-            }
-            Some(Err(error)) => return Err(error),
-            None => {
-                return Err(CodexWebSocketExchangeError::StreamEndedBeforeTerminal {
-                    reason: "stream_eof",
-                    timeout: None,
-                    last_event_type: None,
-                });
-            }
-        }
-    }
-}
-
-fn is_websocket_lifecycle_prelude(frame: &[u8]) -> bool {
+fn is_websocket_replayable_prelude(frame: &[u8]) -> bool {
     serde_json::from_slice::<serde_json::Value>(frame)
         .ok()
         .is_some_and(|value| {
-            matches!(
-                value.get("type").and_then(serde_json::Value::as_str),
-                Some("response.created" | "response.in_progress")
-            )
+            value
+                .get("type")
+                .and_then(serde_json::Value::as_str)
+                .is_some_and(|event_type| is_replaceable_response_prelude(event_type, &value))
         })
 }
 

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

@@ -210,6 +210,17 @@ pub fn transport_requirement(request: &CodexResponsesRequest) -> TransportRequir
     }
 }
 
+/// 仅空响应前导和元数据可在已交付后被下一次尝试替换
+pub(crate) fn is_replaceable_response_prelude(event_type: &str, value: &Value) -> bool {
+    match event_type {
+        "codex.response.metadata" | "codex.rate_limits" => true,
+        "response.created" | "response.in_progress" => value
+            .pointer("/response/output")
+            .is_none_or(|output| output.as_array().is_some_and(Vec::is_empty)),
+        _ => false,
+    }
+}
+
 /// 单个 Responses 事件对计时系统提供的稳定语义信号
 ///
 /// `output_start` 只表示官方输出项开始事件；`semantic_output` 仍要求

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

@@ -4374,7 +4374,10 @@ async fn websocket_idle_timeout_diagnosis_survives_ambiguous_send_wrapping() {
             .await
             .expect("partial response")
             .expect("valid partial response");
-        if event.has_client_event() {
+        if event
+            .wire_event()
+            .is_some_and(|wire| wire.event_type() == Some("response.output_text.delta"))
+        {
             break;
         }
     }
@@ -4593,7 +4596,7 @@ async fn ambiguous_websocket_close_reconnects_and_retains_native_continuation()
         websocket
             .send(Message::Text(
                 json!({
-                    "type": "response.output_item.added",
+                    "type": "response.in_progress",
                     "output_index": 0,
                     "item": {
                         "id": "msg_structural",
@@ -4710,8 +4713,8 @@ async fn ambiguous_websocket_close_reconnects_and_retains_native_continuation()
                     }
                 }
                 assert!(
-                    !event.has_client_event(),
-                    "structural events must remain behind the replay boundary"
+                    !event.has_client_event() || event.is_replayable_prelude(),
+                    "delivered preludes preserve the replay boundary"
                 );
             }
             Some(Err(error)) => break error,
@@ -4736,7 +4739,7 @@ async fn ambiguous_websocket_close_reconnects_and_retains_native_continuation()
     assert_eq!(
         error.diagnostic().map(|diagnostic| diagnostic.as_str()),
         Some(
-            "OpenAI WebSocket closed before a terminal response (close code 1000); last event type: response.output_item.added"
+            "OpenAI WebSocket closed before a terminal response (close code 1000); last event type: response.in_progress"
         )
     );
     let raw_close: Value = serde_json::from_str(
@@ -4751,7 +4754,7 @@ async fn ambiguous_websocket_close_reconnects_and_retains_native_continuation()
     assert_eq!(raw_close.get("reason"), Some(&json!("")));
     assert_eq!(
         raw_close.get("last_event_type"),
-        Some(&json!("response.output_item.added"))
+        Some(&json!("response.in_progress"))
     );
     let connection = error
         .connection_observation()
@@ -8963,7 +8966,11 @@ async fn capacity_feedback_in_stream_only_counts_explicit_overload() {
                 original
             );
             if !semantic_output {
-                assert!(client_events.is_empty());
+                assert!(
+                    client_events
+                        .iter()
+                        .all(|event| event.is_replayable_prelude())
+                );
             }
             client_events.extend(error.take_atomic_client_events());
             let wire = client_events
@@ -8980,7 +8987,7 @@ async fn capacity_feedback_in_stream_only_counts_explicit_overload() {
 }
 
 #[tokio::test]
-async fn ordinary_request_should_hold_created_until_later_failure_can_rotate() {
+async fn ordinary_request_forwards_created_and_preserves_later_rotation() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_first_event_latency").await;
     let (base_url, release, first_chunk_sent, server) = paused_chunked_sse_server(
@@ -9004,39 +9011,18 @@ async fn ordinary_request_should_hold_created_until_later_failure_can_rotate() {
         .await
         .expect("prepare provider stream");
 
-    let mut first_chunk_sent = Box::pin(first_chunk_sent);
-    loop {
-        tokio::select! {
-            sent = &mut first_chunk_sent => {
-                sent.expect("first upstream chunk");
-                break;
-            }
-            next = stream.next() => {
-                let event = next
-                    .expect("provider stream must stay open")
-                    .expect("provider event");
-                assert!(!event.has_client_event(), "response.created must remain replayable");
-            }
-        }
-    }
-
-    let exposed = timeout(Duration::from_millis(100), async {
-        loop {
-            let next = stream
-                .next()
-                .await
-                .expect("provider stream must stay open")
-                .expect("provider event");
-            if next.has_client_event() {
-                return next;
-            }
-        }
-    })
-    .await;
-    assert!(
-        exposed.is_err(),
-        "a structural event must not commit the downstream before a later 429"
+    let created = timeout(
+        Duration::from_secs(1),
+        precommit::next_client_event(&mut stream),
+    )
+    .await
+    .expect("created must arrive before the next upstream chunk");
+    assert_eq!(
+        created.wire_event().unwrap().event_type(),
+        Some("response.created")
     );
+    assert!(created.is_replayable_prelude());
+    first_chunk_sent.await.expect("first chunk sent");
 
     release.send(()).expect("release second upstream chunk");
     let mut failure = loop {
@@ -9060,13 +9046,13 @@ async fn ordinary_request_should_hold_created_until_later_failure_can_rotate() {
             .iter()
             .filter_map(|event| event.wire_event()?.event_type())
             .collect::<Vec<_>>(),
-        vec!["response.created", "response.failed"]
+        vec!["response.failed"]
     );
     server.await.expect("chunked SSE server");
 }
 
 #[tokio::test]
-async fn ordinary_request_should_hold_structural_events_until_upstream_finishes() {
+async fn ordinary_request_forwards_structural_events_before_upstream_finishes() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_bounded_replay_grace").await;
     let (base_url, release, _first_chunk_sent, server) = paused_chunked_sse_server(
@@ -9086,29 +9072,29 @@ async fn ordinary_request_should_hold_structural_events_until_upstream_finishes(
         .await
         .expect("prepare provider stream");
 
-    assert!(
-        timeout(Duration::from_secs(3), async {
-            while let Some(event) = stream.next().await {
-                assert!(!event.expect("provider event").has_client_event());
-            }
-        })
-        .await
-        .is_err()
+    let created = timeout(
+        Duration::from_secs(1),
+        precommit::next_client_event(&mut stream),
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        created.wire_event().unwrap().event_type(),
+        Some("response.created")
     );
+    assert!(created.is_replayable_prelude());
     release.send(()).expect("finish upstream response");
     let events = stream.collect::<Vec<_>>().await;
-    assert!(events.iter().any(|event| {
-        event
-            .as_ref()
-            .unwrap()
-            .wire_event()
-            .is_some_and(|wire| wire.event_type() == Some("response.created"))
-    }));
+    assert!(
+        events
+            .iter()
+            .all(|event| !event.as_ref().unwrap().has_client_event())
+    );
     server.await.expect("chunked SSE server");
 }
 
 #[tokio::test]
-async fn continuation_should_hold_created_until_semantic_output() {
+async fn continuation_forwards_created_before_semantic_output() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_continuation_prefetch").await;
     let (base_url, release, first_chunk_sent, server) = paused_chunked_sse_server(
@@ -9137,42 +9123,24 @@ async fn continuation_should_hold_created_until_semantic_output() {
         .await
         .expect("prepare provider stream");
 
-    let mut first_chunk_sent = Box::pin(first_chunk_sent);
-    loop {
-        tokio::select! {
-            sent = &mut first_chunk_sent => {
-                sent.expect("first upstream chunk");
-                break;
-            }
-            next = stream.next() => {
-                let event = next
-                    .expect("provider stream must stay open")
-                    .expect("provider event");
-                assert!(!event.has_client_event(), "created was exposed before the first chunk barrier");
-            }
-        }
-    }
-
-    let blocked = timeout(Duration::from_millis(100), async {
-        loop {
-            let next = stream
-                .next()
-                .await
-                .expect("provider stream must stay open")
-                .expect("provider event");
-            if next.has_client_event() {
-                return next;
-            }
-        }
-    })
-    .await;
-    assert!(blocked.is_err());
+    let created = timeout(
+        Duration::from_secs(1),
+        precommit::next_client_event(&mut stream),
+    )
+    .await
+    .expect("created must arrive before the next upstream chunk");
+    assert_eq!(
+        created.wire_event().unwrap().event_type(),
+        Some("response.created")
+    );
+    assert!(created.is_replayable_prelude());
+    first_chunk_sent.await.expect("first chunk sent");
 
     release.send(()).expect("release semantic output chunk");
     let first_event = loop {
         let next = timeout(Duration::from_secs(1), stream.next())
             .await
-            .expect("semantic output must release the prefetched batch")
+            .expect("semantic output arrives after release")
             .expect("provider stream must stay open")
             .expect("provider event");
         if next.has_client_event() {
@@ -9181,7 +9149,7 @@ async fn continuation_should_hold_created_until_semantic_output() {
     };
     assert_eq!(
         first_event.wire_event().and_then(|wire| wire.event_type()),
-        Some("response.created")
+        Some("response.content_part.added")
     );
     while let Some(event) = stream.next().await {
         event.expect("continuation response must complete");
@@ -9595,7 +9563,7 @@ async fn exact_websocket_busy_then_replay_scope_relaxation_is_rejected_before_se
 }
 
 #[tokio::test]
-async fn continuation_prefetch_over_128_kib_stays_buffered_until_eof_without_protocol_failure() {
+async fn large_continuation_prelude_is_delivered_before_eof_without_protocol_failure() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_prefetch_limit").await;
     let padding = "x".repeat(128 * 1024);
@@ -9622,36 +9590,13 @@ async fn continuation_prefetch_over_128_kib_stays_buffered_until_eof_without_pro
         )
         .await
         .expect("prepare provider stream");
-    let visible = {
-        let next_visible = async {
-            loop {
-                let event = stream
-                    .next()
-                    .await
-                    .expect("provider stream remains open")
-                    .expect("large prefetch cannot create a protocol failure");
-                if event.has_client_event() {
-                    break event;
-                }
-            }
-        };
-        tokio::pin!(next_visible);
-        assert!(
-            timeout(Duration::from_secs(1), next_visible.as_mut())
-                .await
-                .is_err(),
-            "large structural events must remain buffered before semantic output"
-        );
-        assert!(
-            timeout(Duration::from_secs(3), next_visible.as_mut())
-                .await
-                .is_err()
-        );
-        release.send(()).expect("finish upstream response");
-        timeout(Duration::from_secs(1), next_visible)
-            .await
-            .expect("EOF releases original wire")
-    };
+    let visible = timeout(
+        Duration::from_secs(1),
+        precommit::next_client_event(&mut stream),
+    )
+    .await
+    .expect("large prelude is delivered before EOF");
+    release.send(()).expect("finish upstream response");
 
     assert_eq!(
         visible.wire_event().and_then(|wire| wire.event_type()),
@@ -10415,6 +10360,18 @@ async fn connection_limit_rejection_requests_one_provider_retry_and_reconnects_i
             let (stream, _) = listener.accept().await.expect("accept WS");
             let mut ws = accept_codex_test_websocket(stream).await;
             ws.next().await.expect("request").expect("valid frame");
+            // 真实上游先发送元数据，随后仍可能在空前导后拒绝连接寿命
+            if first {
+                for event in [
+                    json!({"type":"codex.response.metadata","model":"gpt-5.4"}),
+                    json!({"type":"response.created","response":{"id":"resp_expired","output":[]}}),
+                    json!({"type":"response.in_progress","response":{"id":"resp_expired","output":[]}}),
+                ] {
+                    ws.send(Message::Text(event.to_string().into()))
+                        .await
+                        .expect("prelude");
+                }
+            }
             let event = if first {
                 json!({"type":"error","status":400,"headers":{"x-request-id":"req-limit-retry"},"error":{"code":"websocket_connection_limit_reached","type":"invalid_request_error","message":"connection expired"}})
             } else {
@@ -10725,6 +10682,7 @@ async fn assert_websocket_failure_headers(headers: Value, request_id: Option<&st
         )
         .await
         .expect("prepare");
+    let mut delivered = Vec::new();
     let mut observation = None;
     let mut error = timeout(Duration::from_secs(5), async {
         loop {
@@ -10733,6 +10691,7 @@ async fn assert_websocket_failure_headers(headers: Value, request_id: Option<&st
                     if let Some(current) = event.response_observation() {
                         observation = Some(current.clone());
                     }
+                    delivered.push(event);
                 }
                 Some(Err(error)) => break error,
                 None => panic!("expected error"),
@@ -10765,7 +10724,8 @@ async fn assert_websocket_failure_headers(headers: Value, request_id: Option<&st
     assert_eq!(error.send_state(), UpstreamSendState::Sent);
     assert!(!error.replay_is_safe());
     assert_eq!(error.raw_upstream_error().expect("raw error").as_str(), raw);
-    let events = error.take_atomic_client_events();
+    delivered.extend(error.take_atomic_client_events());
+    let events = delivered;
     let event_type = if reuse { "response.failed" } else { "error" };
     assert_eq!(
         events
@@ -10773,7 +10733,7 @@ async fn assert_websocket_failure_headers(headers: Value, request_id: Option<&st
             .filter_map(|event| event.wire_event()?.event_type())
             .collect::<Vec<_>>(),
         vec!["response.metadata", event_type],
-        "original metadata and failure stay atomically deliverable",
+        "original metadata and failure are delivered in order",
     );
     let wire = events
         .iter()
@@ -11633,7 +11593,9 @@ async fn quota_continuation_stream_rejection_only_projects_before_delivery() {
                 client_events.extend(error.take_atomic_client_events());
                 if native_continuation && !semantic_output {
                     assert!(
-                        client_events.is_empty(),
+                        client_events
+                            .iter()
+                            .all(|event| event.is_replayable_prelude()),
                         "raw quota error must not override replay projection"
                     );
                     assert_eq!(
@@ -11694,15 +11656,17 @@ async fn late_quota_continuation_rejection_requires_client_replay_before_output(
         )
         .await
         .unwrap();
-    assert!(
-        timeout(Duration::from_secs(3), async {
-            while let Some(event) = stream.next().await {
-                assert!(!event.unwrap().has_client_event());
-            }
-        })
-        .await
-        .is_err()
+    let created = timeout(
+        Duration::from_secs(1),
+        precommit::next_client_event(&mut stream),
+    )
+    .await
+    .unwrap();
+    assert_eq!(
+        created.wire_event().unwrap().event_type(),
+        Some("response.created")
     );
+    assert!(created.is_replayable_prelude());
     release.send(()).unwrap();
     let mut delivered_failure = false;
     let error = loop {

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

@@ -1,4 +1,4 @@
-//! 验证结构事件缓冲、长时间等待与取消对重试边界的影响
+//! 验证事件即时交付、长时间进度与取消对重试边界的影响
 
 use gateway_core::diagnostics::TraceContext;
 use gateway_core::engine::provider::ProviderStream;
@@ -49,16 +49,6 @@ fn sse(event: &Value) -> String {
     encode_sse_event(event["type"].as_str().unwrap(), &event.to_string())
 }
 
-fn releases(trace: &TraceContext) -> Vec<Value> {
-    trace.snapshot().unwrap()["events"]
-        .as_array()
-        .unwrap()
-        .iter()
-        .filter(|event| event["stage"] == "provider.precommit.released")
-        .cloned()
-        .collect()
-}
-
 pub(super) async fn next_client_event(stream: &mut ProviderStream) -> ProviderEvent {
     loop {
         let event = stream.next().await.unwrap().unwrap();
@@ -81,7 +71,7 @@ pub(super) async fn stream_failure(
 }
 
 #[tokio::test]
-async fn large_structural_events_preserve_overload_replay_on_http_and_websocket() {
+async fn large_preludes_are_forwarded_before_failure_on_http_and_websocket() {
     for websocket in [false, true] {
         let store = Arc::new(MemoryAccountStore::default());
         create_account(&store, "acct_provider_contract").await;
@@ -89,7 +79,7 @@ async fn large_structural_events_preserve_overload_replay_on_http_and_websocket(
         let created = structural_event("response.created", 300 * 1024);
         let progress = structural_event("response.in_progress", 300 * 1024);
         let failure = overload();
-        let expected = vec![created.clone(), progress.clone(), failure.clone()];
+        let expected = vec![failure.clone()];
         let (base_url, release, server) = if websocket {
             let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
             let base_url = format!("http://{}", listener.local_addr().unwrap());
@@ -128,11 +118,13 @@ async fn large_structural_events_preserve_overload_replay_on_http_and_websocket(
             .await
             .unwrap();
 
-        assert!(
-            timeout(Duration::from_millis(3_500), next_client_event(&mut stream))
+        for kind in ["response.created", "response.in_progress"] {
+            let event = timeout(Duration::from_secs(1), next_client_event(&mut stream))
                 .await
-                .is_err()
-        );
+                .unwrap();
+            assert_eq!(event.wire_event().unwrap().event_type(), Some(kind));
+            assert!(event.is_replayable_prelude());
+        }
         release.send(()).unwrap();
         let (events, mut error) = timeout(Duration::from_secs(1), stream_failure(&mut stream))
             .await
@@ -149,10 +141,6 @@ async fn large_structural_events_preserve_overload_replay_on_http_and_websocket(
             .filter_map(|event| event.wire_event().map(|wire| wire.data().clone()))
             .collect();
         assert_eq!(actual, expected);
-        assert!(
-            releases(&trace).is_empty(),
-            "discardable failure must not be reported as a release"
-        );
         server.await.unwrap();
     }
 }
@@ -185,11 +173,13 @@ async fn late_overload_after_multiple_structural_events_remains_replayable() {
         )
         .await
         .unwrap();
-    assert!(
-        timeout(Duration::from_millis(3_500), next_client_event(&mut stream))
+    for kind in ["response.created", "response.in_progress"] {
+        let event = timeout(Duration::from_secs(2), next_client_event(&mut stream))
             .await
-            .is_err()
-    );
+            .unwrap();
+        assert_eq!(event.wire_event().unwrap().event_type(), Some(kind));
+        assert!(event.is_replayable_prelude());
+    }
     release.send(()).unwrap();
     let (events, mut error) = timeout(Duration::from_secs(1), stream_failure(&mut stream))
         .await
@@ -206,14 +196,13 @@ async fn late_overload_after_multiple_structural_events_remains_replayable() {
             .iter()
             .filter_map(|event| event.wire_event()?.event_type())
             .collect::<Vec<_>>(),
-        ["response.created", "response.in_progress", "error"]
+        ["error"]
     );
-    assert!(releases(&trace).is_empty());
     server.await.unwrap();
 }
 
 #[tokio::test]
-async fn immediate_release_preserves_wire_and_records_the_boundary_once() {
+async fn immediate_delivery_preserves_wire_through_output_terminal_and_eof() {
     let semantic = json!({"type": "response.output_text.delta", "output_index": 0, "content_index": 0, "delta": "hello"});
     let terminal = json!({"type": "response.completed", "response": {"id": "resp_precommit", "model": "gpt-5.4", "status": "completed", "output": []}});
     let tool = json!({"type": "response.web_search_call.in_progress", "item_id": "search_precommit", "output_index": 0});
@@ -270,16 +259,12 @@ async fn immediate_release_preserves_wire_and_records_the_boundary_once() {
             }
         }
         assert_eq!(wire, body.as_bytes());
-        let releases = releases(&trace);
-        assert_eq!(releases.len(), 1);
-        assert_eq!(releases[0]["data"]["reason"], reason);
-        assert_eq!(releases[0]["data"]["prefetchedBytes"], body.len());
         server.await.unwrap();
     }
 }
 
 #[tokio::test]
-async fn cancelling_buffered_structural_events_does_not_release_or_offer_replay() {
+async fn cancelling_after_prelude_delivery_does_not_offer_replay() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_provider_contract").await;
     let (base_url, release, _, server) = paused_chunked_sse_server(
@@ -296,10 +281,12 @@ async fn cancelling_buffered_structural_events_does_not_release_or_offer_replay(
         )
         .await
         .unwrap();
-    assert!(
-        timeout(Duration::from_millis(100), next_client_event(&mut stream))
-            .await
-            .is_err()
+    let created = timeout(Duration::from_secs(1), next_client_event(&mut stream))
+        .await
+        .unwrap();
+    assert_eq!(
+        created.wire_event().unwrap().event_type(),
+        Some("response.created")
     );
     cancellation.cancel();
     let (events, error) = timeout(Duration::from_secs(1), stream_failure(&mut stream))
@@ -308,7 +295,6 @@ async fn cancelling_buffered_structural_events_does_not_release_or_offer_replay(
     assert_eq!(error.kind(), ProviderErrorKind::Cancelled);
     assert!(!error.replay_is_safe());
     assert!(events.iter().all(|event| !event.has_client_event()));
-    assert!(releases(&trace).is_empty());
     release.send(()).unwrap();
     server.await.unwrap();
 }
@@ -347,10 +333,12 @@ async fn twenty_minute_structural_progress_preserves_late_overload_recovery() {
         )
         .await
         .unwrap();
-    assert!(
-        timeout(Duration::from_millis(100), next_client_event(&mut stream))
-            .await
-            .is_err()
+    let created = timeout(Duration::from_secs(1), next_client_event(&mut stream))
+        .await
+        .unwrap();
+    assert_eq!(
+        created.wire_event().unwrap().event_type(),
+        Some("response.created")
     );
     // 结构进度保持上游传输活跃，长思考不能额外触发首内容等待期限
     for _ in 0..20 {
@@ -358,11 +346,14 @@ async fn twenty_minute_structural_progress_preserves_late_overload_recovery() {
         tokio::time::advance(Duration::from_secs(60)).await;
         tokio::time::resume();
         progress.send(()).await.unwrap();
-        assert!(
-            timeout(Duration::from_millis(50), next_client_event(&mut stream))
-                .await
-                .is_err()
+        let event = timeout(Duration::from_secs(1), next_client_event(&mut stream))
+            .await
+            .unwrap();
+        assert_eq!(
+            event.wire_event().unwrap().event_type(),
+            Some("response.in_progress")
         );
+        assert!(event.is_replayable_prelude());
     }
     drop(progress);
     let (events, error) = timeout(Duration::from_secs(1), stream_failure(&mut stream))
@@ -374,12 +365,11 @@ async fn twenty_minute_structural_progress_preserves_late_overload_recovery() {
         error.pre_delivery_retry(),
         Some(PreDeliveryRetry::SameAccountTransientRetry { .. })
     ));
-    assert!(releases(&trace).is_empty());
     server.await.unwrap();
 }
 
 #[tokio::test]
-async fn structural_preamble_over_16_mib_preserves_late_overload_recovery() {
+async fn large_prelude_stream_is_forwarded_without_retaining_it_for_failure() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_provider_contract").await;
     // 大段配置回显仍属于前导事件，累计体积不应改变上游失败的恢复方式
@@ -394,11 +384,16 @@ async fn structural_preamble_over_16_mib_preserves_late_overload_recovery() {
         )
         .await
         .unwrap();
-    assert!(
-        timeout(Duration::from_secs(1), next_client_event(&mut stream))
+    for _ in 0..17 {
+        let event = timeout(Duration::from_secs(2), next_client_event(&mut stream))
             .await
-            .is_err()
-    );
+            .unwrap();
+        assert_eq!(
+            event.wire_event().unwrap().event_type(),
+            Some("response.in_progress")
+        );
+        assert!(event.is_replayable_prelude());
+    }
     release.send(()).unwrap();
     let (events, mut error) = timeout(Duration::from_secs(10), stream_failure(&mut stream))
         .await
@@ -415,8 +410,7 @@ async fn structural_preamble_over_16_mib_preserves_late_overload_recovery() {
         .filter_map(|event| event.wire_event()?.raw_sse_frame())
         .map(bytes::Bytes::len)
         .sum();
-    assert!(buffered_bytes > 16 * 1024 * 1024);
-    assert!(releases(&trace).is_empty());
+    assert_eq!(buffered_bytes, sse(&overload()).len());
     server.await.unwrap();
 }
 
@@ -455,7 +449,12 @@ async fn websocket_disconnect_recovery_honors_request_retry_budget_without_sessi
                 .await
                 .unwrap();
             let (events, error) = stream_failure(&mut stream).await;
-            assert!(events.iter().all(|event| !event.has_client_event()));
+            let delivered: Vec<_> = events
+                .iter()
+                .filter(|event| event.has_client_event())
+                .collect();
+            assert_eq!(delivered.len(), 1);
+            assert!(delivered[0].is_replayable_prelude());
             assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
             assert!(!error.replay_is_safe());
             if retry_count < max_retries {
@@ -468,8 +467,107 @@ async fn websocket_disconnect_recovery_honors_request_retry_budget_without_sessi
                     Some(PreDeliveryRetry::SameAccountTransportFallback)
                 );
             }
-            assert!(releases(&trace).is_empty());
         }
         server.await.unwrap();
     }
 }
+
+#[tokio::test]
+async fn sse_comments_are_forwarded_without_closing_the_retry_window() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_provider_contract").await;
+    let heartbeat = ": keep-alive\r\n\r\n";
+    let (base_url, release, _, server) =
+        paused_chunked_sse_server(heartbeat.to_owned(), sse(&overload())).await;
+    let mut stream = provider_with_base_url(&store, base_url)
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            context("req_heartbeat_retry", CancellationToken::new()),
+        )
+        .await
+        .unwrap();
+    let event = timeout(Duration::from_secs(1), next_client_event(&mut stream))
+        .await
+        .expect("heartbeat cannot wait for a later event");
+    assert_eq!(
+        event.wire_event().unwrap().raw_sse_frame().unwrap(),
+        heartbeat.as_bytes()
+    );
+    assert!(event.is_replayable_prelude());
+    release.send(()).unwrap();
+    let (_, error) = stream_failure(&mut stream).await;
+    assert!(error.replay_is_safe());
+    assert!(matches!(
+        error.pre_delivery_retry(),
+        Some(PreDeliveryRetry::SameAccountTransientRetry { .. })
+    ));
+    server.await.unwrap();
+}
+
+#[tokio::test]
+async fn ordinary_streams_forward_compaction_output_and_unknown_events_without_reopening_retry() {
+    for websocket in [false, true] {
+        for first in [
+            json!({"type":"response.compaction.compacting","item_id":"compact_test","output_index":0}),
+            json!({"type":"response.output_text.delta","output_index":0,"content_index":0,"delta":"partial"}),
+            json!({"type":"response.future","value":"opaque"}),
+        ] {
+            let store = Arc::new(MemoryAccountStore::default());
+            create_account(&store, "acct_provider_contract").await;
+            let prelude = structural_event("response.created", 1024);
+            let (base_url, release, server) = if websocket {
+                let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+                let base_url = format!("http://{}", listener.local_addr().unwrap());
+                let (release, released) = oneshot::channel();
+                let first = first.clone();
+                let server = tokio::spawn(async move {
+                    let (socket, _) = listener.accept().await.unwrap();
+                    let mut socket = accept_codex_test_websocket(socket).await;
+                    socket.next().await.unwrap().unwrap();
+                    for event in [first, prelude] {
+                        socket
+                            .send(Message::Text(event.to_string().into()))
+                            .await
+                            .unwrap();
+                    }
+                    released.await.unwrap();
+                    socket
+                        .send(Message::Text(overload().to_string().into()))
+                        .await
+                        .unwrap();
+                });
+                (base_url, release, server)
+            } else {
+                let (base_url, release, _, server) =
+                    paused_chunked_sse_server(sse(&first) + &sse(&prelude), sse(&overload())).await;
+                (base_url, release, server)
+            };
+            let operation = if websocket {
+                generate_operation()
+            } else {
+                http_generate_operation()
+            };
+            let mut stream = provider_with_base_url(&store, base_url)
+                .execute(
+                    planned_request("openai", operation),
+                    context("req_closed_retry", CancellationToken::new()),
+                )
+                .await
+                .unwrap();
+            let delivered = timeout(Duration::from_secs(1), next_client_event(&mut stream))
+                .await
+                .unwrap();
+            assert_eq!(delivered.wire_event().unwrap().data(), &first);
+            assert!(!delivered.is_replayable_prelude());
+            let created = timeout(Duration::from_secs(1), next_client_event(&mut stream))
+                .await
+                .unwrap();
+            assert!(!created.is_replayable_prelude());
+            release.send(()).unwrap();
+            let (_, error) = stream_failure(&mut stream).await;
+            assert!(!error.replay_is_safe());
+            assert!(error.pre_delivery_retry().is_none());
+            server.await.unwrap();
+        }
+    }
+}

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

@@ -105,12 +105,18 @@ async fn ordinary_stream_failures_preserve_original_events_without_proxy_retry()
                 )
                 .await
                 .unwrap();
-            let (events, mut error) = stream_failure(&mut stream).await;
-            assert!(events.iter().all(|event| !event.has_client_event()));
+            let (mut events, mut error) = stream_failure(&mut stream).await;
+            assert!(
+                events
+                    .iter()
+                    .filter(|event| event.has_client_event())
+                    .all(|event| event.is_replayable_prelude())
+            );
             assert!(error.retry_is_prohibited());
             assert!(error.pre_delivery_retry().is_none());
             assert_eq!(error.retry_after(), Some(Duration::from_secs(17)));
-            let atomic = error.take_atomic_client_events();
+            events.extend(error.take_atomic_client_events());
+            let atomic = events;
             assert_eq!(
                 atomic
                     .iter()

deploy/README.md

@@ -564,10 +564,6 @@ docker compose -f deploy/compose.yaml up -d --no-build --wait
 
 升级后通过管理端版本接口或容器 image digest 确认运行实例的版本和 revision
 
-管理端请求默认等待 120 秒。直接构建前端时可通过进程环境 `VITE_API_TIMEOUT_MS` 指定毫秒数，
-例如 `VITE_API_TIMEOUT_MS=180000 pnpm --dir frontend build`，取值须为正整数。
-该值在构建时写入静态资源，修改后需要重新构建和部署前端。单个请求显式指定的时限仍优先
-
 构建元数据通过一次性进程环境传入：
 
 ```bash

docs/api.md

@@ -443,20 +443,22 @@ xAI 的 `reasoning.effort` 接受 `none / minimal / low / medium / high / xhigh
 
 #### 容量拒绝与重试
 
-原生 Responses Lite WebSocket 在未启用中间件链时，按顺序及时交付正常事件，
-让客户端取得中断所需的响应 ID。无输出的 `response.created`、`response.in_progress`
+Responses 的 HTTP/SSE 与 WebSocket 按顺序即时交付正常事件，不暂存前导等待后续输出。
+未启用中间件链时，无输出的 `response.created`、`response.in_progress`
 及 `codex.response.metadata`、`codex.rate_limits` 元数据交付后仍可按已有安全条件重试，
 客户端可能收到下一次尝试的新 `response.created`。
-其他事件也立即交付，但会关闭透明重试窗口，后续前导事件不再恢复重试资格；
-错误事件先按既有重试规则处理，无法重试时交付错误
+其他事件也立即交付，但会关闭透明重试窗口，后续前导事件不再恢复重试资格。
+中间件链不启用可替换前导例外；错误事件先按既有重试规则处理，无法重试时交付错误
 
-其他 WebSocket 与 HTTP/SSE 的前导事件保留到首次有效文本、推理、工具活动或终态再交付。
-前导缓冲不另设等待时长或累计体积上限，等待沿用[请求期限](#请求期限)与传输超时配置。
+等待沿用[请求期限](#请求期限)与传输超时配置。
 WebSocket 独立 Ping 保活；Responses HTTP 流式请求完成准入后，执行仍在等待且连续 15 秒无输出时提交 `200`
-和 SSE 注释心跳，此后的等待继续每 15 秒保活，包括请求中间件等待和前导事件缓冲。
+和 SSE 注释心跳，此后的等待继续每 15 秒保活，包括请求中间件等待和上游尚未返回事件。
 首次心跳前仍保留最终 HTTP 状态与响应头；心跳发出后发生的失败使用 `response.failed` 和 `[DONE]`，
 后续取得的上游或插件响应头无法再补发，`Retry-After` 转入流内错误的 `response.error.headers`。
-心跳不代表业务输出，不关闭已有的安全重试窗口
+心跳不代表业务输出，不关闭已有的安全重试窗口。
+SSE 注释和 WebSocket Ping/Pong 不重置官方 Codex 客户端的事件空闲超时，
+其 `stream_idle_timeout_ms` 默认是 300 秒；长请求需持续收到真实事件，
+或同时调整客户端与上游传输的空闲等待预算，单独缩短心跳间隔不能延长业务等待
 
 WebSocket 在交付有效输出前断线或超时，可按 `stream_max_retries` 做同账号、有退避的尽力重试
 （默认最多 5 次），耗尽后仅在请求允许时改用 HTTP。已发送或发送结果不明不单独阻止这项恢复；
@@ -1952,9 +1954,9 @@ WebSocket 握手默认请求专项计时，客户端显式提供的开关保留
 
 管理端下载的诊断包 `schemaVersion: 4` 用于人工反馈，不是备份或导入格式。它包含关联 ID、错误分类摘要、
 请求与错误事件各自的状态、attempt、时间线阶段、计时和上述失败分类。
-`provider.precommit.released` 事件的 `precommitRelease` 保留 `reason / prefetchedBytes / waitMs`，
-分别表示 Provider 释放缓存的原因、累计预取字节数和等待毫秒数，不等同于下游已提交。
-当前释放原因为 `semantic_output / terminal / eof`，导出也接受历史记录中的 `grace_timeout / byte_limit`。
+诊断包兼容已有 `provider.precommit.released` 记录，其 `precommitRelease` 保留 `reason / prefetchedBytes / waitMs`，
+分别表示 Provider 释放暂存的原因、累计预取字节数和等待毫秒数，不等同于下游已提交。
+导出接受记录中的 `semantic_output / terminal / eof / grace_timeout / byte_limit`。
 数值只允许非负安全整数，缺失或未通过校验的字段为 `null`；其他事件的 `precommitRelease` 为 `null`
 
 诊断包不自动导出 message/raw error、任意 metadata、其他 trace event data、请求响应正文和头部。

docs/architecture.md

@@ -384,15 +384,15 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别客户端，Cor
 - downstream commit 记录不可撤回的首次交付；不可重放事件交付后禁止换号、重试和 fallback
   HTTP 保活可先提交响应头，Core 记录该首字节时间与实际状态，但首个业务批次仍独立提交；
   SSE 注释不构成业务输出或上游已发送的证明，也不改变重放资格
-  原生 Responses Lite WebSocket 及时交付正常事件，仅 Provider 明确标记的可替换前导保留重试资格；Core 单独维护重放屏障，
+  原生 Responses HTTP/SSE 与 WebSocket 即时交付正常事件，仅 Provider 明确标记的可替换前导保留重试资格；Core 单独维护重放屏障，
   仍检查错误分类、续接归属、发送状态、副作用及预算。中间件链不启用此前导例外
 - API 在最终错误编码出口投影客户端恢复信号；该投影不修改 Provider 上游事实，也不改变 Core 的
   重试与提交边界。具体错误合同见 [数据面接口](api.md#3-openai-数据面与模型目录)
 - Provider 可将明确容量拒绝标记为有界同账号退避，Core 在既有安全重放边界内执行，按账号维护请求内
   预算，耗尽后复用普通换号路径。该退避消耗总路由预算，与 WS 传输恢复、OAuth 刷新及账号额度冷却分开
-- OpenAI Provider 在原生 Responses Lite WebSocket 上不等待后续事件再交付正常帧；
-  未知事件立即交付并结束重放窗口。其他 WebSocket 及 HTTP/SSE 保留到有效输出或终态，
-  不按等待时长或累计体积提前交付，中间件链沿用缓冲策略。
+- OpenAI Provider 与 WS 传输层不暂存前导等待后续事件，正常帧按顺序即时交付；
+  Provider 只保留单向推进的重放边界，实际输出、终态或未知事件关闭重放窗口。
+  中间件链中的正常事件同样即时交付，但不保留可替换前导的重试资格。
   有效输出交付前的 WS 传输错误可请求有界同账号尽力恢复，即使发送结果不明也保留此恢复意图；
   发送事实与安全重放证明不因此改写。Provider 限定传输次数与续接范围，Core 保持原账号并检查
   取消、deadline、已观测外部副作用及下游交付边界

docs/plans/413-postgresql-capacity.md

@@ -1,81 +0,0 @@
-# PostgreSQL 3100 万请求优化与验收
-
-本报告对应 [Issue #413](https://github.com/zyycn/codex-proxy-rs/issues/413)，范围为大数据量下的统计、请求分页、账号用量、分组费用与额度预测
-
-3100 万条源请求、窄事实和完整汇总已构建并独立对账。当前 HTTP 矩阵执行 167 次请求，包含 145 次合法业务请求、22 次预期参数或权限拒绝，未出现超时或非预期失败。按实际注册路由计，84 条已有通过记录、70 条未覆盖、6 条不适用；不宣称全接口或长期负载验收通过
-
-## 问题与实现边界
-
-原账号用量读取在命中事实超过 10 万条时拒绝查询；其他统计的事实回退与汇总读取也有固定行数门槛。仅放宽门槛会继续放大应用端逐行合并、深页 OFFSET 和预测窗口排序的成本
-
-实现继续使用原生 PostgreSQL、现有窄事实、十五分钟／小时汇总和投影 Worker：
-
-- 账号与 Key 的必要模型联合维度进入共享汇总，完整桶和精确边界在数据库内聚合，不以命中数据量拒绝查询
-- 账号窗口、用量排序、分组费用和分页计数在同一读快照补齐待投影修改与删除，保留精确总数、金额及未知值语义
-- 请求深页通过时间分段定位，错误列表合并两类来源后保持稳定排序
-- 预测使用开始时间与完成时间二维汇总，累计量保留完整窗口，Provider 文档独立有界采样
-- 投影在同一事务撤销旧贡献并加入新贡献；普通更新维护构建前缀，不反复重启整个桶
-- 观测操作与前端默认保护时限为 120 秒，可分别配置；30 秒为优化方向，不能将提高超时时限当作查询提速
-
-使用记录搜索支持密钥、账号与模型，不匹配请求 ID；请求 ID 的显式精确筛选仍保留。汇总、洞察和诊断不接受搜索参数，管理观测接口不接受路由分组筛选
-
-当前合同与职责见 [API 参考](../api.md)、[系统架构](../architecture.md#10-观测与后台任务)与[部署配置](../../deploy/README.md)，本报告不重复字段清单
-
-## 环境与样本
-
-| 项目 | 实际条件 |
-| --- | --- |
-| 数据 | 31 天、3100 万条合成源请求；50 个账号、100 个 Key、20 个模型、10 个分组 |
-| 时间范围 | `2026-09-12T04:45:00Z` 至 `2026-10-13T04:45:00Z` |
-| 数据库 | PostgreSQL 18.6，容器限制 4 CPU、4 GiB 内存，持久磁盘 |
-| 主要参数 | shared_buffers 512 MiB、work_mem 16 MiB、max_parallel_workers 8、max_parallel_workers_per_gather 2、JIT 开启 |
-| 持久性 | fsync、full_page_writes、synchronous_commit 均开启 |
-| 应用 | release 构建；核心轮次重启应用清空应用缓存，16 个查询客户端并发 |
-| 其他轮次 | 剩余只读接口 4 并发；合成配置写入、鉴权及本机 Provider 场景串行执行 |
-| 缓存边界 | 未清空数据库与宿主文件缓存，应用冷缓存不等于数据库冷缓存 |
-
-造数与离线汇总构建使用的临时资源配置已恢复，搜索索引试验已撤销。3100 万条容量基线不是生产 Worker 全量回填吞吐的测量结果；生产投影的正确性通过真实源表回归及样本对账验证
-
-## HTTP 结果
-
-以下为当前矩阵中的单次短样本，不能推断稳定的 P95／P99。并发样本包含资源竞争，不能与隔离单请求耗时直接当作前后优化对照
-
-| 场景 | 耗时 |
-| --- | ---: |
-| 用量明细宽范围模型前缀搜索 | 32.63 秒 |
-| 账号用量降序 | 20.50 秒 |
-| 账号最近使用降序 | 20.39 秒 |
-| 账号额度预测 | 19.94 秒 |
-| Key 维度诊断 | 18.46 秒 |
-| Dashboard 概览 | 12.30 秒 |
-| 账号维度诊断 | 10.82 秒 |
-| 用量洞察 | 6.20 秒 |
-| 用量汇总 | 6.17 秒 |
-| 用量中间页 | 3.60 秒 |
-| 用量首／尾／超尾页 | 3.46／1.63／1.39 秒 |
-| 错误列表第 6200 页 | 1.70 秒 |
-
-宽范围搜索命中大量明细，32.63 秒超过 30 秒优化期望，但未触发 120 秒保护时限。本次保留该性能局限，不再为此扩展索引或引入其他存储系统
-
-正确性核对覆盖源请求总量、Token、费用、主要筛选组合、诊断分母及分页结果。移除的搜索与分组参数按预期返回 400，权限负例按预期返回拒绝；这些负例不计作合法容量查询通过
-
-## 回归与自审范围
-
-- 超过 10 万条真实源请求经过生产 Worker，验证账号／Key 汇总、热点边界和未知 Token；锁住窄事实表时，干净完整桶仍可读取
-- 待投影跨账号／模型／时间桶修改、删除、重复投递、写入失败回滚及构建前缀持续更新
-- 精确分页首中尾页、超尾页、跨时间段与同时间戳排序，两类错误来源合并
-- 预测微秒边界、同完成时间、截止时间、缺失费用、恢复请求、重建与待投影补差
-- 连接排队计入总预算，超时与取消后的事务回滚及连接归还仍持有查询名额
-- 分组费用沿用当前成员关系；账号窗口的 Token 兜底与普通统计的未知值口径分别保留
-- 投影与极值重算共用维度定义，已删除请求 ID 搜索捷径及淘汰的逐行读取映射
-- Rustfmt、相关 Store／Admin／API 回归、严格 Clippy、架构边界、迁移冻结清单、前端类型构建与 ESLint 对应实际最终工作区执行
-
-页面仅调整搜索提示；按本次任务要求不提供截图
-
-## 升级与验证局限
-
-迁移 0030／0031 增加统计字段和索引，将既有时间桶排入可恢复重建，不清空窄事实或重置已完成的源投影游标。重建中的桶走事实回退，索引创建与桶重建仍需要额外 I/O、磁盘空间和时间；未完成生产环境在线升级的时延与负载验收
-
-本次收尾聚焦 3100 万请求瓶颈，不再扩展外围插件、OAuth、备份和升级场景。数据库冷缓存、长期持续负载、生产 Worker 全量回填性能及未覆盖外部集成保留为验证局限；小规模生命周期回归不等于相同场景的 3100 万级持续写入验收
-
-临时造数、HTTP 驱动、SQL 计划及逐请求记录保存在忽略跟踪的 `.runtime/capacity`，不进入工程提交。完整请求与路由结果为 `http-31m-latest.json`／`http-31m-latest.md`，原失败轮次分别保留，最终检查日志使用 `final-*` 前缀

frontend/env.d.ts

@@ -1,9 +1,5 @@
 /// <reference types="vite/client" />
 
-interface ImportMetaEnv {
-  readonly VITE_API_TIMEOUT_MS?: string
-}
-
 declare module '@fontsource-variable/inter'
 declare module '@fontsource-variable/jetbrains-mono'
 

frontend/src/api/constants.ts

@@ -1,6 +1,2 @@
 export const API_BASE_URL = import.meta.env.DEV ? '/dev' : ''
-const timeout = Number(import.meta.env.VITE_API_TIMEOUT_MS ?? 120_000)
-if (!Number.isSafeInteger(timeout) || timeout <= 0 || timeout > 2_147_483_647) {
-  throw new Error('VITE_API_TIMEOUT_MS must be a positive 32-bit integer')
-}
-export const API_TIMEOUT_MS = timeout
+export const API_TIMEOUT_MS = 120_000

release/notes.md

@@ -1,25 +1,12 @@
-# v3.26.0
-
-## 新增与调整
-
-- 移除「最小请求间隔」及其前后端、调度、数据库和插件设置字段；同一账号不再因上次请求开始时间而等待，并发、排队、额度与冷却限制继续生效
-- 使用记录的首字耗时和总耗时按固定阈值分档着色，便于识别慢请求，不改变请求成功或失败状态
-- 管理端请求和后端观测操作的默认保护时限调整为 120 秒，分别支持前端构建变量 `VITE_API_TIMEOUT_MS` 与部署配置 `store.pool.observability_timeout_seconds`；模型请求期限仍独立控制
+# v3.26.1
 
 ## 修复与改进
 
-- 优化大规模请求历史下的统计、账号与 Key 用量、精确分页和额度预测，复用数据库聚合与时间分段查询，移除固定十万行拒绝条件；保留精确总数、金额及未知值语义，并及时修正尚未投影的修改和删除
-- 修复同一额度窗口内上游额度使用比例回落时预测累计量被错误重置的问题，保留已确认的历史用量
-- 管理端导入 OpenAI 多账号 JSON 文件时逐账号处理、分别记录结果，自动按条目数和请求体大小分批；单个账号失败不再影响整份文件，已知过期且没有刷新令牌的账号会明确拒绝导入
-- 优化 Codex 隐私规则对长请求的处理，减少重复复制、序列化和上下文遍历；取消固定 100 毫秒墙钟限制，保留大小、执行步数与请求取消检查，以及规则失败时的回滚语义
-- 保留租约丢失、续期超时、升级会话结束和备份清理的真实失败原因；S3 连接测试在探针删除失败时报告失败，文件交换回滚失败时保留主错误及恢复路径
-- 调整隐私策略文案与规则弹窗按钮，优化模型价格表的行高及单位提示
-- 更新 `quinn-proto`，修复相关依赖安全问题
+- OpenAI Responses 的 HTTP/SSE 与 WebSocket 即时转发前导事件和压缩进度，移除等待后续输出的暂存，避免上游已有进度却迟迟无法到达客户端；关联 [#405](https://github.com/zyycn/codex-proxy-rs/issues/405)
+- 未启用中间件链且仅交付可替换前导时，保留已有的有界重试；修复 WebSocket 先收到响应元数据后，连接寿命限制错误未进入恢复流程的问题
+- 管理端 API 请求超时固定为 120 秒，不再读取构建变量 `VITE_API_TIMEOUT_MS`
 
-## 升级事项
+## 使用说明
 
-- 新增 `0030_analytics_and_scheduling.sql`，创建统计索引、安排时间桶重建并删除请求间隔字段；保留已有统计事实和已完成的历史投影进度。升级前备份数据库，大数据量下索引创建会增加启动时间、磁盘占用与 I/O，迁移完成后服务才开始监听
-- 已发布的 `0001–0029` 迁移保持不变，可从 `v3.25.0` 正常升级。若源码部署已执行本轮 `main` 中合并前的 `0030–0032`，不能直接沿用该迁移历史升级，需单独处理开发库数据；不要修改 `_sqlx_migrations` 的校验值绕过检查
-- 管理 API 和插件设置不再提供 `requestIntervalMs`／`request_interval_ms`，依赖该字段的自定义调用方需同步调整
-- 使用记录搜索仅按密钥、账号和模型做前缀匹配，密钥与账号名称使用历史快照；请求 ID 和上游 ID 改用对应精确筛选参数。搜索不再改变顶部统计范围，汇总、洞察和诊断不接受搜索参数，管理观测接口不再接受路由分组筛选
-- 分组费用按当前组成员的请求费用汇总，调整成员关系后，留存窗口内的费用随当前成员重新计算
+- 本版本不调整既有心跳频率和默认空闲超时；SSE 注释与 WebSocket Ping/Pong 不延长官方 Codex 的事件等待期限，上游长时间没有新事件时仍可能超时
+- 正文、工具活动、压缩进度或未知事件交付后，网关不再透明重放；中间件链不启用可替换前导的重试例外

release/version.yaml

@@ -1 +1 @@
-version: "3.26.0"
+version: "3.26.1"