From 57e284d7d1a72090fbb5c463061162412194073d Mon Sep 17 00:00:00 2001 From: Zyy <39424122+zyycn@users.noreply.github.com> Date: Sat, 10 Oct 2026 16:30:26 +0800 Subject: [PATCH] fix(openai): preserve stream recovery before semantic output (#380) Co-authored-by: Zyy Co-authored-by: Codex --- .../gateway-core/src/engine/coordinator.rs | 8 +- backend/crates/gateway-core/src/error/mod.rs | 6 +- .../gateway-core/tests/engine/coordinator.rs | 28 ++- .../openai/src/provider/execution.rs | 20 +- .../providers/openai/src/provider/failure.rs | 80 ++----- .../providers/openai/src/provider/mod.rs | 3 - .../openai/tests/provider/contract/mod.rs | 103 +++++---- .../tests/provider/contract/precommit.rs | 210 +++++++++++++++--- docs/api.md | 17 +- docs/architecture.md | 6 +- 10 files changed, 313 insertions(+), 168 deletions(-) diff --git a/backend/crates/gateway-core/src/engine/coordinator.rs b/backend/crates/gateway-core/src/engine/coordinator.rs index 5f8c8b64e..4ec577695 100644 --- a/backend/crates/gateway-core/src/engine/coordinator.rs +++ b/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, diff --git a/backend/crates/gateway-core/src/error/mod.rs b/backend/crates/gateway-core/src/error/mod.rs index 2455a26e7..6f1f3c0fe 100644 --- a/backend/crates/gateway-core/src/error/mod.rs +++ b/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, diff --git a/backend/crates/gateway-core/tests/engine/coordinator.rs b/backend/crates/gateway-core/tests/engine/coordinator.rs index 7b41085e4..c0ee329c8 100644 --- a/backend/crates/gateway-core/tests/engine/coordinator.rs +++ b/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] diff --git a/backend/crates/providers/openai/src/provider/execution.rs b/backend/crates/providers/openai/src/provider/execution.rs index 2675e9777..6e996957b 100644 --- a/backend/crates/providers/openai/src/provider/execution.rs +++ b/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, diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 1b83c312c..f18f16062 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -69,28 +69,14 @@ impl MappedProviderFailure { } } -pub(super) enum PreCommitPoll { - Upstream(T), - GraceElapsed, -} - -pub(super) async fn wait_for_replay_grace(deadline: Option) { - 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, prefetched_bytes: usize, - replay_grace_started_at: Option, + 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) -> Vec { 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 { - 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 { // 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" ); diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 48e1a7154..ed3b2241b 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/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"; diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index dc0152e16..e001b07b6 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/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::>().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) ); diff --git a/backend/crates/providers/openai/tests/provider/contract/precommit.rs b/backend/crates/providers/openai/tests/provider/contract/precommit.rs index 1e1f86852..2ddd59dbb 100644 --- a/backend/crates/providers/openai/tests/provider/contract/precommit.rs +++ b/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::>(), + ["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(); + } +} diff --git a/docs/api.md b/docs/api.md index 420dbccd4..1b68b9cb8 100644 --- a/docs/api.md +++ b/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` 不代表没有发生错误。 diff --git a/docs/architecture.md b/docs/architecture.md index ddb1db30e..fdf4289af 100644 --- a/docs/architecture.md +++ b/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 隔离账号绑定， 不将缓存路由键或请求内容推断为账号身份。账号亲和模式随请求策略冻结：