diff --git a/backend/crates/providers/openai/src/provider/observation.rs b/backend/crates/providers/openai/src/provider/observation.rs index e1f269516..b00e785d9 100644 --- a/backend/crates/providers/openai/src/provider/observation.rs +++ b/backend/crates/providers/openai/src/provider/observation.rs @@ -153,8 +153,8 @@ impl OpenAiResponseObservationState { started_at: Instant, ) -> bool { let mut changed = false; - // 首字采用官方输出项边界,真实推理与正文仍分别计时 - if signals.output_start { + // 首字采用首响应载荷边界,真实推理与正文仍分别计时 + if signals.response_data { changed |= insert_first_timing(&mut self.timings.first_token_ms, started_at); } if signals.reasoning_output { diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 78c1be6ae..bcc225702 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/backend/crates/providers/openai/src/transport/canonical.rs @@ -342,6 +342,14 @@ impl CodexCanonicalDecoder { raw_sse_frame: Option, output: &mut Vec, ) -> Result<(), CodexCanonicalError> { + // 首字只要求响应载荷,不依赖 JSON、事件类型或实际输出内容 + let data = event.data.trim(); + self.timing_signals.response_data |= + if matches!(raw_sse_frame, Some(RawResponseFrame::WebSocket(_))) { + !event.data.is_empty() + } else { + !data.is_empty() && !data.starts_with("[DONE]") + }; if event.data.trim() == "[DONE]" && !matches!(raw_sse_frame, Some(RawResponseFrame::WebSocket(_))) { @@ -371,7 +379,7 @@ impl CodexCanonicalDecoder { .or_else(|| value.get("type").and_then(Value::as_str)); if event_type == Some("codex.rate_limits") { // HTTP 传输已将此控制帧投影为本地额度事实 - // 此帧不能成为客户端输出,也不能启动首个输出计时 + // 此帧只贡献首响应观测,不能成为客户端输出 return Ok(()); } if event_type == Some("responsesapi.websocket_timing") && self.started && !self.completed { @@ -438,7 +446,6 @@ impl CodexCanonicalDecoder { fn merge_timing_signals(&mut self, signals: ResponseEventSignals) { self.timing_signals.protocol_progress |= signals.protocol_progress; - self.timing_signals.output_start |= signals.output_start; self.timing_signals.semantic_output |= signals.semantic_output; self.timing_signals.reasoning_output |= signals.reasoning_output; self.timing_signals.text_output |= signals.text_output; diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index 3a1e4b960..8841170b8 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -222,14 +222,14 @@ pub(crate) fn is_replaceable_response_prelude(event_type: &str, value: &Value) - } } -/// 单个 Responses 事件对计时系统提供的稳定语义信号 +/// Responses 响应载荷与实际内容的观测信号 /// -/// `output_start` 只表示官方输出项开始事件;`semantic_output` 仍要求 +/// `response_data` 表示收到非空响应载荷;`semantic_output` 仍要求 /// 实际内容,供重试与交付边界使用,不能与首字观测互换 #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct ResponseEventSignals { pub protocol_progress: bool, - pub output_start: bool, + pub response_data: bool, pub semantic_output: bool, pub reasoning_output: bool, pub text_output: bool, @@ -313,12 +313,11 @@ impl ResponseTimingMetrics { /// 从已解析的 Responses 事件提取计时语义 /// -/// 首字采用官方 Codex 的首个 output_item.added 边界 +/// 首响应载荷由解码器在 JSON 解析前观测 /// 语义输出仍要求文本、推理、工具参数、图片结果或工具执行 pub fn response_event_signals(event_type: Option<&str>, value: &Value) -> ResponseEventSignals { let mut signals = ResponseEventSignals { protocol_progress: !matches!(event_type, Some("response.failed" | "error")), - output_start: event_type == Some("response.output_item.added"), ..ResponseEventSignals::default() }; match event_type { diff --git a/backend/crates/providers/openai/tests/provider/contract/timing.rs b/backend/crates/providers/openai/tests/provider/contract/timing.rs index 4fff42bfa..fc0348894 100644 --- a/backend/crates/providers/openai/tests/provider/contract/timing.rs +++ b/backend/crates/providers/openai/tests/provider/contract/timing.rs @@ -138,7 +138,6 @@ async fn official_response_duration_is_observed_independently_of_the_request_clo assert_eq!(timings.upstream_response_ms, Some(7_000)); assert_eq!(timings.upstream_api_overhead_ms, Some(120.25)); assert_eq!(timings.upstream_engine_ms, Some(6400.0)); - assert_eq!(timings.first_token_ms, None); assert!(timings.first_event_ms.is_some_and(|value| value >= 20_000)); } } @@ -195,7 +194,6 @@ async fn response_critical_path_is_observed_without_logical_turn_totals_on_both_ assert_eq!(timings.upstream_engine_ms, Some(6710.68507)); assert!((timings.upstream_api_overhead_ms.unwrap() - 81.409594).abs() < 0.000001); assert_eq!(timings.upstream_response_ms, Some(7000)); - assert_eq!(timings.first_token_ms, None); } } @@ -289,12 +287,39 @@ async fn timing_frames_outside_the_active_response_do_not_enter_request_metrics( let timings = finish_timings(&mut stream, ProviderResponseTimings::default()).await; assert_eq!(timings.upstream_api_overhead_ms, None); assert_eq!(timings.upstream_engine_ms, None); - assert_eq!(timings.first_token_ms, None); server.await.unwrap(); } #[tokio::test] -async fn content_without_output_item_start_keeps_ttft_unknown_on_every_attempt() { +async fn sse_keepalive_and_empty_data_do_not_start_ttft() { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let (base_url, release, _, server) = paused_chunked_sse_server( + ": ping\n\nevent: response.created\n\ndata: \t\n\ndata: [DONE]\n\n".to_owned(), + CAPTURE_COMPLETED_SSE.to_owned(), + ) + .await; + let mut stream = provider_with_base_url(&store, base_url) + .execute( + planned_request("openai", http_generate_operation()), + timed_context(Instant::now() - Duration::from_secs(3), 1), + ) + .await + .unwrap(); + let first = first_upstream_timings(&mut stream).await; + assert_eq!(first.first_token_ms, None); + release.send(()).unwrap(); + let final_timings = finish_timings(&mut stream, first).await; + assert!( + final_timings + .first_token_ms + .is_some_and(|value| value >= 3_000) + ); + server.await.unwrap(); +} + +#[tokio::test] +async fn first_response_uses_request_clock_on_every_attempt() { for attempt in [1, 2] { let store = Arc::new(MemoryAccountStore::default()); create_account(&store, "acct_provider_contract").await; @@ -318,7 +343,7 @@ async fn content_without_output_item_start_keeps_ttft_unknown_on_every_attempt() .unwrap(); let first = first_upstream_timings(&mut stream).await; assert!(first.first_event_ms.is_some_and(|value| value >= 3_000)); - assert_eq!(first.first_token_ms, None); + assert!(first.first_token_ms.is_some_and(|value| value >= 3_000)); assert!(first.first_text_ms.is_some_and(|value| value >= 3_000)); release.send(()).unwrap(); let final_timings = finish_timings(&mut stream, first).await; @@ -328,7 +353,7 @@ async fn content_without_output_item_start_keeps_ttft_unknown_on_every_attempt() } #[tokio::test] -async fn structural_frames_start_ttft_before_content_on_http_and_websocket() { +async fn created_response_starts_ttft_before_output_on_http_and_websocket() { for websocket in [false, true] { for (output, has_content) in [ ( @@ -382,13 +407,11 @@ async fn structural_frames_start_ttft_before_content_on_http_and_websocket() { ) .await; ws.next().await.unwrap().unwrap(); - for event in [created, added] { - ws.send(Message::Text(event.to_string().into())) - .await - .unwrap(); - } + ws.send(Message::Text(created.to_string().into())) + .await + .unwrap(); released.await.unwrap(); - for event in [output, completed] { + for event in [added, output, completed] { ws.send(Message::Text(event.to_string().into())) .await .unwrap(); @@ -397,8 +420,8 @@ async fn structural_frames_start_ttft_before_content_on_http_and_websocket() { (base_url, release, server) } else { let (base_url, release, _, server) = paused_chunked_sse_server( - format!("data: {created}\n\ndata: {added}\n\n"), - format!("data: {output}\n\ndata: {completed}\n\n"), + format!("data: {created}\n\n"), + format!("data: {added}\n\ndata: {output}\n\ndata: {completed}\n\n"), ) .await; (base_url, release, server) @@ -423,10 +446,10 @@ async fn structural_frames_start_ttft_before_content_on_http_and_websocket() { return observation.timings(); } } - panic!("missing structural first-token observation"); + panic!("missing first-response observation"); }) .await - .expect("structural first token before releasing content"); + .expect("first response before releasing output"); assert!(first.first_token_ms.is_some_and(|value| value >= 3_000)); assert_eq!(first.first_text_ms, None); assert_eq!(first.first_reasoning_ms, None); diff --git a/backend/crates/providers/openai/tests/transport/canonical.rs b/backend/crates/providers/openai/tests/transport/canonical.rs index 8f7b279dc..b6d80bf44 100644 --- a/backend/crates/providers/openai/tests/transport/canonical.rs +++ b/backend/crates/providers/openai/tests/transport/canonical.rs @@ -141,11 +141,42 @@ fn official_response_duration_does_not_cross_response_identity_or_incomplete_bou } #[test] -fn official_timing_frames_do_not_start_first_output_wait() { +fn first_response_observation_uses_nonempty_data_before_json_or_event_classification() { + for data in [ + r#"{"type":"response.created","response":{"id":"resp_first"}}"#, + r#"{"type":"response.failed","response":{"error":{"message":"failed"}}}"#, + r#"{"type":"future.event"}"#, + "not-json", + ] { + for websocket in [false, true] { + let mut decoder = CodexCanonicalDecoder::new("fallback").with_raw_sse_passthrough(); + if websocket { + let _ = decoder.push_websocket(data); + } else { + let _ = decoder + .push(b": ping\r\nevent: response.created\r\ndata: \t\r\n\r\ndata: [DONE]\n\n"); + assert!(!decoder.take_timing_signals().response_data); + let frame = format!("data: {data}\r\n\r\n"); + let split = frame.len() / 2; + let _ = decoder.push(&frame.as_bytes()[..split]); + assert!(!decoder.take_timing_signals().response_data); + let _ = decoder.push(&frame.as_bytes()[split..]); + } + let signals = decoder.take_timing_signals(); + assert!(signals.response_data, "websocket={websocket}, data={data}"); + assert!(!signals.semantic_output); + assert!(!signals.reasoning_output); + assert!(!signals.text_output); + } + } +} + +#[test] +fn official_timing_frames_start_first_response_without_content() { let mut decoder = CodexCanonicalDecoder::new("gpt-5.4"); let events = decoder.push(b"data: {\"type\":\"responsesapi.websocket_timing\",\"timing_metrics\":{\"total_turn_time_s\":20}}\n\n").expect("timing frame"); assert!(!events.is_empty()); - assert!(!decoder.take_timing_signals().output_start); + assert!(decoder.take_timing_signals().response_data); assert_eq!(decoder.upstream_response_ms(), None); } @@ -169,7 +200,7 @@ fn decoder_should_not_forward_codex_rate_limit_metadata_fixture_as_openai_wire() } #[test] -fn raw_sse_passthrough_should_drop_rate_limit_control_without_timing_side_effects() { +fn raw_sse_passthrough_should_observe_rate_limit_response_without_forwarding_it() { let frame = concat!( "event: codex.rate_limits\n", "data: {\"type\":\"codex.rate_limits\",\"rate_limits\":{\"primary\":{\"used_percent\":90}}}\n\n", @@ -183,7 +214,7 @@ fn raw_sse_passthrough_should_drop_rate_limit_control_without_timing_side_effect assert!(events.is_empty()); assert!(!signals.protocol_progress); - assert!(!signals.output_start); + assert!(signals.response_data); assert!(!signals.semantic_output); } diff --git a/backend/crates/providers/openai/tests/transport/protocol/responses.rs b/backend/crates/providers/openai/tests/transport/protocol/responses.rs index 22c0a82ce..0ef24c067 100644 --- a/backend/crates/providers/openai/tests/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/tests/transport/protocol/responses.rs @@ -88,47 +88,6 @@ fn semantic_output_should_ignore_structural_empty_and_terminal_frames() { } } -#[test] -fn response_signals_should_count_structural_start_without_semantic_output() { - let signals = response_event_signals( - Some("response.output_item.added"), - &json!({"item": {"type": "message", "content": []}}), - ); - - assert!(signals.protocol_progress); - assert!(signals.output_start); - assert!(!signals.semantic_output); -} - -#[test] -fn first_token_should_use_only_the_official_output_item_added_boundary() { - for event in [ - "response.created", - "response.in_progress", - "keepalive", - "codex.rate_limits", - "codex.response.metadata", - "response.metadata", - "responsesapi.websocket_timing", - "response.content_part.added", - "response.reasoning_summary_part.added", - "response.output_text.delta", - "response.output_item.done", - "response.completed", - "response.failed", - "error", - ] { - let signals = response_event_signals(Some(event), &json!({"type": event})); - assert!(!signals.output_start, "not an output item start: {event}"); - } - let signals = response_event_signals( - Some("response.output_item.added"), - &json!({"type": "response.output_item.added", "item": {"type": "message", "content": []}}), - ); - assert!(signals.output_start); - assert!(!signals.semantic_output); -} - #[test] fn response_signals_should_count_text_inside_a_completed_output_item() { let signals = response_event_signals( diff --git a/docs/api.md b/docs/api.md index 8576f95a2..9e806fa05 100644 --- a/docs/api.md +++ b/docs/api.md @@ -2047,7 +2047,8 @@ OpenAI 优先采用服务端 `openai-model` / `x-openai-model` 报告(流内 `latencyMs` 从模型执行会话开始计到终结,包含账号选择、重试及流交付等待,不包含此前的入口解析、路由和准入。 `firstTokenLatencyMs` 与它使用同一计时起点,首字边界由 Provider 协议定义。 -OpenAI Responses 采用官方 Codex 的首个 `response.output_item.added` 边界; +OpenAI Responses 在 HTTP SSE 收到首个非空且非 `[DONE]` 的 `data` 载荷、 +或 WebSocket 收到首个非空响应消息时记录首字,不要求实际输出内容; xAI 采用首个非空文本、推理或工具参数输出。对应事件缺失时保留未知,不用其他事件补齐首字。 请求级首字统一通过 `firstTokenLatencyMs` 返回,由前端格式化。 `latencyDetails` 的首事件、首推理和首正文时间也使用请求级起点,首推理与首正文仍要求实际内容,