From 6ff670bd030f7f94ce956d8a176c226deb427666 Mon Sep 17 00:00:00 2001 From: xli-oai Date: Thu, 2 Jul 2026 04:45:03 -0700 Subject: [PATCH] [codex] emit per-request TTFT completion telemetry (#30883) ## Why Codex telemetry pipeline needs a per-request TTFT value. The existing `codex.turn_ttft` is recorded once per turn, so it cannot represent later inference requests in the same turn and can miss the beginning of hidden reasoning. This restores the low-volume per-request signal proposed in https://github.com/bk-nvidia/codex/pull/3 without bringing back per-WebSocket-event TRACE logging. ## What changed - start a timer when each mapped Responses stream begins - latch the timer on the first `response.output_item.added`, including an empty hidden-reasoning item - attach `ttft_ms` to the existing `codex.sse_event` / `response.completed` telemetry record - cover the new completion field with an integration test ## Semantics The value is per inference request, not per turn. It measures mapped-stream-to-first-output-item latency, matching the customer-proposed metric. For HTTP, the stream is already established before timing begins, so request setup and response-header latency are excluded. `response.output_item.added` is a client-visible proxy for the start of hidden reasoning; this does not claim access to the server's internal first raw-token timestamp. ## Validation - `just test -p codex-otel` (47 passed) - `just test -p codex-core process_sse_emits_completed_telemetry` (1 passed after the final timer-placement change) - attempted `just test -p codex-core`: 2,855 passed and 53 failed because of unrelated local-environment failures (missing `test_stdio_server` fixture binary, shell startup noise, and timing-sensitive tests); the focused telemetry test passed in that run as well --- codex-rs/core/src/client.rs | 7 +++++ codex-rs/core/tests/suite/otel.rs | 30 +++++++++++-------- codex-rs/otel/src/events/session_telemetry.rs | 2 ++ 3 files changed, 27 insertions(+), 12 deletions(-) diff --git a/codex-rs/core/src/client.rs b/codex-rs/core/src/client.rs index 48623a00e8..a8ef00bcc1 100644 --- a/codex-rs/core/src/client.rs +++ b/codex-rs/core/src/client.rs @@ -1912,6 +1912,7 @@ where let mut logged_error = false; let mut tx_last_response = Some(tx_last_response); let mut items_added: Vec = Vec::new(); + let (request_start, mut ttft_ms) = (Instant::now(), None); let mut api_stream = api_stream; let upstream_request_id = upstream_request_id.as_deref(); if let Some(upstream_request_id) = upstream_request_id { @@ -1961,6 +1962,7 @@ where Some(usage.cached_input_tokens), Some(usage.reasoning_output_tokens), usage.total_tokens, + ttft_ms, ); } inference_trace_attempt.record_completed( @@ -1988,6 +1990,11 @@ where } } Ok(event) => { + if matches!(&event, ResponseEvent::OutputItemAdded(_)) && ttft_ms.is_none() { + ttft_ms = Some( + i64::try_from(request_start.elapsed().as_millis()).unwrap_or(i64::MAX), + ); + } if tx_event.send(Ok(event)).await.is_err() { inference_trace_attempt.record_cancelled( STREAM_DROPPED_REASON, diff --git a/codex-rs/core/tests/suite/otel.rs b/codex-rs/core/tests/suite/otel.rs index e3415d945b..4a2b2993ad 100644 --- a/codex-rs/core/tests/suite/otel.rs +++ b/codex-rs/core/tests/suite/otel.rs @@ -567,19 +567,22 @@ async fn process_sse_emits_completed_telemetry() { mount_sse_once( &server, - sse(vec![serde_json::json!({ - "type": "response.completed", - "response": { - "id": "resp1", - "usage": { - "input_tokens": 3, - "input_tokens_details": { "cached_tokens": 1 }, - "output_tokens": 5, - "output_tokens_details": { "reasoning_tokens": 2 }, - "total_tokens": 9 + sse(vec![ + ev_reasoning_item_added("reasoning-1", &[]), + serde_json::json!({ + "type": "response.completed", + "response": { + "id": "resp1", + "usage": { + "input_tokens": 3, + "input_tokens_details": { "cached_tokens": 1 }, + "output_tokens": 5, + "output_tokens_details": { "reasoning_tokens": 2 }, + "total_tokens": 9 + } } - } - })]), + }), + ]), ) .await; @@ -612,6 +615,9 @@ async fn process_sse_emits_completed_telemetry() { && line.contains("cached_token_count=1") && line.contains("reasoning_token_count=2") && line.contains("tool_token_count=9") + && extract_log_field(line, "ttft_ms") + .and_then(|ttft_ms| ttft_ms.parse::().ok()) + .is_some() }) .map(|_| Ok(())) .unwrap_or(Err("missing response.completed telemetry".to_string())) diff --git a/codex-rs/otel/src/events/session_telemetry.rs b/codex-rs/otel/src/events/session_telemetry.rs index 8210c0e5b9..440f8f04a5 100644 --- a/codex-rs/otel/src/events/session_telemetry.rs +++ b/codex-rs/otel/src/events/session_telemetry.rs @@ -910,6 +910,7 @@ impl SessionTelemetry { cached_token_count: Option, reasoning_token_count: Option, tool_token_count: i64, + ttft_ms: Option, ) { log_and_trace_event!( self, @@ -921,6 +922,7 @@ impl SessionTelemetry { cached_token_count = cached_token_count, reasoning_token_count = reasoning_token_count, tool_token_count = %tool_token_count, + ttft_ms = ttft_ms, service_tier = self.metadata.service_tier.as_deref(), model_reasoning_effort = self.metadata.model_reasoning_effort.as_deref(), },