fix(openai): preserve stream recovery before semantic output (#380)

Co-authored-by: Zyy <zyycn@users.noreply.github.com>
Co-authored-by: Codex <noreply@openai.com>

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

@@ -1511,6 +1511,8 @@ where
                 error.pre_delivery_retry(),
                 Some(crate::error::PreDeliveryRetry::SameAccountConnectionRetry { .. })
             );
+        // Provider 的有界同账号传输恢复是尽力重试策略，不要求把 Sent/Ambiguous
+        // 改写成安全证明；外部副作用、取消、预算与下游提交仍由 Core 拦截
         let transport_recovery = match error.pre_delivery_retry() {
             Some(crate::error::PreDeliveryRetry::SameAccountConnectionRetry { transport })
                 if !execution_effect_observed
@@ -1534,8 +1536,7 @@ where
             }) if !execution_effect_observed
                 && !self.connection_budget.exhausted()
                 && self.downstream_committed_at.is_none()
-                && !self.delivery_pending
-                && attempt_send_state != UpstreamSendState::Ambiguous =>
+                && !self.delivery_pending =>
             {
                 Some((AttemptTransport::Retry(retry_index), delay))
             }
@@ -1543,8 +1544,7 @@ where
                 if !execution_effect_observed
                     && !self.connection_budget.exhausted()
                     && self.downstream_committed_at.is_none()
