fix(usage): parse response critical path timings

Read response-scoped engine and API overhead metrics while excluding logical-turn totals and incomplete observations. Validate nested response IDs and cover the observed payload over SSE and WebSocket. Co-authored-by: Codex <noreply@openai.com>
2026-11-06 08:06:19 +00:00 · 2026-10-09 13:51:40 +08:00
parent b07bca312a
commit 98704dfde8
5 changed files with 248 additions and 75 deletions
@@ -500,7 +500,7 @@ pub struct ProviderResponseTimings {
    pub upstream_response_ms: Option<u64>,
    /// 上游排除引擎与客户端工具时间后的 API 耗时
    pub upstream_api_overhead_ms: Option<f64>,
-    /// 上游引擎服务总耗时；以下专项指标保留毫秒小数及各自内部计量层级
+    /// 上游响应级引擎耗时；以下专项指标保留毫秒小数及各自内部计量层级
    pub upstream_engine_ms: Option<f64>,
    pub upstream_engine_iapi_ttft_ms: Option<f64>,
    pub upstream_engine_service_ttft_ms: Option<f64>,
@@ -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));
            }
@@ -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",
            ),
        }
@@ -55,11 +55,71 @@ async fn finish_timings(
    .expect("complete upstream stream")
 }

+async fn observe_timing_events(websocket: bool, events: Vec<Value>) -> 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<Value> {
+    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));
    }
 }

@@ -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 间隔 |