[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

@@ -1912,6 +1912,7 @@ where
         let mut logged_error = false;
         let mut tx_last_response = Some(tx_last_response);
         let mut items_added: Vec<ResponseItem> = 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,

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::<i64>().ok())
+                        .is_some()
             })
             .map(|_| Ok(()))
             .unwrap_or(Err("missing response.completed telemetry".to_string()))

codex-rs/otel/src/events/session_telemetry.rs

@@ -910,6 +910,7 @@ impl SessionTelemetry {
         cached_token_count: Option<i64>,
         reasoning_token_count: Option<i64>,
         tool_token_count: i64,
+        ttft_ms: Option<i64>,
     ) {
         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(),
             },