diff --git a/backend/crates/gateway-core/src/event.rs b/backend/crates/gateway-core/src/event.rs index 099dc87e6..a0b4f081d 100644 --- a/backend/crates/gateway-core/src/event.rs +++ b/backend/crates/gateway-core/src/event.rs @@ -500,7 +500,7 @@ pub struct ProviderResponseTimings { pub upstream_response_ms: Option, /// 上游排除引擎与客户端工具时间后的 API 耗时 pub upstream_api_overhead_ms: Option, - /// 上游引擎服务总耗时；以下专项指标保留毫秒小数及各自内部计量层级 + /// 上游响应级引擎耗时；以下专项指标保留毫秒小数及各自内部计量层级 pub upstream_engine_ms: Option, pub upstream_engine_iapi_ttft_ms: Option, pub upstream_engine_service_ttft_ms: Option, diff --git a/backend/crates/providers/openai/src/transport/canonical.rs b/backend/crates/providers/openai/src/transport/canonical.rs index 4e0cd8d2e..2158c649c 100644 --- a/backend/crates/providers/openai/src/transport/canonical.rs +++ b/backend/crates/providers/openai/src/transport/canonical.rs @@ -344,10 +344,15 @@ impl CodexCanonicalDecoder { } if event_type == Some("responsesapi.websocket_timing") && self.started && !self.completed { // 无 ID 的官方计时帧仅属于当前已开始的响应，复用连接的尾帧不能跨边界归属 - let response_id = value - .get("response_id") - .or_else(|| value.pointer("/response/id")); - if response_id.is_none_or(|id| id.as_str() == self.response_id.as_deref()) { + let response_ids = [ + value.get("response_id"), + value.pointer("/response/id"), + value.pointer("/timing_metrics/response_id"), + ]; + if response_ids.into_iter().flatten().all(|id| { + id.as_str() + .is_some_and(|id| Some(id) == self.response_id.as_deref()) + }) { self.upstream_timing_metrics .merge(ResponseTimingMetrics::from_event(&value)); } diff --git a/backend/crates/providers/openai/src/transport/protocol/responses.rs b/backend/crates/providers/openai/src/transport/protocol/responses.rs index ab6c86b6a..97a5463e1 100644 --- a/backend/crates/providers/openai/src/transport/protocol/responses.rs +++ b/backend/crates/providers/openai/src/transport/protocol/responses.rs @@ -251,22 +251,54 @@ pub(crate) struct ResponseTimingMetrics { impl ResponseTimingMetrics { pub(crate) fn from_event(value: &Value) -> Self { - let metrics = value.get("timing_metrics"); - let milliseconds = |key: &str| { - metrics? - .get(key)? - .as_f64() - .filter(|value| value.is_finite() && *value >= 0.0 && *value < i64::MAX as f64) + let Some(metrics) = value.get("timing_metrics") else { + return Self::default(); }; + let valid_milliseconds = + |value: &f64| value.is_finite() && *value >= 0.0 && *value < i64::MAX as f64; + let milliseconds = + |source: &Value, key: &str| source.get(key)?.as_f64().filter(valid_milliseconds); + if let Some(path) = metrics.get("critical_path") { + // 外层 logical_turn 可累计多次响应与工具等待，只取完整的当前响应阶段 + if path.get("scope").and_then(Value::as_str) != Some("response") + || path.get("coverage").and_then(Value::as_str) != Some("complete") + || path.get("boundary_type").and_then(Value::as_str) + != Some("actionable_output_item_done") + { + return Self::default(); + } + // API 开销由推理前与其他处理构成，缺少任一分项时不能按零补齐 + let overhead = milliseconds(path, "responses_pre_inference_ms") + .zip(milliseconds(path, "responses_other_ms")) + .map(|(before, other)| before + other) + .filter(valid_milliseconds); + return Self { + upstream_api_overhead_ms: overhead, + upstream_engine_ms: milliseconds(path, "engine_wall_ms"), + ..Self::default() + }; + } + // 直接返回的专项指标保留原口径，明确标为整轮累计的值不能归入单次请求 + if metrics + .get("timing_scope") + .is_some_and(|scope| scope.as_str() != Some("response")) + { + return Self::default(); + } Self { upstream_api_overhead_ms: milliseconds( + metrics, "responses_duration_excl_engine_and_client_tool_time_ms", ), - upstream_engine_ms: milliseconds("engine_service_total_ms"), - upstream_engine_iapi_ttft_ms: milliseconds("engine_iapi_ttft_total_ms"), - upstream_engine_service_ttft_ms: milliseconds("engine_service_ttft_total_ms"), - upstream_engine_iapi_tbt_ms: milliseconds("engine_iapi_tbt_across_engine_calls_ms"), + upstream_engine_ms: milliseconds(metrics, "engine_service_total_ms"), + upstream_engine_iapi_ttft_ms: milliseconds(metrics, "engine_iapi_ttft_total_ms"), + upstream_engine_service_ttft_ms: milliseconds(metrics, "engine_service_ttft_total_ms"), + upstream_engine_iapi_tbt_ms: milliseconds( + metrics, + "engine_iapi_tbt_across_engine_calls_ms", + ), upstream_engine_service_tbt_ms: milliseconds( + metrics, "engine_service_tbt_across_engine_calls_ms", ), } diff --git a/backend/crates/providers/openai/tests/provider/contract/timing.rs b/backend/crates/providers/openai/tests/provider/contract/timing.rs index 2fecd417e..81cf635a5 100644 --- a/backend/crates/providers/openai/tests/provider/contract/timing.rs +++ b/backend/crates/providers/openai/tests/provider/contract/timing.rs @@ -55,11 +55,71 @@ async fn finish_timings( .expect("complete upstream stream") } +async fn observe_timing_events(websocket: bool, events: Vec) -> ProviderResponseTimings { + let store = Arc::new(MemoryAccountStore::default()); + create_account(&store, "acct_provider_contract").await; + let (base_url, server) = if websocket { + 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 { + let (socket, _) = listener.accept().await.unwrap(); + let mut ws = crate::transport::accept_codex_test_websocket_with( + socket, + |request: &tokio_tungstenite::tungstenite::handshake::server::Request, response| { + assert_eq!( + request.headers()["x-responsesapi-include-timing-metrics"], + "true" + ); + response.headers_mut().insert( + "sec-websocket-extensions", + "permessage-deflate".parse().unwrap(), + ); + }, + ) + .await; + ws.next().await.unwrap().unwrap(); + for event in events { + ws.send(Message::Text(event.to_string().into())) + .await + .unwrap(); + } + }); + (base_url, server) + } else { + let (base_url, release, _, server) = paused_chunked_sse_server( + events[..1] + .iter() + .map(|event| format!("data: {event}\n\n")) + .collect(), + events[1..] + .iter() + .map(|event| format!("data: {event}\n\n")) + .collect(), + ) + .await; + release.send(()).unwrap(); + (base_url, server) + }; + let operation = if websocket { + generate_operation() + } else { + http_generate_operation() + }; + let mut stream = provider_with_base_url(&store, base_url) + .execute( + planned_request("openai", operation), + timed_context(Instant::now() - Duration::from_secs(20), 2), + ) + .await + .unwrap(); + let timings = finish_timings(&mut stream, ProviderResponseTimings::default()).await; + server.await.unwrap(); + timings +} + #[tokio::test] async fn official_response_duration_is_observed_independently_of_the_request_clock() { for websocket in [false, true] { - let store = Arc::new(MemoryAccountStore::default()); - create_account(&store, "acct_provider_contract").await; let created = json!({"type":"response.created","response":{"id":"resp_official_time","created_at":100}}); let completed = json!({"type":"response.completed","response":{"id":"resp_official_time","status":"completed","created_at":100,"completed_at":107,"output":[],"usage":{"input_tokens":1,"output_tokens":112,"total_tokens":113}}}); let metrics = json!({"type":"responsesapi.websocket_timing","timing_metrics":{ @@ -81,62 +141,7 @@ async fn official_response_duration_is_observed_independently_of_the_request_clo }}); let stale = json!({"type":"responsesapi.websocket_timing","timing_metrics":{"engine_service_total_ms":999.0}}); let events = [stale, created, metrics, wrong_id, invalid, completed]; - let (base_url, server) = if websocket { - 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 { - let (socket, _) = listener.accept().await.unwrap(); - let mut ws = crate::transport::accept_codex_test_websocket_with( - socket, - |request: &tokio_tungstenite::tungstenite::handshake::server::Request, - response| { - assert_eq!( - request.headers()["x-responsesapi-include-timing-metrics"], - "true" - ); - response.headers_mut().insert( - "sec-websocket-extensions", - "permessage-deflate".parse().unwrap(), - ); - }, - ) - .await; - ws.next().await.unwrap().unwrap(); - for event in events { - ws.send(Message::Text(event.to_string().into())) - .await - .unwrap(); - } - }); - (base_url, server) - } else { - let (base_url, release, _, server) = paused_chunked_sse_server( - events[..3] - .iter() - .map(|event| format!("data: {event}\n\n")) - .collect(), - events[3..] - .iter() - .map(|event| format!("data: {event}\n\n")) - .collect(), - ) - .await; - release.send(()).unwrap(); - (base_url, server) - }; - let operation = if websocket { - generate_operation() - } else { - http_generate_operation() - }; - let mut stream = provider_with_base_url(&store, base_url) - .execute( - planned_request("openai", operation), - timed_context(Instant::now() - Duration::from_secs(20), 2), - ) - .await - .unwrap(); - let timings = finish_timings(&mut stream, ProviderResponseTimings::default()).await; + let timings = observe_timing_events(websocket, events.to_vec()).await; 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)); @@ -146,7 +151,133 @@ async fn official_response_duration_is_observed_independently_of_the_request_clo assert_eq!(timings.upstream_engine_service_tbt_ms, Some(20.12)); assert_eq!(timings.first_token_ms, None); assert!(timings.first_event_ms.is_some_and(|value| value >= 20_000)); - server.await.unwrap(); + } +} + +// 保留实测计时帧的层级和小数，响应身份使用虚构值，不包含请求正文 +fn response_critical_path_event() -> Value { + json!({"type":"responsesapi.websocket_timing","timing_metrics":{ + "timing_scope":"logical_turn", + "response_id":"resp_critical_path", + "pre_inference_ms":211.691468, + "client_tool_pause_total_ms":3730.088369, + "total_turn_time_s":25.017284596, + "num_engine_calls":4, + "first_sampled_message_ttft_ms":null, + "critical_path":{ + "scope":"response", + "coverage":"complete", + "boundary_type":"actionable_output_item_done", + "num_engine_calls":1, + "responses_harness_unblocked_ms":6792.094664, + "responses_pre_inference_ms":35.747883, + "engine_wall_ms":6710.68507, + "client_tool_pause_ms":0.0, + "responses_other_ms":45.66171099999974, + "engine_iapi_total_ms":null, + "taas_request_setup_total_ms":null, + "taas_provider_wait_total_ms":null, + "sampling_and_stream_total_ms":null, + "inference_residual_total_ms":null + } + }}) +} + +fn critical_path_response_events(metrics: Value) -> Vec { + vec![ + json!({"type":"response.created","response":{"id":"resp_critical_path"}}), + metrics, + json!({"type":"response.completed","response":{"id":"resp_critical_path","status":"completed","created_at":100,"completed_at":107,"output":[],"usage":{"input_tokens":1,"output_tokens":1,"total_tokens":2}}}), + ] +} + +#[tokio::test] +async fn response_critical_path_is_observed_without_logical_turn_totals_on_both_transports() { + for websocket in [false, true] { + let mut metrics = response_critical_path_event(); + // 同帧即使携带整轮专项值，也不能与响应级 critical path 混算 + metrics["timing_metrics"]["engine_service_total_ms"] = json!(25_000); + metrics["timing_metrics"]["engine_iapi_ttft_total_ms"] = json!(500); + let mut events = critical_path_response_events(metrics); + let mut wrong_id = response_critical_path_event(); + wrong_id["timing_metrics"]["response_id"] = json!("resp_other"); + wrong_id["timing_metrics"]["critical_path"]["engine_wall_ms"] = json!(999); + events.insert(2, wrong_id); + let timings = observe_timing_events(websocket, events).await; + 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); + assert_eq!(timings.upstream_engine_iapi_ttft_ms, None); + assert_eq!(timings.upstream_engine_service_ttft_ms, None); + assert_eq!(timings.upstream_engine_iapi_tbt_ms, None); + assert_eq!(timings.upstream_engine_service_tbt_ms, None); + } +} + +#[tokio::test] +async fn critical_path_rejects_mismatched_identity_and_incomplete_or_cumulative_scope() { + for (pointer, value) in [ + ("/timing_metrics/response_id", json!("resp_other")), + ("/timing_metrics/response_id", Value::Null), + ("/timing_metrics/critical_path/scope", json!("logical_turn")), + ("/timing_metrics/critical_path/scope", Value::Null), + ("/timing_metrics/critical_path/coverage", json!("partial")), + ("/timing_metrics/critical_path/coverage", Value::Null), + ( + "/timing_metrics/critical_path/boundary_type", + json!("unknown"), + ), + ("/timing_metrics/critical_path", Value::Null), + ] { + let mut metrics = response_critical_path_event(); + // 正确的顶层 ID 不能掩盖嵌套 ID 的冲突 + metrics["response_id"] = json!("resp_critical_path"); + *metrics.pointer_mut(pointer).unwrap() = value; + let timings = observe_timing_events(false, critical_path_response_events(metrics)).await; + assert_eq!(timings.upstream_engine_ms, None, "{pointer}"); + assert_eq!(timings.upstream_api_overhead_ms, None, "{pointer}"); + } + for identity in ["response_id", "response"] { + let mut metrics = response_critical_path_event(); + metrics[identity] = if identity == "response" { + json!({"id":"resp_other"}) + } else { + json!("resp_other") + }; + let timings = observe_timing_events(false, critical_path_response_events(metrics)).await; + assert_eq!(timings.upstream_engine_ms, None); + } + let metrics = json!({"type":"responsesapi.websocket_timing","timing_metrics":{ + "timing_scope":"logical_turn", + "engine_service_total_ms":25_000, + "responses_duration_excl_engine_and_client_tool_time_ms":500, + "engine_iapi_ttft_total_ms":200 + }}); + let timings = observe_timing_events(false, critical_path_response_events(metrics)).await; + assert_eq!(timings.upstream_engine_ms, None); + assert_eq!(timings.upstream_api_overhead_ms, None); + assert_eq!(timings.upstream_engine_iapi_ttft_ms, None); +} + +#[tokio::test] +async fn critical_path_keeps_missing_or_invalid_components_unknown_and_preserves_zero() { + for value in [Value::Null, json!(-1), json!("45"), json!(1e99)] { + let mut metrics = response_critical_path_event(); + metrics["timing_metrics"]["critical_path"]["responses_other_ms"] = value; + let timings = observe_timing_events(false, critical_path_response_events(metrics)).await; + assert_eq!(timings.upstream_api_overhead_ms, None); + assert_eq!(timings.upstream_engine_ms, Some(6710.68507)); + } + for (component, expected) in [(0.0, Some(0.0)), (5e18, None)] { + let mut metrics = response_critical_path_event(); + let path = &mut metrics["timing_metrics"]["critical_path"]; + path["responses_other_ms"] = json!(component); + path["responses_pre_inference_ms"] = json!(component); + path["engine_wall_ms"] = json!(0); + let timings = observe_timing_events(false, critical_path_response_events(metrics)).await; + assert_eq!(timings.upstream_api_overhead_ms, expected); + assert_eq!(timings.upstream_engine_ms, Some(0.0)); } } diff --git a/docs/api.md b/docs/api.md index 1f2883c58..fc58c7309 100644 --- a/docs/api.md +++ b/docs/api.md @@ -1806,10 +1806,15 @@ OpenAI 从当前响应的 `responsesapi.websocket_timing.timing_metrics` 读取 WebSocket 握手默认请求专项计时，客户端显式提供的开关保留原值。 各字段独立可选；响应开始前、终态后的计时和明确属于其他响应的事件不写入当前请求，重试时清空 +`critical_path` 仅采集 `scope=response`、`coverage=complete` 且边界为 +`actionable_output_item_done` 的响应级指标：引擎耗时取 `engine_wall_ms`，API 开销为 +`responses_pre_inference_ms` 与 `responses_other_ms` 之和，缺失分项不按零补齐。 +外层 `logical_turn` 累计时间不进入请求指标，也不用于补齐引擎 TTFT 或 Token 间隔 + | 字段 | 上游口径 | | --- | --- | | `upstreamApiOverheadMs` | API 排除引擎与客户端工具时间后的耗时 | -| `upstreamEngineMs` | Engine Service 总耗时 | +| `upstreamEngineMs` | 响应级引擎耗时，取完整 critical path 的墙钟时间或直接返回的 Engine Service 时间 | | `upstreamEngineIapiTtftMs` | Engine IAPI TTFT 总计 | | `upstreamEngineServiceTtftMs` | Engine Service TTFT 总计 | | `upstreamEngineIapiTbtMs` | 跨引擎调用的 IAPI Token 间隔 |