diff --git a/backend/crates/providers/openai/src/provider/failure.rs b/backend/crates/providers/openai/src/provider/failure.rs index 2fb6226ab..dcb622a3d 100644 --- a/backend/crates/providers/openai/src/provider/failure.rs +++ b/backend/crates/providers/openai/src/provider/failure.rs @@ -84,10 +84,8 @@ pub(super) async fn wait_for_replay_grace(deadline: Option) { /// 提交边界前的上游事件预取 /// -/// 原始 chunk 计数而不是重编码后的 event 大小 -/// 时间与字节阈值共同限定无感换号 -/// 窗口；任一边界到达都会提交已缓存 wire，不能因网关私有资源规则伪造上游协议 -/// 失败 +/// 结构事件保留到宽限期结束、语义输出、终态或 EOF，再提交已缓存 wire +/// 大段配置回显不应提前结束无感换号窗口，原始 chunk 字节数只用于诊断 /// 一旦提交，后续事件不再具备无痕重放资格 pub(super) struct PreCommitClientEvents { pending: Vec, @@ -102,7 +100,6 @@ pub(super) struct PreCommitClientEvents { pub(super) enum PreCommitReleaseReason { SemanticOutput, Terminal, - ByteLimit, GraceTimeout, Eof, } @@ -141,9 +138,6 @@ impl PreCommitClientEvents { if completed { return self.commit_pending(PreCommitReleaseReason::Terminal); } - if self.prefetched_bytes > MAX_STREAM_PREFETCH_BYTES { - return self.commit_pending(PreCommitReleaseReason::ByteLimit); - } if starts_replay_grace && self.replay_grace_started_at.is_none() { self.replay_grace_started_at = Some(Instant::now()); } diff --git a/backend/crates/providers/openai/src/provider/mod.rs b/backend/crates/providers/openai/src/provider/mod.rs index 63e0c3b2c..ddde47be2 100644 --- a/backend/crates/providers/openai/src/provider/mod.rs +++ b/backend/crates/providers/openai/src/provider/mod.rs @@ -125,10 +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; -/// 提交边界前预取 128 KiB 原始上游 chunk；容纳携带配置回显的前导事件， -/// 超过阈值后结束无感换号窗口（最后一个 chunk 可越过阈值）， -/// 但不会把上游数据改写成协议失败 -const MAX_STREAM_PREFETCH_BYTES: usize = 128 * 1024; /// 短暂保留 response.created 等结构事件，让随后到达的明确拒绝可以无感换号； /// 到期即放行，避免模型长时间思考时让客户端一直收不到首事件 const STREAM_REPLAY_GRACE: Duration = Duration::from_millis(2_500); diff --git a/backend/crates/providers/openai/tests/provider/contract/mod.rs b/backend/crates/providers/openai/tests/provider/contract/mod.rs index 39b332598..f070bf76f 100644 --- a/backend/crates/providers/openai/tests/provider/contract/mod.rs +++ b/backend/crates/providers/openai/tests/provider/contract/mod.rs @@ -9578,7 +9578,7 @@ async fn exact_websocket_busy_then_replay_scope_relaxation_is_rejected_before_se } #[tokio::test] -async fn continuation_prefetch_over_128_kib_should_commit_wire_without_protocol_failure() { +async fn continuation_prefetch_over_128_kib_should_wait_for_grace_without_protocol_failure() { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_prefetch_limit").await; let padding = "x".repeat(128 * 1024); @@ -9596,7 +9596,7 @@ async fn continuation_prefetch_over_128_kib_should_commit_wire_without_protocol_ ); assert!(body.len() > 128 * 1024); let (base_url, release, _first_chunk_sent, server) = - paused_chunked_sse_server(body, String::new()).await; + paused_chunked_sse_server(body.clone(), String::new()).await; let mut stream = provider_with_base_url(&store, base_url) .execute( planned_request("openai", http_generate_operation()), @@ -9605,25 +9605,52 @@ async fn continuation_prefetch_over_128_kib_should_commit_wire_without_protocol_ ) .await .expect("prepare provider stream"); - let visible = loop { - let event = timeout(Duration::from_secs(1), stream.next()) + 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 during the grace period" + ); + timeout(Duration::from_secs(3), next_visible) .await - .expect("prefetch threshold must release buffered wire") - .expect("provider stream remains open") - .expect("threshold cannot create a protocol failure"); - if event.has_client_event() { - break event; - } + .expect("grace expiry must release buffered wire") }; assert_eq!( visible.wire_event().and_then(|wire| wire.event_type()), Some("response.created") ); + let mut wire = visible + .wire_event() + .unwrap() + .raw_sse_frame() + .unwrap() + .to_vec(); release.send(()).expect("finish upstream response"); while let Some(event) = stream.next().await { - event.expect("clean upstream EOF cannot become a protocol failure"); + if let Some(frame) = event + .expect("clean upstream EOF cannot become a protocol failure") + .wire_event() + .and_then(|wire| wire.raw_sse_frame()) + { + wire.extend_from_slice(frame); + } } + assert_eq!(wire, body.as_bytes()); server.await.expect("chunked SSE server"); } diff --git a/backend/crates/providers/openai/tests/provider/contract/precommit.rs b/backend/crates/providers/openai/tests/provider/contract/precommit.rs index b7c102a1d..1e1f86852 100644 --- a/backend/crates/providers/openai/tests/provider/contract/precommit.rs +++ b/backend/crates/providers/openai/tests/provider/contract/precommit.rs @@ -79,14 +79,13 @@ async fn stream_failure(stream: &mut ProviderStream) -> (Vec, Pro } #[tokio::test] -async fn large_structural_events_preserve_overload_replay_past_the_old_grace_on_http_and_websocket() -{ +async fn large_structural_events_preserve_overload_replay_on_http_and_websocket() { for websocket in [false, true] { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; - // #259 仅保留了事件长度；用合成配置回显复现尺寸，不依赖现场私有正文 - let created = structural_event("response.created", 38_781); - let progress = structural_event("response.in_progress", 38_785); + // 大段配置回显不应提前结束重试窗口；使用合成正文覆盖单帧与累计的大体积前导事件 + 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 (base_url, release, server) = if websocket { @@ -160,8 +159,8 @@ async fn large_structural_events_preserve_overload_replay_past_the_old_grace_on_ async fn later_structural_events_do_not_extend_grace_and_late_overload_is_not_replayable() { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; - let created = sse(&structural_event("response.created", 38_781)); - let progress = sse(&structural_event("response.in_progress", 38_785)); + 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()); @@ -220,21 +219,20 @@ async fn immediate_release_preserves_wire_and_records_the_boundary_once() { for (reason, body) in [ ( "semantic_output", - sse(&structural_event("response.created", 512)) + &sse(&semantic), + sse(&structural_event("response.created", 300 * 1024)) + &sse(&semantic), ), ( "semantic_output", - sse(&structural_event("response.created", 512)) + &sse(&tool), + sse(&structural_event("response.created", 300 * 1024)) + &sse(&tool), ), ( "terminal", - sse(&structural_event("response.created", 512)) + &sse(&terminal), + sse(&structural_event("response.created", 300 * 1024)) + &sse(&terminal), ), ( - "byte_limit", - sse(&structural_event("response.created", 129 * 1024)), + "eof", + sse(&structural_event("response.created", 300 * 1024)), ), - ("eof", sse(&structural_event("response.created", 512))), ] { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; @@ -274,11 +272,7 @@ async fn immediate_release_preserves_wire_and_records_the_boundary_once() { let releases = releases(&trace); assert_eq!(releases.len(), 1); assert_eq!(releases[0]["data"]["reason"], reason); - if reason == "byte_limit" { - assert!(releases[0]["data"]["prefetchedBytes"].as_u64().unwrap() > 128 * 1024); - } else { - assert_eq!(releases[0]["data"]["prefetchedBytes"], body.len()); - } + assert_eq!(releases[0]["data"]["prefetchedBytes"], body.len()); server.await.unwrap(); } } @@ -288,7 +282,7 @@ async fn cancelling_buffered_structural_events_does_not_release_or_offer_replay( let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; let (base_url, release, _, server) = paused_chunked_sse_server( - sse(&structural_event("response.created", 38_781)), + sse(&structural_event("response.created", 300 * 1024)), String::new(), ) .await; diff --git a/docs/api.md b/docs/api.md index 0ae4e512c..e0a999a64 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1855,7 +1855,8 @@ WebSocket 握手默认请求专项计时，客户端显式提供的开关保留 请求与错误事件各自的状态、attempt、时间线阶段、计时和上述失败分类。 `provider.precommit.released` 事件的 `precommitRelease` 保留 `reason / prefetchedBytes / waitMs`， 分别表示 Provider 释放缓存的原因、累计预取字节数和等待毫秒数，不等同于下游已提交。 -原因只允许 `semantic_output / terminal / byte_limit / grace_timeout / eof`，数值只允许非负安全整数， +当前释放原因为 `semantic_output / terminal / grace_timeout / eof`，导出也接受历史记录中的 `byte_limit`。 +累计字节数只用于诊断，不触发缓存释放；数值只允许非负安全整数， 缺失或未通过校验的字段为 `null`；其他事件的 `precommitRelease` 为 `null` 诊断包不自动导出 message/raw error、任意 metadata、其他 trace event data、请求响应正文和头部。