-                    && !self.delivery_pending
-                    && attempt_send_state != UpstreamSendState::Ambiguous =>
+                    && !self.delivery_pending =>
             {
                 Some((
                     AttemptTransport::Fallback,

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

@@ -135,6 +135,7 @@ pub enum PreDeliveryRetry {
         max_delay: Duration,
     },
     /// 固定本次账号，并按 Provider 给出的序号重试当前传输
+    /// 可用于发送结果不明确时的有界尽力恢复，不等于上游未执行的证明
     SameAccountTransportRetry {
         /// Provider-owned 传输重试序号，从 1 开始
         retry_index: NonZeroU32,
@@ -630,8 +631,8 @@ impl ProviderError {
 
     /// 允许 Core 在首个客户端事件前固定原账号并改用 Provider 备用传输
     ///
-    /// 与普通换号恢复相同，Core 仍会拒绝 continuation、指定账号、预算耗尽或
-    /// 下游已提交后的隐藏重放
+    /// Provider 必须确认备用传输满足当前 continuation scope 并限制恢复次数
+    /// Core 保持原账号，拒绝预算耗尽、外部副作用或下游已提交后的隐藏重放
     #[must_use]
     pub fn with_pre_delivery_transport_fallback(mut self) -> Self {
         self.set_pre_delivery_transport_fallback();
@@ -647,6 +648,7 @@ impl ProviderError {
     ///
     /// 重试预算和退避策略由 Provider 所属协议定义；Core 只负责同账号钉选、
     /// deadline/取消边界以及下游 commit barrier
+    /// 该意图允许发送结果不明确时尽力恢复，但不修改发送事实或安全重放证明
     #[must_use]
     pub fn with_pre_delivery_transport_retry(
         mut self,

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

@@ -3493,7 +3493,7 @@ fn ambiguous_pre_delivery_retry_marker_does_not_rotate_account() {
 }
 
 #[test]
-fn ambiguous_transport_retry_is_rejected_even_with_provider_session_state() {
+fn ambiguous_transport_retry_keeps_the_account_without_claiming_replay_safety() {
     let operation = generate_operation();
     let route_plan = plan(&operation);
     let checkpoint = ProviderSessionState::new(
@@ -3513,7 +3513,6 @@ fn ambiguous_transport_retry_is_rejected_even_with_provider_session_state() {
                 Ok(checkpoint_event),
                 Err(
                     ProviderError::new(ProviderErrorKind::Transport, UpstreamSendState::Ambiguous)
-                        .with_replay_safe()
                         .with_pre_delivery_transport_retry(retry_index, Duration::ZERO),
                 ),
             ],
@@ -3533,14 +3532,25 @@ fn ambiguous_transport_retry_is_rejected_even_with_provider_session_state() {
         CancellationToken::new(),
     ))
     .expect("start execution");
-    let error =
-        block_on(session.collect_uncommitted()).expect_err("ambiguous send cannot be replayed");
-    assert!(matches!(error, EngineError::Provider(_)));
-    assert_eq!(provider.contexts.lock().expect("contexts lock").len(), 1);
+    block_on(session.collect_uncommitted()).expect("bounded transport recovery succeeds");
+    block_on(session.commit_downstream(Some(200))).expect("commit winning response");
+    let contexts = provider.contexts.lock().expect("contexts lock");
+    assert_eq!(contexts.len(), 2);
+    assert_eq!(
+        contexts[1].transport(),
+        AttemptTransport::Retry(retry_index)
+    );
+    assert_eq!(
+        contexts[1].required_account(),
+        Some(&ProviderAccountId::new("acct_first").unwrap())
+    );
+    assert!(contexts[1].excluded_accounts().is_empty());
     let state = store.state.lock().expect("store lock");
-    assert_eq!(state.intermediate_failures, 0);
-    assert_eq!(state.finalizations[0].attempt_count, 1);
-    assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Failed);
+    assert_eq!(state.intermediate_failures, 1);
+    assert_eq!(state.finalizations[0].attempt_count, 2);
+    assert_eq!(state.finalizations[0].outcome, ExecutionOutcome::Succeeded);
+    assert!(state.send_states.contains(&UpstreamSendState::Ambiguous));
+    assert_eq!(state.finalizations[0].send_state, UpstreamSendState::Sent);
 }
 
 #[test]

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

@@ -850,6 +850,7 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                         &mut failure,
                         WebSocketRecoveryContext {
                             policy,
+                            replay_boundary: ReplayBoundary::BeforeSemanticOutput,
                             requirement: request_transport_requirement,
                             retry_count: websocket_retry_count,
                             max_retries: stream_max_retries,
@@ -964,7 +965,6 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                 Err(provider_error(ProviderErrorKind::Timeout, UpstreamSendState::Sent))?;
                 return;
             };
-            let replay_grace_deadline = pre_commit_events.replay_grace_deadline();
             let next = tokio::select! {
                 biased;
                 _ = cancellation.cancelled() => Err(MappedProviderFailure::plain(provider_error(
@@ -975,9 +975,8 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                     ProviderErrorKind::Timeout,
                     UpstreamSendState::Sent,
                 ))),
-                _ = wait_for_replay_grace(replay_grace_deadline) => Ok(PreCommitPoll::GraceElapsed),
                 chunk = body.next() => match chunk {
-                    Some(Ok(chunk)) => Ok(PreCommitPoll::Upstream(Some(chunk))),
+                    Some(Ok(chunk)) => Ok(Some(chunk)),
                     Some(Err(error)) => {
                         log_client_upstream_error(
                             UpstreamErrorLogContext::new(
@@ -989,17 +988,11 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                         );
                         Err(map_stream_error(error))
                     }
-                    None => Ok(PreCommitPoll::Upstream(None)),
+                    None => Ok(None),
                 },
             };
             let next = match next {
-                Ok(PreCommitPoll::Upstream(next)) => next,
-                Ok(PreCommitPoll::GraceElapsed) => {
-                    for event in pre_commit_events.commit_pending(PreCommitReleaseReason::GraceTimeout) {
-                        yield event;
-                    }
-                    continue;
-                }
+                Ok(next) => next,
                 Err(mut failure) => {
                     let updates = take_rate_limit_updates(rate_limit_updates.as_ref()).await;
                     let rate_limits_changed = if updates.is_empty() {
@@ -1023,15 +1016,12 @@ pub(super) fn cold_response_stream(response: ColdResponse) -> EventStream {
                     };
                     if failure.websocket_transport_retryable
                         && response_transport == CodexBackendTransport::WebSocket
-                        && (matches!(
-                            failure.error.send_state(),
-                            UpstreamSendState::Sent | UpstreamSendState::Ambiguous
-                        ) || !pre_commit_events.is_committed())
                     {
                         apply_websocket_recovery_policy(
                             &mut failure,
                             WebSocketRecoveryContext {
                                 policy: WebSocketFailurePolicy::Budgeted,
+                                replay_boundary: ReplayBoundary::from_semantic_output(pre_commit_events.is_committed()),
                                 requirement: request_transport_requirement,
                                 retry_count: websocket_retry_count,
                                 max_retries: stream_max_retries,

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

@@ -69,28 +69,14 @@ impl MappedProviderFailure {
     }
 }
 
-pub(super) enum PreCommitPoll<T> {
-    Upstream(T),
-    GraceElapsed,
-}
-
-pub(super) async fn wait_for_replay_grace(deadline: Option<Instant>) {
-    if let Some(deadline) = deadline {
-        tokio::time::sleep_until(tokio::time::Instant::from_std(deadline)).await;
-    } else {
-        std::future::pending::<()>().await;
-    }
-}
-
 /// 提交边界前的上游事件预取
 ///
-/// 结构事件保留到宽限期结束、语义输出、终态或 EOF，再提交已缓存 wire
-/// 大段配置回显不应提前结束无感换号窗口，原始 chunk 字节数只用于诊断
+/// 结构事件保留到语义输出、终态或 EOF，再提交已缓存 wire
 /// 一旦提交，后续事件不再具备无痕重放资格
 pub(super) struct PreCommitClientEvents {
     pending: Vec<ProviderEvent>,
     prefetched_bytes: usize,
-    replay_grace_started_at: Option<Instant>,
+    started_at: tokio::time::Instant,
     committed: bool,
     trace: TraceContext,
 }
@@ -100,16 +86,15 @@ pub(super) struct PreCommitClientEvents {
 pub(super) enum PreCommitReleaseReason {
     SemanticOutput,
     Terminal,
-    GraceTimeout,
     Eof,
 }
 
 impl PreCommitClientEvents {
-    pub(super) const fn new(trace: TraceContext) -> Self {
+    pub(super) fn new(trace: TraceContext) -> Self {
         Self {
             pending: Vec::new(),
             prefetched_bytes: 0,
-            replay_grace_started_at: None,
+            started_at: tokio::time::Instant::now(),
             committed: false,
             trace,
         }
@@ -130,7 +115,6 @@ impl PreCommitClientEvents {
         if self.committed {
             return incoming;
         }
-        let starts_replay_grace = incoming.iter().any(ProviderEvent::has_client_event);
         self.pending.extend(incoming);
         if timing_signals.semantic_output {
             return self.commit_pending(PreCommitReleaseReason::SemanticOutput);
@@ -138,9 +122,6 @@ impl PreCommitClientEvents {
         if completed {
             return self.commit_pending(PreCommitReleaseReason::Terminal);
         }
-        if starts_replay_grace && self.replay_grace_started_at.is_none() {
-            self.replay_grace_started_at = Some(Instant::now());
-        }
         Vec::new()
     }
 
@@ -170,15 +151,9 @@ impl PreCommitClientEvents {
     pub(super) fn take_for_failure(&mut self, incoming: Vec<ProviderEvent>) -> Vec<ProviderEvent> {
         self.pending.extend(incoming);
         self.prefetched_bytes = 0;
-        self.replay_grace_started_at = None;
         std::mem::take(&mut self.pending)
     }
 
-    pub(super) fn replay_grace_deadline(&self) -> Option<Instant> {
-        self.replay_grace_started_at
-            .and_then(|started| started.checked_add(STREAM_REPLAY_GRACE))
-    }
-
     pub(super) const fn is_committed(&self) -> bool {
         self.committed
     }
@@ -186,15 +161,17 @@ impl PreCommitClientEvents {
     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.replay_grace_started_at.map_or(0, |started| started.elapsed().as_millis()),
-            }));
+            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;
-        self.replay_grace_started_at = None;
         std::mem::take(&mut self.pending)
     }
 }
@@ -484,6 +461,7 @@ pub(super) enum WebSocketFailurePolicy {
 
 pub(super) struct WebSocketRecoveryContext<'a> {
     pub(super) policy: WebSocketFailurePolicy,
+    pub(super) replay_boundary: ReplayBoundary,
     pub(super) requirement: TransportRequirement,
     pub(super) retry_count: u32,
     pub(super) max_retries: u32,
@@ -576,34 +554,10 @@ pub(super) fn apply_websocket_recovery_policy(
                 .session_transport_recovery
                 .record_websocket_failure(key, context.max_retries)
         });
-    let send_state = failure.error.send_state();
-    let post_send = matches!(
-        send_state,
-        UpstreamSendState::Sent | UpstreamSendState::Ambiguous
-    );
-    if send_state == UpstreamSendState::Ambiguous || (post_send && !failure.error.replay_is_safe())
-    {
-        tracing::warn!(
-            request_id = context.request_id,
-            attempt_index = context.attempt_index,
-            account_id = context.account_id,
-            websocket_failure_kind = failure.error.kind().as_str(),
-            websocket_failure_code = failure
-                .error
-                .diagnostic()
-                .and_then(ProviderDiagnostic::code)
-                .unwrap_or(""),
-            upstream_send_state = ?send_state,
-            transport_requirement = context.requirement.as_str(),
-            continuation_recovery_action = "client_replay_required",
-            session_http_fallback = session_budget_exhausted,
-            session_affinity_present = context.session_transport_key.is_some(),
-            session_transport_key_hash = context.session_transport_key_hash.unwrap_or(""),
-            "OpenAI upstream WebSocket failed after payload send; proxy replay was suppressed"
-        );
+    // 已交付响应只影响后续请求的会话传输选择，不能重放当前请求
+    if matches!(context.replay_boundary, ReplayBoundary::AfterSemanticOutput) {
         return;
     }
-
     if matches!(
         context.requirement,
         TransportRequirement::ExactWebSocketContinuation | TransportRequirement::HttpRequired
@@ -611,6 +565,8 @@ pub(super) fn apply_websocket_recovery_policy(
         return;
     }
 
+    // 有效输出前，已发送或结果不明也允许按预算尽力恢复
+    // 保留发送事实和 replay_is_safe 原值，重试意图不能伪装成上游未执行的证明
     // 传输恢复必须保持原账号可调度；最终 HTTP attempt 若仍失败，再按真实 HTTP
     // 结果更新账号健康度，避免中间 WS 错误把同账号钉选提前冷却掉
     failure.account_failure = None;
@@ -643,6 +599,8 @@ pub(super) fn apply_websocket_recovery_policy(
             websocket_retry_count = retry_index.get(),
             websocket_max_retries = context.max_retries,
             websocket_retry_delay_ms = u64::try_from(delay.as_millis()).unwrap_or(u64::MAX),
+            upstream_send_state = ?failure.error.send_state(),
+            replay_proved_safe = failure.error.replay_is_safe(),
             transport_requirement = context.requirement.as_str(),
             "OpenAI upstream WebSocket failed; retrying the same account"
         );

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

@@ -125,9 +125,6 @@ const WEBSOCKET_TRANSPORT: &str = "websocket";
 // 在已观测到的 Codex OAuth 上游 16 MiB 附近消息边界前留出传输 metadata 余量
 const WEBSOCKET_HTTP_FALLBACK_THRESHOLD_BYTES: usize = 15 * 1024 * 1024;
 const MAX_COOKIE_HEADER_BYTES: usize = 16 * 1024;
-/// 短暂保留 response.created 等结构事件，让随后到达的明确拒绝可以无感换号；
-/// 到期即放行，避免模型长时间思考时让客户端一直收不到首事件
-const STREAM_REPLAY_GRACE: Duration = Duration::from_millis(2_500);
 // 额度拒绝后的 usage 补查自身执行时长上限；结算等待的随机延迟见 jitter 模块
 const QUOTA_FAILURE_REFRESH_TIMEOUT: Duration = Duration::from_secs(5);
 pub const OFFICIAL_CODEX_BASE_PATH: &str = "/backend-api";

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

@@ -4220,7 +4220,9 @@ async fn abrupt_websocket_disconnect_preserves_diagnosis_and_ambiguous_send_stat
     assert_eq!(error.kind(), ProviderErrorKind::Transport);
     assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
     assert!(!error.replay_is_safe());
-    assert_eq!(error.pre_delivery_retry(), None);
+    assert!(
+        matches!(error.pre_delivery_retry(), Some(PreDeliveryRetry::SameAccountTransportRetry { retry_index, .. }) if retry_index.get() == 1)
+    );
     assert_eq!(
         error.diagnostic().map(|diagnostic| diagnostic.as_str()),
         Some(
@@ -4729,7 +4731,9 @@ async fn ambiguous_websocket_close_reconnects_and_retains_native_continuation()
 
     assert!(saw_websocket_observation);
     assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
-    assert_eq!(error.pre_delivery_retry(), None);
+    assert!(
+        matches!(error.pre_delivery_retry(), Some(PreDeliveryRetry::SameAccountTransportRetry { retry_index, .. }) if retry_index.get() == 1)
+    );
     assert!(error.upstream_code().is_none());
     let close: Value = serde_json::from_str(error.raw_upstream_error().unwrap().as_str()).unwrap();
     assert_eq!(close["type"], "websocket.close");
@@ -4945,7 +4949,17 @@ async fn repeated_websocket_failures_exhaust_budget_then_use_http_and_report_htt
             }
         };
         assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
-        assert_eq!(error.pre_delivery_retry(), None);
+        if index < 2 {
+            assert!(matches!(
+                error.pre_delivery_retry(),
+                Some(PreDeliveryRetry::SameAccountTransportRetry { .. })
+            ));
+        } else {
+            assert_eq!(
+                error.pre_delivery_retry(),
+                Some(PreDeliveryRetry::SameAccountTransportFallback)
+            );
+        }
         assert!(!error.replay_is_safe());
     }
     for index in 0..2 {
@@ -5950,7 +5964,7 @@ async fn websocket_turn_state_metadata_is_exposed_through_response_observation()
 }
 
 #[tokio::test]
-async fn websocket_turn_state_metadata_close_does_not_authorize_replay() {
+async fn websocket_metadata_close_requests_bounded_retry_without_claiming_replay_safety() {
     const ACCOUNT_ID: &str = "acct_websocket_metadata_close";
     const TURN_STATE: &str = "turn-state-before-close";
 
@@ -6032,7 +6046,9 @@ async fn websocket_turn_state_metadata_close_does_not_authorize_replay() {
     assert!(session_update.is_none());
     assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
     assert!(!error.replay_is_safe());
-    assert_eq!(error.pre_delivery_retry(), None);
+    assert!(
+        matches!(error.pre_delivery_retry(), Some(PreDeliveryRetry::SameAccountTransportRetry { retry_index, .. }) if retry_index.get() == 1)
+    );
     assert!(error.upstream_code().is_none());
     let close: Value = serde_json::from_str(error.raw_upstream_error().unwrap().as_str()).unwrap();
     assert_eq!(close["type"], "websocket.close");
@@ -9059,7 +9075,7 @@ async fn ordinary_request_should_hold_created_until_later_failure_can_rotate() {
 }
 
 #[tokio::test]
-async fn ordinary_request_should_bound_structural_event_replay_grace() {
+async fn ordinary_request_should_hold_structural_events_until_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(
@@ -9079,29 +9095,24 @@ async fn ordinary_request_should_bound_structural_event_replay_grace() {
         .await
         .expect("prepare provider stream");
 
-    let first_event = timeout(Duration::from_secs(4), async {
-        loop {
-            let event = stream
-                .next()
-                .await
-                .expect("provider stream must stay open")
-                .expect("provider event");
-            if event.has_client_event() {
-                return event;
+    assert!(
+        timeout(Duration::from_secs(3), async {
+            while let Some(event) = stream.next().await {
+                assert!(!event.expect("provider event").has_client_event());
             }
-        }
-    })
-    .await
-    .expect("response.created must be released after the bounded grace period");
-
-    assert_eq!(
-        first_event.wire_event().and_then(|wire| wire.event_type()),
-        Some("response.created")
+        })
+        .await
+        .is_err()
     );
     release.send(()).expect("finish upstream response");
-    while let Some(event) = stream.next().await {
-        event.expect("clean upstream EOF");
-    }
+    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"))
+    }));
     server.await.expect("chunked SSE server");
 }
 
@@ -9593,7 +9604,7 @@ async fn exact_websocket_busy_then_replay_scope_relaxation_is_rejected_before_se
 }
 
 #[tokio::test]
-async fn continuation_prefetch_over_128_kib_should_wait_for_grace_without_protocol_failure() {
+async fn continuation_prefetch_over_128_kib_stays_buffered_until_eof_without_protocol_failure() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_prefetch_limit").await;
     let padding = "x".repeat(128 * 1024);
@@ -9638,11 +9649,17 @@ async fn continuation_prefetch_over_128_kib_should_wait_for_grace_without_protoc
             timeout(Duration::from_secs(1), next_visible.as_mut())
                 .await
                 .is_err(),
-            "large structural events must remain buffered during the grace period"
+            "large structural events must remain buffered before semantic output"
         );
-        timeout(Duration::from_secs(3), next_visible)
+        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("grace expiry must release buffered wire")
+            .expect("EOF releases original wire")
     };
 
     assert_eq!(
@@ -9655,7 +9672,6 @@ async fn continuation_prefetch_over_128_kib_should_wait_for_grace_without_protoc
         .raw_sse_frame()
         .unwrap()
         .to_vec();
-    release.send(()).expect("finish upstream response");
     while let Some(event) = stream.next().await {
         if let Some(frame) = event
             .expect("clean upstream EOF cannot become a protocol failure")
@@ -11671,7 +11687,7 @@ async fn quota_continuation_stream_rejection_only_projects_before_delivery() {
 }
 
 #[tokio::test]
-async fn quota_continuation_after_structural_commit_preserves_original_failure() {
+async fn late_quota_continuation_rejection_requires_client_replay_before_output() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_provider_contract").await;
     let original = json!({"type":"response.failed","response":{"id":"resp_quota","error":{"code":"usage_limit_reached","message":"limit reached"}}});
@@ -11687,16 +11703,15 @@ async fn quota_continuation_after_structural_commit_preserves_original_failure()
         )
         .await
         .unwrap();
-    timeout(Duration::from_secs(3), async {
-        while let Some(event) = stream.next().await {
-            if event.unwrap().has_client_event() {
-                return;
+    assert!(
+        timeout(Duration::from_secs(3), async {
+            while let Some(event) = stream.next().await {
+                assert!(!event.unwrap().has_client_event());
             }
-        }
-        panic!("expected grace commit");
-    })
-    .await
-    .expect("bounded grace");
+        })
+        .await
+        .is_err()
+    );
     release.send(()).unwrap();
     let mut delivered_failure = false;
     let error = loop {
@@ -11710,9 +11725,9 @@ async fn quota_continuation_after_structural_commit_preserves_original_failure()
             None => panic!("expected failure"),
         }
     };
-    assert!(delivered_failure);
-    assert!(!error.replay_is_safe());
-    assert_ne!(
+    assert!(!delivered_failure);
+    assert!(error.replay_is_safe());
+    assert_eq!(
         error.continuation_recovery_disposition(),
         Some(ContinuationRecoveryDisposition::ClientReplayRequired)
     );

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

@@ -1,4 +1,4 @@
-//! 验证结构事件缓冲、交付宽限与取消对重试边界的影响
+//! 验证结构事件缓冲、长时间等待与取消对重试边界的影响
 
 use gateway_core::diagnostics::TraceContext;
 use gateway_core::engine::provider::ProviderStream;
@@ -16,7 +16,7 @@ fn traced_context(trace: &TraceContext, cancellation: CancellationToken) -> Atte
         )
         .with_trace(trace.clone()),
         NonZeroU32::new(1).unwrap(),
-        SystemTime::now() + Duration::from_secs(30),
+        SystemTime::now() + Duration::from_secs(3_600),
         account_policy(),
         AccountAttemptContext::new(BTreeSet::new(), None, None)
             .with_account_scope(contract_account_scope()),
@@ -127,7 +127,7 @@ async fn large_structural_events_preserve_overload_replay_on_http_and_websocket(
             .unwrap();
 
         assert!(
-            timeout(Duration::from_millis(1_500), next_client_event(&mut stream))
+            timeout(Duration::from_millis(3_500), next_client_event(&mut stream))
                 .await
                 .is_err()
         );
@@ -156,12 +156,11 @@ async fn large_structural_events_preserve_overload_replay_on_http_and_websocket(
 }
 
 #[tokio::test]
-async fn later_structural_events_do_not_extend_grace_and_late_overload_is_not_replayable() {
+async fn late_overload_after_multiple_structural_events_remains_replayable() {
     let store = Arc::new(MemoryAccountStore::default());
     create_account(&store, "acct_provider_contract").await;
     let created = sse(&structural_event("response.created", 96 * 1024));
     let progress = sse(&structural_event("response.in_progress", 96 * 1024));
-    let expected_bytes = created.len() + progress.len();
     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();
@@ -184,30 +183,30 @@ async fn later_structural_events_do_not_extend_grace_and_late_overload_is_not_re
         )
         .await
         .unwrap();
-    let first = timeout(Duration::from_millis(3_500), next_client_event(&mut stream))
-        .await
-        .unwrap();
-    assert_eq!(
-        first.wire_event().unwrap().event_type(),
-        Some("response.created")
+    assert!(
+        timeout(Duration::from_millis(3_500), next_client_event(&mut stream))
+            .await
+            .is_err()
     );
     release.send(()).unwrap();
-    let (events, error) = timeout(Duration::from_secs(1), stream_failure(&mut stream))
+    let (events, mut error) = timeout(Duration::from_secs(1), stream_failure(&mut stream))
         .await
         .unwrap();
-    assert!(!error.replay_is_safe());
-    assert!(error.pre_delivery_retry().is_none());
-    assert!(events.iter().any(|event| {
-        event
-            .wire_event()
-            .is_some_and(|wire| wire.data() == &overload())
-    }));
-    let releases = releases(&trace);
-    assert_eq!(releases.len(), 1);
-    assert_eq!(releases[0]["attemptIndex"], 1);
-    assert_eq!(releases[0]["data"]["reason"], "grace_timeout");
-    assert_eq!(releases[0]["data"]["prefetchedBytes"], expected_bytes);
-    assert!((2_500..3_500).contains(&releases[0]["data"]["waitMs"].as_u64().unwrap()));
+    assert!(events.iter().all(|event| !event.has_client_event()));
+    assert!(error.replay_is_safe());
+    assert!(matches!(
+        error.pre_delivery_retry(),
+        Some(PreDeliveryRetry::SameAccountTransientRetry { .. })
+    ));
+    assert_eq!(
+        error
+            .take_atomic_client_events()
+            .iter()
+            .filter_map(|event| event.wire_event()?.event_type())
+            .collect::<Vec<_>>(),
+        ["response.created", "response.in_progress", "error"]
+    );
+    assert!(releases(&trace).is_empty());
     server.await.unwrap();
 }
 
@@ -311,3 +310,164 @@ async fn cancelling_buffered_structural_events_does_not_release_or_offer_replay(
     release.send(()).unwrap();
     server.await.unwrap();
 }
+
+#[tokio::test]
+async fn twenty_minute_structural_progress_preserves_late_overload_recovery() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_provider_contract").await;
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let base_url = format!("http://{}", listener.local_addr().unwrap());
+    let (progress, mut progress_requests) = tokio::sync::mpsc::channel::<()>(1);
+    let server = tokio::spawn(async move {
+        let (mut socket, _) = listener.accept().await.unwrap();
+        read_http_request(&mut socket).await;
+        socket.write_all(b"HTTP/1.1 200 OK\r\ncontent-type: text/event-stream\r\ntransfer-encoding: chunked\r\n\r\n").await.unwrap();
+        write_http_chunk(
+            &mut socket,
+            &sse(&structural_event("response.created", 1024)),
+        )
+        .await;
+        while progress_requests.recv().await.is_some() {
+            write_http_chunk(
+                &mut socket,
+                &sse(&structural_event("response.in_progress", 1024)),
+            )
+            .await;
+        }
+        write_http_chunk(&mut socket, &sse(&overload())).await;
+        socket.write_all(b"0\r\n\r\n").await.unwrap();
+    });
+    let trace = TraceContext::new("req_precommit");
+    let mut stream = provider_with_base_url(&store, base_url)
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            traced_context(&trace, CancellationToken::new()),
+        )
+        .await
+        .unwrap();
+    assert!(
+        timeout(Duration::from_millis(100), next_client_event(&mut stream))
+            .await
+            .is_err()
+    );
+    // 结构进度保持上游传输活跃，长思考不能额外触发首内容等待期限
+    for _ in 0..20 {
+        tokio::time::pause();
+        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()
+        );
+    }
+    drop(progress);
+    let (events, error) = timeout(Duration::from_secs(1), stream_failure(&mut stream))
+        .await
+        .unwrap();
+    assert!(events.iter().all(|event| !event.has_client_event()));
+    assert!(error.replay_is_safe());
+    assert!(matches!(
+        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() {
+    let store = Arc::new(MemoryAccountStore::default());
+    create_account(&store, "acct_provider_contract").await;
+    // 大段配置回显仍属于前导事件，累计体积不应改变上游失败的恢复方式
+    let preamble = sse(&structural_event("response.in_progress", 1024 * 1024)).repeat(17);
+    let (base_url, release, _, server) =
+        paused_chunked_sse_server(preamble, sse(&overload())).await;
+    let trace = TraceContext::new("req_precommit");
+    let mut stream = provider_with_base_url(&store, base_url)
+        .execute(
+            planned_request("openai", http_generate_operation()),
+            traced_context(&trace, CancellationToken::new()),
+        )
+        .await
+        .unwrap();
+    assert!(
+        timeout(Duration::from_secs(1), next_client_event(&mut stream))
+            .await
+            .is_err()
+    );
+    release.send(()).unwrap();
+    let (events, mut error) = timeout(Duration::from_secs(10), stream_failure(&mut stream))
+        .await
+        .unwrap();
+    assert!(events.iter().all(|event| !event.has_client_event()));
+    assert!(error.replay_is_safe());
+    assert!(matches!(
+        error.pre_delivery_retry(),
+        Some(PreDeliveryRetry::SameAccountTransientRetry { .. })
+    ));
+    let buffered_bytes: usize = error
+        .take_atomic_client_events()
+        .iter()
+        .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());
+    server.await.unwrap();
+}
+
+#[tokio::test]
+async fn websocket_disconnect_recovery_honors_request_retry_budget_without_session_affinity() {
+    for max_retries in [0, 5] {
+        let store = Arc::new(MemoryAccountStore::default());
+        create_account(&store, "acct_provider_contract").await;
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let base_url = format!("http://{}", listener.local_addr().unwrap());
+        let server = tokio::spawn(async move {
+            for _ in 0..=max_retries {
+                let (socket, _) = listener.accept().await.unwrap();
+                let mut ws = accept_codex_test_websocket(socket).await;
+                ws.next().await.unwrap().unwrap();
+                ws.send(Message::Text(
+                    structural_event("response.created", 1024)
+                        .to_string()
+                        .into(),
+                ))
+                .await
+                .unwrap();
+                ws.close(None).await.unwrap();
+            }
+        });
+        let provider = provider_with_base_url_and_retry_budget(&store, base_url, max_retries);
+        for retry_count in 0..=max_retries {
+            let transport = NonZeroU32::new(retry_count)
+                .map_or(AttemptTransport::Default, AttemptTransport::Retry);
+            let trace = TraceContext::new("req_precommit");
+            let mut stream = Arc::clone(&provider)
+                .execute(
+                    planned_request("openai", generate_operation()),
+                    traced_context(&trace, CancellationToken::new()).with_transport(transport),
+                )
+                .await
+                .unwrap();
+            let (events, error) = stream_failure(&mut stream).await;
+            assert!(events.iter().all(|event| !event.has_client_event()));
+            assert_eq!(error.send_state(), UpstreamSendState::Ambiguous);
+            assert!(!error.replay_is_safe());
+            if retry_count < max_retries {
+                assert!(
+                    matches!(error.pre_delivery_retry(), Some(PreDeliveryRetry::SameAccountTransportRetry { retry_index, .. }) if retry_index.get() == retry_count + 1)
+                );
+            } else {
+                assert_eq!(
+                    error.pre_delivery_retry(),
+                    Some(PreDeliveryRetry::SameAccountTransportFallback)
+                );
+            }
+            assert!(releases(&trace).is_empty());
+        }
+        server.await.unwrap();
+    }
+}

docs/api.md

@@ -342,7 +342,7 @@ OAuth 的大 connection-local WS 续接，以及 HTTP `store=false` 成功后标
 的 WS 增量，均在发送前返回 `status: 400 / previous_response_not_found`。客户端应清除旧 ID，
 携带完整历史及工具调用／输出重试。官方 Codex 支持重连 WS 后完整重发，也可能按自身重试预算
 切到 HTTP；其他客户端需要自行实现此合同。代理不缓存 transcript，不把增量输入当作独立新链，
-不自动重放发送结果不确定的请求；HTTP 链后续步骤可能增加一次恢复信号、重连及全量上传
+连接内续接失去原连接时仍需客户端恢复；HTTP 链后续步骤可能增加一次恢复信号、重连及全量上传
 
 已建立模型执行的 Responses、Images 和 Search HTTP 响应按以下规则返回关联 ID：
 `x-gateway-request-id` 为模型执行 ID；`x-request-id` 保留有效上游值，只有上游
@@ -435,6 +435,16 @@ xAI 的 `reasoning.effort` 接受 `none / minimal / low / medium / high / xhigh
 
 #### 容量拒绝与重试
 
+OpenAI Responses 的 `response.created`、`response.in_progress` 等前导事件保留到首次有效文本、
+推理、工具活动或终态再交付，SSE 与 WebSocket 使用相同边界。前导缓冲不另设等待时长或累计体积上限，
+等待沿用[请求期限](#请求期限)与传输超时配置。WebSocket 独立 Ping 保活；
+HTTP 在首个业务事件前保留状态码和响应头，SSE 注释保活从提交响应后开始
+
+WebSocket 在交付有效输出前断线或超时，可按 `stream_max_retries` 做同账号、有退避的尽力重试
+（默认最多 5 次），耗尽后仅在请求允许时改用 HTTP。已发送或发送结果不明不单独阻止这项恢复；
+次数限制不保证上游只执行一次，诊断保留原始发送状态。精确连接续接、显式终止拒绝、已观测的
+外部副作用和下游提交仍限制恢复；取消与请求截止时间随时终止等待
+
 OpenAI 明确返回 `server_is_overloaded`、`slow_down` 或模型容量不足错误时，代理在允许安全重放且
 尚未交付输出的前提下，先做最多 3 次同账号重试，再通过现有调度换号。没有有效服务器建议时，
 按 500ms 起步的指数退避等待，单次最多 8 秒；有效 `Retry-After` 优先，接受秒数、HTTP 日期和零延迟，
@@ -1911,9 +1921,8 @@ WebSocket 握手默认请求专项计时，客户端显式提供的开关保留
 请求与错误事件各自的状态、attempt、时间线阶段、计时和上述失败分类。
 `provider.precommit.released` 事件的 `precommitRelease` 保留 `reason / prefetchedBytes / waitMs`，
 分别表示 Provider 释放缓存的原因、累计预取字节数和等待毫秒数，不等同于下游已提交。
-当前释放原因为 `semantic_output / terminal / grace_timeout / eof`，导出也接受历史记录中的 `byte_limit`。
-累计字节数只用于诊断，不触发缓存释放；数值只允许非负安全整数，
-缺失或未通过校验的字段为 `null`；其他事件的 `precommitRelease` 为 `null`
+当前释放原因为 `semantic_output / terminal / eof`，导出也接受历史记录中的 `grace_timeout / byte_limit`。
+数值只允许非负安全整数，缺失或未通过校验的字段为 `null`；其他事件的 `precommitRelease` 为 `null`
 
 诊断包不自动导出 message/raw error、任意 metadata、其他 trace event data、请求响应正文和头部。
 `availability` 与 `omitted` 明示未采集、不完整或主动省略的内容，`null` 不代表没有发生错误。

docs/architecture.md

@@ -381,6 +381,10 @@ Client Key 鉴权完成后，API adapter 从有界请求头识别客户端，Cor
   重试与提交边界。具体错误合同见 [数据面接口](api.md#3-openai-数据面与模型目录)
 - Provider 可将明确容量拒绝标记为有界同账号退避，Core 在既有安全重放边界内执行，按账号维护请求内
   预算，耗尽后复用普通换号路径。该退避消耗总路由预算，与 WS 传输恢复、OAuth 刷新及账号额度冷却分开
+- OpenAI 的结构事件由 Provider 保留到有效输出或终态，不按等待时长或累计体积提前交付。
+  交付前的 WS 传输错误可请求有界同账号尽力恢复，即使发送结果不明也保留此恢复意图；
+  发送事实与安全重放证明不因此改写。Provider 限定传输次数与续接范围，Core 保持原账号并检查
+  取消、deadline、已观测外部副作用及下游交付边界
 - 换号预算从运行设置冻结到请求计划，选中账号与上一 attempt 不同时计数；首次选号、同账号重试和
   选号阶段过滤候选不消耗预算。跨 Provider 候选推进与 continuation 排除重放同样受限，预算耗尽后
   以最后一个原始上游错误终态；总路由尝试仍最多 32 次
@@ -642,7 +646,7 @@ Continuation 受原请求的 Client Key、账号范围、Provider 和发送/交
   已知为 connection-local 的原生续接进入重放阶段时，在选号前返回 `ClientReplayRequired`，不认领或迁移会话绑定。
   正常 native 续接仍由连接池核对原连接及 response ID 是否可用
 - xAI 使用客户端提交的完整历史作为重放输入
-- scope 外账号、跨 Key 复用或不明确发送结果均 fail closed
+- scope 外账号与跨 Key 复用均拒绝；发送结果不明时，仅接受 Provider 明确给出的有界传输恢复意图
 
 OpenAI Provider 解释官方逻辑 `session_id` 与 `thread_id`，按 Client Key 隔离账号绑定，
 不将缓存路由键或请求内容推断为账号身份。账号亲和模式随请求策略冻结：