diff --git a/crates/aisix-obs/src/otlp_http_sink.rs b/crates/aisix-obs/src/otlp_http_sink.rs
index cdc780a0..1e2cb606 100644
--- a/crates/aisix-obs/src/otlp_http_sink.rs
+++ b/crates/aisix-obs/src/otlp_http_sink.rs
@@ -592,8 +592,11 @@ fn build_otlp_span(record: &SinkRecord, exporter_name: &str) -> Value {
     // "now" so the span isn't silently dropped.
     let end_unix_nano =
         parse_rfc3339_to_unix_nano(&event.occurred_at).unwrap_or_else(now_unix_nano);
+    // The span represents this ATTEMPT, so its duration is the
+    // attempt-scoped upstream latency (the request-scoped
+    // `downstream_latency_ms` rides along as an attribute instead).
     // Latency landed in milliseconds; widen + multiply.
-    let latency_nanos = (event.latency_ms as u128).saturating_mul(1_000_000);
+    let latency_nanos = (event.upstream_latency_ms as u128).saturating_mul(1_000_000);
     let start_unix_nano = end_unix_nano.saturating_sub(latency_nanos);
 
     // Status: OK (1) for 2xx, ERROR (2) otherwise.
@@ -653,8 +656,20 @@ fn build_otlp_span(record: &SinkRecord, exporter_name: &str) -> Value {
     }
     attributes.push(attr_string("aisix.exporter_name", exporter_name));
     attributes.push(attr_string("aisix.request_id", &event.request_id));
-    if event.ttft_ms > 0 {
-        attributes.push(attr_int("aisix.ttft_ms", event.ttft_ms as i64));
+    if event.upstream_ttft_ms > 0 {
+        attributes.push(attr_int(
+            "aisix.upstream_ttft_ms",
+            event.upstream_ttft_ms as i64,
+        ));
+    }
+    // Request-scoped: present only on the attempt that delivered the
+    // terminal response, so consumers read a request's caller-facing
+    // latency off that one span rather than summing the group.
+    if event.downstream_latency_ms > 0 {
+        attributes.push(attr_int(
+            "aisix.downstream_latency_ms",
+            event.downstream_latency_ms as i64,
+        ));
     }
     // Per-attempt telemetry (#655). `request_id` is the trace/group key; a
     // failover request emits one span per attempt sharing it, ordered by
@@ -892,7 +907,7 @@ mod tests {
             api_key_id: "ak-uuid".into(),
             prompt_tokens: 10,
             completion_tokens: 5,
-            latency_ms: 250,
+            upstream_latency_ms: 250,
             status_code: 200,
             provider_request_id: "chatcmpl-abc".into(),
             provider_model_version: "gpt-4o-2024-08-06".into(),
@@ -1468,19 +1483,22 @@ mod tests {
         assert!(!keys.contains(&"gen_ai.response.model"));
         assert!(!keys.contains(&"gen_ai.response.finish_reasons"));
         // ttft_ms = 0 (default) → omitted
-        assert!(!keys.contains(&"aisix.ttft_ms"));
+        assert!(!keys.contains(&"aisix.upstream_ttft_ms"));
     }
 
     #[test]
     fn payload_includes_ttft_when_set() {
         let mut ev = sample_event();
-        ev.ttft_ms = 42;
+        ev.upstream_ttft_ms = 42;
         let body = build_otlp_traces_payload(&ev, "test-exp");
         let attrs = body["resourceSpans"][0]["scopeSpans"][0]["spans"][0]["attributes"]
             .as_array()
             .unwrap();
-        let ttft_attr = attrs.iter().find(|a| a["key"] == "aisix.ttft_ms");
-        assert!(ttft_attr.is_some(), "aisix.ttft_ms should be present");
+        let ttft_attr = attrs.iter().find(|a| a["key"] == "aisix.upstream_ttft_ms");
+        assert!(
+            ttft_attr.is_some(),
+            "aisix.upstream_ttft_ms should be present"
+        );
         assert_eq!(ttft_attr.unwrap()["value"]["intValue"], "42");
     }
 
diff --git a/crates/aisix-obs/src/sink/datadog.rs b/crates/aisix-obs/src/sink/datadog.rs
index 04bb9993..4ebaf27d 100644
--- a/crates/aisix-obs/src/sink/datadog.rs
+++ b/crates/aisix-obs/src/sink/datadog.rs
@@ -474,7 +474,7 @@ mod tests {
             status_code: 200,
             prompt_tokens: 5,
             completion_tokens: 7,
-            latency_ms: 123,
+            upstream_latency_ms: 123,
             provider_model_version: "gpt-4o-2024-08-06".into(),
             finish_reason: "stop".into(),
             ..UsageEvent::default()
@@ -523,7 +523,7 @@ mod tests {
         // AISIX custom dimensions under the `aisix.` prefix.
         assert_eq!(log["aisix.request_id"], "req-42");
         assert_eq!(log["aisix.model_id"], "gpt-4o");
-        assert_eq!(log["aisix.latency_ms"], 123);
+        assert_eq!(log["aisix.upstream_latency_ms"], 123);
 
         // The API key must NEVER appear in the body anywhere.
         let body_text = serde_json::to_string(&logs).unwrap();
diff --git a/crates/aisix-obs/src/sink/sls.rs b/crates/aisix-obs/src/sink/sls.rs
index af6159b2..6ddf6363 100644
--- a/crates/aisix-obs/src/sink/sls.rs
+++ b/crates/aisix-obs/src/sink/sls.rs
@@ -487,7 +487,7 @@ mod tests {
             status_code: 200,
             prompt_tokens: 5,
             completion_tokens: 7,
-            latency_ms: 123,
+            upstream_latency_ms: 123,
             ..UsageEvent::default()
         };
         let ack = sink
@@ -559,7 +559,10 @@ mod tests {
             contents.get("completion_tokens").map(String::as_str),
             Some("7")
         );
-        assert_eq!(contents.get("latency_ms").map(String::as_str), Some("123"));
+        assert_eq!(
+            contents.get("upstream_latency_ms").map(String::as_str),
+            Some("123")
+        );
         // Empty metadata is omitted uniformly: `api_key_id` (serde `default`
         // only, would serialize as "") and `finish_reason` (`skip_serializing_if`)
         // both drop out, so the SLS log carries no blank columns.
diff --git a/crates/aisix-obs/src/usage.rs b/crates/aisix-obs/src/usage.rs
index 88ab6341..9d3ecefe 100644
--- a/crates/aisix-obs/src/usage.rs
+++ b/crates/aisix-obs/src/usage.rs
@@ -112,14 +112,57 @@ pub struct UsageEvent {
     #[serde(default, skip_serializing_if = "is_false")]
     pub usage_estimated: bool,
 
-    pub latency_ms: u32,
-
-    /// Time to first token in milliseconds. Only meaningful on the
-    /// streaming path — measures elapsed time from request entry to
-    /// the first upstream SSE chunk. 0 on non-streaming, error, and
+    /// How long THIS attempt spent on the upstream, in milliseconds:
+    /// from the moment the attempt began to the moment it settled —
+    /// end-of-stream for a streamed attempt, not first-chunk.
+    ///
+    /// Attempt-scoped, so it excludes request parsing, guardrail scans,
+    /// routing, and the inter-attempt retry backoff. Summing a request's
+    /// attempts yields upstream time, NOT what the caller waited — that
+    /// is `downstream_latency_ms`.
+    pub upstream_latency_ms: u32,
+
+    /// Time to the upstream's first token, in milliseconds — measured
+    /// from the start of THIS attempt to the first upstream SSE chunk
+    /// carrying generated output (role-only preamble chunks don't
+    /// count). Same attempt scope as `upstream_latency_ms`, so the two
+    /// are directly comparable. 0 on non-streaming, error, and
     /// cache-hit paths (omitted from the wire via skip_serializing_if).
+    ///
+    /// This is what the UPSTREAM delivered on this attempt. What the
+    /// caller actually waited for is `downstream_latency_ms`, which also
+    /// covers gateway-side work — most visibly an output guardrail that
+    /// holds the stream back to mask it — and, when the request retried,
+    /// the earlier attempts too.
     #[serde(default, skip_serializing_if = "is_zero_u32")]
-    pub ttft_ms: u32,
+    pub upstream_ttft_ms: u32,
+
+    /// What the CALLER waited for, in milliseconds: from the gateway
+    /// receiving the request to it handing the client the first thing
+    /// it can use —
+    ///
+    /// - non-streaming: the complete response is written;
+    /// - streaming: the first token is forwarded downstream.
+    ///
+    /// Request-scoped (unlike the two `upstream_*` fields above), so it
+    /// spans request parsing, guardrail scans, every failed attempt,
+    /// the retry backoff, and any output-guardrail hold-back. Recorded
+    /// once per request, on the attempt that produced the terminal
+    /// response — including a failing one, so a request that never
+    /// succeeded still shows what its caller waited for.
+    ///
+    /// `downstream_latency_ms - upstream_ttft_ms` is the wait the final
+    /// attempt's upstream did NOT account for. On a first-try request
+    /// that is gateway-side work (parsing, guardrail scans, hold-back).
+    /// On a request that retried or failed over it also contains every
+    /// earlier attempt plus the backoff, so it is NOT gateway overhead
+    /// there — read it together with `attempt_index` before attributing
+    /// the difference to anything.
+    ///
+    /// Absent (0) on the non-terminal attempts of a request, and on any
+    /// path that never reached response delivery.
+    #[serde(default, skip_serializing_if = "is_zero_u32")]
+    pub downstream_latency_ms: u32,
 
     /// HTTP status code the proxy returned to the downstream caller.
     pub status_code: u16,
@@ -253,21 +296,21 @@ pub struct UsageEvent {
     // Each UsageEvent now represents ONE upstream attempt. A request
     // that fails over emits multiple events sharing `request_id` (the
     // grouping/trace key); they are ordered by `attempt_index`. This
-    // mirrors a per-call logging model — `status_code`, `latency_ms`,
-    // and `ttft_ms` are scoped to THIS attempt. Direct (non-routing)
-    // requests emit a single event with attempt_index=0,
-    // attempt_kind="initial".
+    // mirrors a per-call logging model — `status_code`,
+    // `upstream_latency_ms` and `upstream_ttft_ms` are scoped to THIS
+    // attempt. Direct (non-routing) requests emit a single event with
+    // attempt_index=0, attempt_kind="initial".
+    //
+    // The two latency families answer different questions and are
+    // deliberately measured against different clocks:
+    //
+    //   upstream_*   — attempt-scoped. How the upstream behaved on this
+    //                  one call. Comparable across attempts.
+    //   downstream_* — request-scoped, written once per request. What
+    //                  the caller waited for, gateway overhead included.
     //
-    // `latency_ms` measures that attempt alone: from the moment the
-    // attempt begins to the moment it settles — for a streamed attempt
-    // that is end-of-stream, not first-chunk (`ttft_ms` carries the
-    // first-token figure). It therefore excludes everything outside the
-    // attempt: request parsing, guardrail scans, routing, and the
-    // inter-attempt retry backoff. Summing a request's attempts yields
-    // upstream time, NOT the user-perceived total — that lives in the
-    // access log's `latency_ms`, which spans the whole request (and, on
-    // a streamed request, stops when the response head is handed to the
-    // client rather than when the body finishes).
+    // So a request's caller-facing latency is read off the single event
+    // carrying `downstream_latency_ms` — never by summing attempts.
     /// 0-based index of this attempt within the request. Together with
     /// `request_id` it uniquely identifies one attempt.
     #[serde(default)]
@@ -795,7 +838,7 @@ mod tests {
             requested_model: "smart-group".into(),
             prompt_tokens: 12,
             completion_tokens: 34,
-            latency_ms: 56,
+            upstream_latency_ms: 56,
             status_code: 200,
             cost_usd: 0.0012,
             guardrail_blocked: false,
@@ -831,7 +874,7 @@ mod tests {
         assert!(!json.contains("provider_request_id"));
         assert!(!json.contains("provider_model_version"));
         assert!(!json.contains("finish_reason"));
-        assert!(!json.contains("ttft_ms"));
+        assert!(!json.contains("upstream_ttft_ms"));
         // ProviderKey telemetry tag wire-compat (#302 M17 /
         // AISIX-Cloud#436). Pre-attribution DP images would emit
         // empty / false defaults, which must NOT appear on the wire.
@@ -953,7 +996,7 @@ mod tests {
             provider_request_id: "chatcmpl-abc".into(),
             provider_model_version: "gpt-4o-2024-08-06".into(),
             finish_reason: "stop".into(),
-            ttft_ms: 123,
+            upstream_ttft_ms: 123,
             ..Default::default()
         };
         let json = serde_json::to_string(&ev).unwrap();
@@ -964,7 +1007,7 @@ mod tests {
         assert!(json.contains(r#""provider_request_id":"chatcmpl-abc""#));
         assert!(json.contains(r#""provider_model_version":"gpt-4o-2024-08-06""#));
         assert!(json.contains(r#""finish_reason":"stop""#));
-        assert!(json.contains(r#""ttft_ms":123"#));
+        assert!(json.contains(r#""upstream_ttft_ms":123"#));
     }
 
     #[test]
@@ -978,7 +1021,7 @@ mod tests {
             status_code: 502,
             error_class: "upstream_status".into(),
             error_message: "upstream returned 502".into(),
-            latency_ms: 2000,
+            upstream_latency_ms: 2000,
             ..Default::default()
         };
         let json = serde_json::to_string(&failed).unwrap();
diff --git a/crates/aisix-proxy/src/a2a.rs b/crates/aisix-proxy/src/a2a.rs
index 3ced0d00..e27b1f8e 100644
--- a/crates/aisix-proxy/src/a2a.rs
+++ b/crates/aisix-proxy/src/a2a.rs
@@ -294,7 +294,10 @@ fn emit_a2a_usage(
         occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
         api_key_id: auth.entry.id.clone(),
         status_code,
-        latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
         inbound_protocol: "a2a".to_string(),
         a2a_agent_name: agent.to_string(),
         a2a_method: method.to_string(),
diff --git a/crates/aisix-proxy/src/audio.rs b/crates/aisix-proxy/src/audio.rs
index dda7865b..7eadf3a4 100644
--- a/crates/aisix-proxy/src/audio.rs
+++ b/crates/aisix-proxy/src/audio.rs
@@ -1207,7 +1207,10 @@ fn emit_usage_event(
         requested_model: requested_model.to_string(),
         prompt_tokens,
         completion_tokens,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         status_code,
         inbound_protocol: "openai".to_string(),
         applied_guardrails: applied_guardrails.to_vec(),
diff --git a/crates/aisix-proxy/src/chat.rs b/crates/aisix-proxy/src/chat.rs
index 2a096d0c..f8a9407c 100644
--- a/crates/aisix-proxy/src/chat.rs
+++ b/crates/aisix-proxy/src/chat.rs
@@ -250,7 +250,11 @@ pub async fn chat_completions(
                         cache_status: success.cache_status.as_str().to_string(),
                         cache_hit_saved_input_tokens: success.cache_hit_saved_input_tokens,
                         cache_hit_saved_output_tokens: success.cache_hit_saved_output_tokens,
-                        ttft_ms: 0,
+                        // Non-streaming: nothing was streamed, so there is no
+                        // upstream TTFT; the caller waited for the whole
+                        // response, which is the request clock.
+                        upstream_ttft_ms: 0,
+                        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
                         attempt_index: winner.map(|w| w.index).unwrap_or(0),
                         attempt_kind: winner.map(|w| w.kind).unwrap_or("initial").to_string(),
                         attempt_model: winner.map(|w| w.target_model.clone()).unwrap_or_default(),
@@ -525,7 +529,8 @@ pub async fn chat_completions(
                             cache_status: c.cache_status.as_str().to_string(),
                             cache_hit_saved_input_tokens: 0,
                             cache_hit_saved_output_tokens: 0,
-                            ttft_ms: 0,
+                            upstream_ttft_ms: 0,
+                            downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
                             attempt_index: winner.map(|w| w.index).unwrap_or(0),
                             attempt_kind: winner.map(|w| w.kind).unwrap_or("initial").to_string(),
                             attempt_model: winner
@@ -1626,6 +1631,7 @@ async fn dispatch(
             now,
             stream_guardrail,
             started,
+            winner_attempt_started,
             req.model.clone(),
             content_cap,
             client_requested_usage,
@@ -1685,7 +1691,8 @@ async fn dispatch(
                         cache_status: CacheStatus::Disabled.as_str().to_string(),
                         cache_hit_saved_input_tokens: 0,
                         cache_hit_saved_output_tokens: 0,
-                        ttft_ms: comp.ttft_ms,
+                        upstream_ttft_ms: comp.upstream_ttft_ms,
+                        downstream_latency_ms: comp.downstream_latency_ms,
                         // #554: the winning attempt may be a fallback target,
                         // not the initial one — record the real index/kind.
                         attempt_index: winner_idx,
@@ -1772,7 +1779,7 @@ async fn dispatch(
                         status: 200,
                         streaming: true,
                     },
-                    Duration::from_millis(u64::from(comp.ttft_ms)),
+                    Duration::from_millis(u64::from(comp.upstream_ttft_ms)),
                 );
                 metrics_for_stream.record_time_to_first_token(
                     UsageLabels {
@@ -1788,7 +1795,7 @@ async fn dispatch(
                         user_id: user_id_for_metrics.as_deref().unwrap_or("unknown"),
                         user_name: user_name_for_metrics.as_deref().unwrap_or("unknown"),
                     },
-                    Duration::from_millis(u64::from(comp.ttft_ms)),
+                    Duration::from_millis(u64::from(comp.upstream_ttft_ms)),
                 );
                 // Release the concurrency permit(s) now that the stream has
                 // completed (or was cancelled). on_complete is fired by the
@@ -2896,6 +2903,10 @@ async fn dispatch_ensemble(
             }
         };
 
+        // The judge is this response's upstream call, so its own clock is
+        // what `upstream_ttft_ms` should be measured against — `started`
+        // additionally covers the whole panel that ran before it.
+        let judge_started = Instant::now();
         let judge_stream = match judge_bridge.chat_stream(&judge_req, &judge_ctx).await {
             Ok(s) => s,
             // Judge connect failed AFTER the panel round-tripped: bill the
@@ -3022,6 +3033,7 @@ async fn dispatch_ensemble(
             created_ts,
             stream_guardrail,
             started,
+            judge_started,
             // Re-stamp the client-facing ensemble model name (e.g. "council")
             // onto every chunk — never the judge's upstream model id.
             req.model.clone(),
@@ -3115,7 +3127,11 @@ async fn dispatch_ensemble(
                             comp.bypass_reason
                         },
                         cache_status: CacheStatus::Disabled.as_str().to_string(),
-                        ttft_ms: comp.ttft_ms,
+                        upstream_ttft_ms: comp.upstream_ttft_ms,
+                        // The judge's stream is what the caller sees, so its
+                        // event carries the request-scoped figure; the panel
+                        // members' sub-call events leave it 0.
+                        downstream_latency_ms: comp.downstream_latency_ms,
                         attempt_index: judge_attempt_index,
                         attempt_kind: "judge".to_string(),
                         attempt_model: judge_attempt_model.clone(),
@@ -3161,7 +3177,7 @@ async fn dispatch_ensemble(
                         status: 200,
                         streaming: true,
                     },
-                    Duration::from_millis(u64::from(comp.ttft_ms)),
+                    Duration::from_millis(u64::from(comp.upstream_ttft_ms)),
                 );
                 // Release the concurrency permit(s) now the stream is done
                 // (or was cancelled) — on_complete fires on both paths (#450).
@@ -3650,7 +3666,8 @@ fn emit_usage_event(
         cache_creation_tokens: extras.cache_creation_tokens,
         cache_read_tokens: extras.cache_read_tokens,
         usage_estimated: extras.usage_estimated,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: extras.downstream_latency_ms,
         status_code,
         provider_request_id: extras.provider_request_id,
         provider_model_version: extras.provider_model_version,
@@ -3664,7 +3681,7 @@ fn emit_usage_event(
         cache_status: extras.cache_status,
         cache_hit_saved_input_tokens: extras.cache_hit_saved_input_tokens,
         cache_hit_saved_output_tokens: extras.cache_hit_saved_output_tokens,
-        ttft_ms: extras.ttft_ms,
+        upstream_ttft_ms: extras.upstream_ttft_ms,
         // chat.rs is the OpenAI-shape /v1/chat/completions handler.
         // /v1/responses / /v1/embeddings / /v1/audio* / /v1/images* /
         // /v1/rerank don't emit UsageEvents today; when they do they
@@ -3779,7 +3796,12 @@ struct UsageExtras {
     /// ingest from these + its pricing catalog (see #88).
     cache_hit_saved_input_tokens: u32,
     cache_hit_saved_output_tokens: u32,
-    ttft_ms: u32,
+    /// Attempt-scoped time to the upstream's first generated chunk.
+    upstream_ttft_ms: u32,
+    /// Request-scoped time until the caller got its first usable byte.
+    /// Set only on the attempt that delivered the terminal response;
+    /// left 0 on the others so a request carries exactly one figure.
+    downstream_latency_ms: u32,
     // ─── Per-attempt telemetry (#655) ───
     /// 0-based attempt index within the request.
     attempt_index: u32,
@@ -4007,9 +4029,14 @@ struct StreamCompletion {
     /// fail-opened on a streamed response. Empty string = no bypass.
     /// First-bypass-wins matches the non-streaming convention.
     bypass_reason: String,
-    /// Time to first token in milliseconds. Set once when the first
-    /// Ok(chunk) arrives in `build_sse_stream`.
-    ttft_ms: u32,
+    /// Attempt-scoped time to the UPSTREAM's first generated chunk.
+    /// Set once when that chunk arrives in `build_sse_stream`.
+    upstream_ttft_ms: u32,
+    /// Request-scoped time until the first chunk was handed DOWNSTREAM.
+    /// Under a hold-back output guardrail this trails
+    /// `upstream_ttft_ms` by the scan; without one they nearly coincide.
+    /// 0 when the stream ended before any content reached the client.
+    downstream_latency_ms: u32,
     /// Count of SSE events the **consumer actually pulled** from the
     /// stream — incremented on the post-yield resume in
     /// `build_sse_stream`. `async_stream::stream!` semantics: code
@@ -4177,7 +4204,13 @@ fn build_sse_stream<F>(
     upstream: aisix_gateway::ChatChunkStream,
     created: i64,
     output_guardrail: Option<StreamGuardrailContext>,
+    // Request clock — when the gateway received the request. Measures
+    // what the CALLER waited for (`downstream_latency_ms`).
     started: Instant,
+    // Attempt clock — when this upstream attempt began. Measures how the
+    // UPSTREAM behaved (`upstream_ttft_ms`), so retries and pre-dispatch
+    // work don't inflate it.
+    attempt_started: Instant,
     // Customer-facing model name (alias / routing group), re-stamped
     // onto every SSE chunk's `model` field per AISIX-Cloud#410. Owned
     // so it can move into the `async_stream::stream!` closure.
@@ -4287,6 +4320,17 @@ where
         // the Err arm mirrors the pre-hold-back defensive error frame.
         macro_rules! chunk_event {
             ($chunk:expr) => {{
+                // The caller's wait ends here — not when the upstream chunk
+                // arrived; a hold-back guardrail can sit between the two.
+                // Every content chunk is rendered through this macro, on
+                // both the live-forward and the hold-back release paths, so
+                // this catches the first one either way. Written straight
+                // into the accumulator rather than a local so a client that
+                // disconnects mid-stream still reports what it waited for.
+                if guard.comp().downstream_latency_ms == 0 {
+                    guard.comp().downstream_latency_ms =
+                        started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                }
                 let rendered = render_chunk(created, $chunk, &client_facing_model);
                 match serde_json::to_string(&rendered) {
                     Ok(json) => Event::default().data(json),
@@ -4309,8 +4353,8 @@ where
                         && (chunk.delta.content.is_some() || chunk.delta.tool_calls.is_some())
                     {
                         first_chunk_seen = true;
-                        guard.comp().ttft_ms =
-                            started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                        guard.comp().upstream_ttft_ms =
+                            attempt_started.elapsed().as_millis().min(u32::MAX as u128) as u32;
                     }
                     let comp = guard.comp();
                     if !chunk.id.is_empty() {
diff --git a/crates/aisix-proxy/src/completions.rs b/crates/aisix-proxy/src/completions.rs
index 7dbf5881..d036f4d1 100644
--- a/crates/aisix-proxy/src/completions.rs
+++ b/crates/aisix-proxy/src/completions.rs
@@ -638,7 +638,10 @@ fn emit_usage_event(
         prompt_tokens: usage.prompt_tokens,
         completion_tokens: usage.completion_tokens,
         usage_estimated: usage.usage_estimated,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         status_code,
         inbound_protocol: "openai".to_string(),
         client_source_ip: client.source_ip.clone(),
diff --git a/crates/aisix-proxy/src/embeddings.rs b/crates/aisix-proxy/src/embeddings.rs
index 44c52dcb..087ad154 100644
--- a/crates/aisix-proxy/src/embeddings.rs
+++ b/crates/aisix-proxy/src/embeddings.rs
@@ -633,7 +633,10 @@ fn emit_usage_event(
         requested_model: requested_model.to_string(),
         prompt_tokens,
         usage_estimated,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         status_code,
         inbound_protocol: "openai".to_string(),
         applied_guardrails: applied_guardrails.to_vec(),
diff --git a/crates/aisix-proxy/src/images.rs b/crates/aisix-proxy/src/images.rs
index ce580866..a0c8978b 100644
--- a/crates/aisix-proxy/src/images.rs
+++ b/crates/aisix-proxy/src/images.rs
@@ -443,7 +443,10 @@ fn emit_usage_event(
         requested_model: requested_model.to_string(),
         prompt_tokens,
         completion_tokens,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         status_code,
         inbound_protocol: "openai".to_string(),
         applied_guardrails: applied_guardrails.to_vec(),
diff --git a/crates/aisix-proxy/src/jobs.rs b/crates/aisix-proxy/src/jobs.rs
index aff4dc6b..a0b4bdc6 100644
--- a/crates/aisix-proxy/src/jobs.rs
+++ b/crates/aisix-proxy/src/jobs.rs
@@ -581,7 +581,10 @@ fn emit_job_usage_event(
         api_key_id: auth.entry.id.clone(),
         requested_model: target.display_name().to_string(),
         status_code,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         inbound_protocol: "openai".to_string(),
         client_source_ip: client.source_ip.clone(),
         client_user_agent: client.user_agent.clone(),
diff --git a/crates/aisix-proxy/src/mcp.rs b/crates/aisix-proxy/src/mcp.rs
index 46d0fce3..bbe98701 100644
--- a/crates/aisix-proxy/src/mcp.rs
+++ b/crates/aisix-proxy/src/mcp.rs
@@ -382,7 +382,10 @@ fn emit_tool_call_usage(
         occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
         api_key_id: auth.entry.id.clone(),
         status_code,
-        latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: latency.as_millis().min(u32::MAX as u128) as u32,
         inbound_protocol: "mcp".to_string(),
         mcp_server_name: mcp_server.to_string(),
         mcp_tool_name: mcp_tool.to_string(),
diff --git a/crates/aisix-proxy/src/messages.rs b/crates/aisix-proxy/src/messages.rs
index 536b04c3..faf5f874 100644
--- a/crates/aisix-proxy/src/messages.rs
+++ b/crates/aisix-proxy/src/messages.rs
@@ -258,6 +258,11 @@ pub async fn messages(
                 let winner_latency = winner
                     .map(|w| Duration::from_millis(u64::from(w.latency_ms)))
                     .unwrap_or(elapsed);
+                // Non-streaming: the caller waited for the complete response,
+                // which is exactly the request clock. (The streaming paths
+                // stamp this from inside the stream and skip this branch.)
+                let mut metrics = metrics;
+                metrics.downstream_latency_ms = elapsed.as_millis().min(u32::MAX as u128) as u32;
                 emit_anthropic_usage_event(
                     &state,
                     &request_id,
@@ -1283,6 +1288,7 @@ async fn anthropic_passthrough_dispatch(
         let parsed_stream = build_anthropic_passthrough_stream(
             body_stream,
             started,
+            attempt_started,
             stream_guardrail,
             model_name.to_string(),
             content_cap,
@@ -1320,7 +1326,8 @@ async fn anthropic_passthrough_dispatch(
                     provider_request_id: usage.provider_request_id,
                     provider_model_version: usage.provider_model_version,
                     finish_reason: usage.finish_reason,
-                    ttft_ms: usage.ttft_ms,
+                    upstream_ttft_ms: usage.upstream_ttft_ms,
+                    downstream_latency_ms: usage.downstream_latency_ms,
                 };
                 state_c.metrics.record_request_e2e_latency(
                     LatencyLabels {
@@ -1686,7 +1693,9 @@ fn anthropic_metrics_from_response_json(body: &Value) -> AnthropicUsageMetrics {
             .and_then(Value::as_str)
             .unwrap_or("")
             .to_string(),
-        ttft_ms: 0,
+        upstream_ttft_ms: 0,
+        // Non-streaming: stamped by the handler, which holds the request clock.
+        downstream_latency_ms: 0,
     }
 }
 
@@ -1920,6 +1929,7 @@ async fn cross_provider_dispatch(
             upstream,
             encoder,
             started,
+            attempt_started,
             stream_guardrail,
             model_name.to_string(),
             content_cap,
@@ -1952,7 +1962,8 @@ async fn cross_provider_dispatch(
                     provider_request_id: comp.provider_request_id,
                     provider_model_version: comp.provider_model_version,
                     finish_reason: comp.finish_reason,
-                    ttft_ms: comp.ttft_ms,
+                    upstream_ttft_ms: comp.upstream_ttft_ms,
+                    downstream_latency_ms: comp.downstream_latency_ms,
                 };
                 state_for_telem.metrics.record_request_e2e_latency(
                     LatencyLabels {
@@ -2103,7 +2114,9 @@ async fn cross_provider_dispatch(
         provider_request_id: resp.id.clone(),
         provider_model_version: resp.model.clone(),
         finish_reason: finish_reason_label(&resp.finish_reason),
-        ttft_ms: 0,
+        upstream_ttft_ms: 0,
+        // Non-streaming: stamped by the handler, which holds the request clock.
+        downstream_latency_ms: 0,
     };
     // Token-estimation fallback (AISIX-Cloud#1074): fill counters the
     // bridged upstream never reported. Telemetry only — the rendered
@@ -2154,7 +2167,12 @@ async fn cross_provider_dispatch(
 fn build_anthropic_sse_stream(
     upstream: aisix_gateway::ChatChunkStream,
     encoder: aisix_provider_anthropic::AnthropicSseEncoder,
+    // Request clock — what the CALLER waited for
+    // (`downstream_latency_ms`), spanning every earlier attempt.
     started: Instant,
+    // Attempt clock — how the UPSTREAM behaved on this call
+    // (`upstream_ttft_ms`).
+    attempt_started: Instant,
     output_guardrail: Option<std::sync::Arc<aisix_guardrails::GuardrailChain>>,
     model_label: String,
     // Largest content cap any content-capturing exporter wants, or `None` to
@@ -2168,6 +2186,19 @@ fn build_anthropic_sse_stream(
     use futures::StreamExt;
 
     let mut encoder = encoder;
+    // Stamp the caller-facing figure on the first SSE bytes that actually
+    // leave for the client. Wrapping the encoder output here covers both
+    // the live-forward drain and the hold-back release; putting it on the
+    // outermost stream instead would misfire on a keep-alive heartbeat.
+    macro_rules! downstream_bytes {
+        ($guard:expr, $ev:expr) => {{
+            if $guard.comp().downstream_latency_ms == 0 {
+                $guard.comp().downstream_latency_ms =
+                    started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+            }
+            bytes::Bytes::from($ev.to_sse_string())
+        }};
+    }
     // #932 / #466-class: when the chain's streamed-output policy is the
     // whole-response hold-back (BufferFull — keyword/pii/bedrock output
     // guardrails), chunks are withheld from the encoder until the
@@ -2211,8 +2242,8 @@ fn build_anthropic_sse_stream(
                         && (chunk.delta.content.is_some() || chunk.delta.tool_calls.is_some())
                     {
                         first_chunk_seen = true;
-                        guard.comp().ttft_ms =
-                            started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                        guard.comp().upstream_ttft_ms =
+                            attempt_started.elapsed().as_millis().min(u32::MAX as u128) as u32;
                     }
                     let comp = guard.comp();
                     if !chunk.id.is_empty() {
@@ -2292,7 +2323,7 @@ fn build_anthropic_sse_stream(
                         continue;
                     }
                     for ev in encoder.next_events(&chunk) {
-                        yield Ok::<_, std::io::Error>(bytes::Bytes::from(ev.to_sse_string()));
+                        yield Ok::<_, std::io::Error>(downstream_bytes!(guard, ev));
                     }
                     if encoder.is_finished() {
                         break;
@@ -2417,7 +2448,7 @@ fn build_anthropic_sse_stream(
             }
             for chunk in held_chunks.drain(..) {
                 for ev in encoder.next_events(&chunk) {
-                    yield Ok::<_, std::io::Error>(bytes::Bytes::from(ev.to_sse_string()));
+                    yield Ok::<_, std::io::Error>(downstream_bytes!(guard, ev));
                 }
                 if encoder.is_finished() {
                     break;
@@ -2426,7 +2457,7 @@ fn build_anthropic_sse_stream(
         }
         if !encoder.is_finished() {
             for ev in encoder.force_finish() {
-                yield Ok(bytes::Bytes::from(ev.to_sse_string()));
+                yield Ok(downstream_bytes!(guard, ev));
             }
         }
     };
@@ -2479,7 +2510,12 @@ struct AnthropicStreamCompletion {
     provider_request_id: String,
     provider_model_version: String,
     finish_reason: String,
-    ttft_ms: u32,
+    /// Attempt-scoped time to the upstream's first generated chunk.
+    upstream_ttft_ms: u32,
+    /// Request-scoped time until the caller got its first response
+    /// bytes. Trails `upstream_ttft_ms` by whatever the gateway did
+    /// in between — most visibly a hold-back output guardrail.
+    downstream_latency_ms: u32,
     /// Generated output (content + reasoning + tool-call text) accumulated
     /// for the token-estimation fallback (AISIX-Cloud#1074). Always on,
     /// bounded to `token_estimate::OUTPUT_ACCUMULATION_CAP`; never leaves
@@ -2612,7 +2648,8 @@ struct AnthropicUsageMetrics {
     provider_request_id: String,
     provider_model_version: String,
     finish_reason: String,
-    ttft_ms: u32,
+    upstream_ttft_ms: u32,
+    downstream_latency_ms: u32,
 }
 
 /// Emit a UsageEvent for a `/v1/messages` request. Mirrors
@@ -2684,12 +2721,13 @@ fn emit_anthropic_usage_event(
         cache_creation_tokens: metrics.cache_creation_tokens,
         cache_read_tokens: metrics.cache_read_tokens,
         usage_estimated: metrics.usage_estimated,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: metrics.downstream_latency_ms,
         status_code,
         provider_request_id: metrics.provider_request_id,
         provider_model_version: metrics.provider_model_version,
         finish_reason: metrics.finish_reason,
-        ttft_ms: metrics.ttft_ms,
+        upstream_ttft_ms: metrics.upstream_ttft_ms,
         inbound_protocol: "anthropic".to_string(),
         attempt_index: attempt.index,
         attempt_kind: attempt.kind,
@@ -2757,7 +2795,7 @@ fn emit_anthropic_usage_event(
         u64::from(metrics.completion_tokens),
         total_tokens_all,
     );
-    if metrics.ttft_ms > 0 {
+    if metrics.upstream_ttft_ms > 0 {
         state.metrics.record_request_ttft(
             LatencyLabels {
                 endpoint: "/v1/messages",
@@ -2766,7 +2804,7 @@ fn emit_anthropic_usage_event(
                 status: status_code,
                 streaming: true,
             },
-            Duration::from_millis(u64::from(metrics.ttft_ms)),
+            Duration::from_millis(u64::from(metrics.upstream_ttft_ms)),
         );
         state.metrics.record_time_to_first_token(
             UsageLabels {
@@ -2782,7 +2820,7 @@ fn emit_anthropic_usage_event(
                 user_id: user_id.unwrap_or("unknown"),
                 user_name: user_name.unwrap_or("unknown"),
             },
-            Duration::from_millis(u64::from(metrics.ttft_ms)),
+            Duration::from_millis(u64::from(metrics.upstream_ttft_ms)),
         );
     }
 }
@@ -2829,7 +2867,12 @@ struct AnthropicStreamUsage {
     provider_request_id: String,
     provider_model_version: String,
     finish_reason: String,
-    ttft_ms: u32,
+    /// Attempt-scoped time to the upstream's first content frame.
+    upstream_ttft_ms: u32,
+    /// Request-scoped time until the caller got its first response
+    /// bytes. Trails `upstream_ttft_ms` by whatever the gateway did
+    /// in between — most visibly a hold-back output guardrail.
+    downstream_latency_ms: u32,
     /// Count of upstream byte-chunks actually delivered to the client
     /// (read by the Drop guard for the #419 cost-leak gate).
     chunks_delivered: u32,
@@ -2854,12 +2897,14 @@ struct AnthropicStreamUsage {
 }
 
 /// Update the accumulator from one parsed SSE `data:` JSON object.
-/// Best-effort: unrecognised `type` values are ignored. `started` +
-/// `first_token_seen` drive the TTFT measurement (first content frame).
+/// Best-effort: unrecognised `type` values are ignored. The TTFT
+/// measurement (first content frame) is driven by `attempt_started` and
+/// `first_token_seen`, and is attempt-scoped — see
+/// `UsageEvent::upstream_ttft_ms`.
 fn update_anthropic_usage(
     acc: &mut AnthropicStreamUsage,
     json: &Value,
-    started: Instant,
+    attempt_started: Instant,
     first_token_seen: &mut bool,
 ) {
     match json.get("type").and_then(Value::as_str) {
@@ -2897,7 +2942,8 @@ fn update_anthropic_usage(
             // First content frame → record time-to-first-token.
             if !*first_token_seen {
                 *first_token_seen = true;
-                acc.ttft_ms = started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                acc.upstream_ttft_ms =
+                    attempt_started.elapsed().as_millis().min(u32::MAX as u128) as u32;
             }
             // Accumulate assistant output for the end-of-stream output
             // guardrail (#448). text streams as `delta.text`; tool_use
@@ -3004,7 +3050,7 @@ fn update_anthropic_usage(
 fn drain_anthropic_sse_frames(
     buf: &mut Vec<u8>,
     acc: &mut AnthropicStreamUsage,
-    started: Instant,
+    attempt_started: Instant,
     first_token_seen: &mut bool,
 ) {
     // SSE event delimiter is a blank line. Anthropic emits `\n\n`;
@@ -3013,7 +3059,7 @@ fn drain_anthropic_sse_frames(
         let frame: Vec<u8> = buf.drain(..end).collect();
         if let Some(data) = extract_sse_data_line(&frame) {
             if let Ok(json) = serde_json::from_slice::<Value>(data) {
-                update_anthropic_usage(acc, &json, started, first_token_seen);
+                update_anthropic_usage(acc, &json, attempt_started, first_token_seen);
             }
         }
     }
@@ -3152,9 +3198,15 @@ impl<T> Stream for AnthropicDeliveryCounter<T> {
 /// in-flight and `on_complete` fires once at end-of-stream (or
 /// client-disconnect) with the accumulated counts. Bytes are forwarded
 /// verbatim — the client sees the exact upstream SSE wire shape.
+#[allow(clippy::too_many_arguments)]
 fn build_anthropic_passthrough_stream<S, F>(
     upstream: S,
+    // Request clock — what the CALLER waited for
+    // (`downstream_latency_ms`), spanning every earlier attempt.
     started: Instant,
+    // Attempt clock — how the UPSTREAM behaved on this call
+    // (`upstream_ttft_ms`).
+    attempt_started: Instant,
     output_guardrail: Option<std::sync::Arc<aisix_guardrails::GuardrailChain>>,
     model_label: String,
     // When `Some`, the assembled `response_text` is preserved (not taken by the
@@ -3206,7 +3258,7 @@ where
                 drain_anthropic_sse_frames(
                     &mut buf,
                     guard.usage(),
-                    started,
+                    attempt_started,
                     &mut first_token_seen,
                 );
                 // Bound the frame buffer (PR #436 audit MEDIUM-2). The
@@ -3255,6 +3307,10 @@ where
             // hold-back mode an Err lands here too: it is forwarded and
             // the held (unscanned) content is dropped — fail closed.
             let errored = item.is_err();
+            if !errored && guard.usage().downstream_latency_ms == 0 {
+                guard.usage().downstream_latency_ms =
+                    started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+            }
             yield item;
             if errored && hold_policy.is_some() {
                 return;
@@ -3371,9 +3427,17 @@ where
                         &mut guard.usage().redacted_entity_counts,
                         counts,
                     );
+                    if guard.usage().downstream_latency_ms == 0 {
+                        guard.usage().downstream_latency_ms =
+                            started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                    }
                     yield Ok(Bytes::from(rewritten));
                 }
                 None => {
+                    if guard.usage().downstream_latency_ms == 0 {
+                        guard.usage().downstream_latency_ms =
+                            started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                    }
                     yield Ok(Bytes::from(std::mem::take(&mut held)));
                 }
             }
@@ -4192,7 +4256,7 @@ data: [DONE]\n\n";
         assert_eq!(event.provider_model_version, "gpt-4o-2024-08-06");
         assert_eq!(event.finish_reason, "stop");
         assert!(
-            event.ttft_ms > 0,
+            event.upstream_ttft_ms > 0,
             "streaming /v1/messages telemetry must record TTFT"
         );
         assert!(rx.try_recv().is_err(), "usage event should be emitted once");
@@ -4634,7 +4698,7 @@ event: message_stop\ndata: {\"type\":\"message_stop\"}\n\n";
         assert_eq!(event.finish_reason, "end_turn");
         assert_eq!(event.status_code, 200);
         assert!(
-            event.ttft_ms > 0,
+            event.upstream_ttft_ms > 0,
             "streaming /v1/messages telemetry must record TTFT",
         );
         assert!(rx.try_recv().is_err(), "usage event should be emitted once");
diff --git a/crates/aisix-proxy/src/passthrough.rs b/crates/aisix-proxy/src/passthrough.rs
index 88186073..b0329546 100644
--- a/crates/aisix-proxy/src/passthrough.rs
+++ b/crates/aisix-proxy/src/passthrough.rs
@@ -700,7 +700,10 @@ fn emit_usage_event(
         occurred_at: chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Secs, true),
         api_key_id: api_key_id.to_string(),
         status_code,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         inbound_protocol: "passthrough".to_string(),
         client_source_ip: client.source_ip.clone(),
         client_user_agent: client.user_agent.clone(),
diff --git a/crates/aisix-proxy/src/realtime.rs b/crates/aisix-proxy/src/realtime.rs
index 4020572d..2d62f076 100644
--- a/crates/aisix-proxy/src/realtime.rs
+++ b/crates/aisix-proxy/src/realtime.rs
@@ -639,7 +639,10 @@ async fn run_session(
         completion_tokens: usage.output_tokens.min(u32::MAX as u64) as u32,
         cached_prompt_tokens: usage.cached_tokens.min(u32::MAX as u64) as u32,
         status_code: close_status,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         cost_usd: model_entry
             .value
             .cost
diff --git a/crates/aisix-proxy/src/rerank.rs b/crates/aisix-proxy/src/rerank.rs
index 71aa31ed..64fcd9d7 100644
--- a/crates/aisix-proxy/src/rerank.rs
+++ b/crates/aisix-proxy/src/rerank.rs
@@ -607,7 +607,10 @@ fn emit_usage_event(
         api_key_id: api_key_id.to_string(),
         requested_model: requested_model.to_string(),
         prompt_tokens: usage.prompt_tokens,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         status_code,
         inbound_protocol: "openai".to_string(),
         applied_guardrails: applied_guardrails.to_vec(),
diff --git a/crates/aisix-proxy/src/responses.rs b/crates/aisix-proxy/src/responses.rs
index 060913bb..c0979e53 100644
--- a/crates/aisix-proxy/src/responses.rs
+++ b/crates/aisix-proxy/src/responses.rs
@@ -136,6 +136,14 @@ struct ResponseUsage {
     /// verbatim OpenAI path (OpenAI surfaces cache hits via
     /// `cached_prompt_tokens` instead).
     cache_read_tokens: u32,
+    /// Attempt-scoped time to the upstream's first content delta. 0 on the
+    /// non-streaming paths. Before this existed `/v1/responses` reported no
+    /// TTFT at all, so codex-class clients showed blank.
+    upstream_ttft_ms: u32,
+    /// Request-scoped time until the caller got its first response bytes.
+    /// 0 until the stream forwards something (or the handler stamps it on
+    /// the non-streaming paths).
+    downstream_latency_ms: u32,
 }
 
 pub async fn responses(
@@ -239,7 +247,12 @@ pub async fn responses(
                     },
                     elapsed,
                 );
-                if let Some(usage) = success.usage {
+                if let Some(mut usage) = success.usage {
+                    // Non-streaming: the caller waited for the complete
+                    // response, which is exactly the request clock. Streamed
+                    // responses stamp this from inside the stream and never
+                    // reach this branch (`usage` is None there).
+                    usage.downstream_latency_ms = elapsed.as_millis().min(u32::MAX as u128) as u32;
                     // Winning-attempt classification (#655). Direct models
                     // have no recorded attempt → AttemptInfo defaults.
                     let winner = success.routing.winner();
@@ -1361,6 +1374,8 @@ async fn responses_to_target(
         );
         let parsed_stream = build_responses_passthrough_stream(
             body_stream,
+            started,
+            attempt_started,
             content_cap,
             eos_scan,
             move |mut usage, out_text, output_hits| {
@@ -1824,6 +1839,7 @@ async fn responses_cross_provider_to_target(
             upstream,
             encoder,
             started,
+            attempt_started,
             output_guardrail,
             hold_back,
             max_buffer_bytes,
@@ -1856,6 +1872,8 @@ async fn responses_cross_provider_to_target(
                     cache_creation_tokens: comp.cache_creation_tokens,
                     cache_read_tokens: comp.cache_read_tokens,
                     usage_estimated: comp.usage_estimated,
+                    upstream_ttft_ms: comp.upstream_ttft_ms,
+                    downstream_latency_ms: comp.downstream_latency_ms,
                 };
                 // A clean stream is a committed 200; an output-guardrail block
                 // (or fail-closed overflow) bills the upstream tokens but is
@@ -1965,6 +1983,8 @@ async fn responses_cross_provider_to_target(
             cache_creation_tokens: resp.usage.cache_creation_tokens,
             cache_read_tokens: resp.usage.cache_read_tokens,
             usage_estimated: false,
+            upstream_ttft_ms: 0,
+            downstream_latency_ms: 0,
         };
         // Token-estimation fallback (AISIX-Cloud#1074): fill counters the
         // bridged upstream never reported. Telemetry only — the re-encoded
@@ -2135,6 +2155,10 @@ fn extract_response_usage(body: &Value) -> Option<ResponseUsage> {
         // OpenAI verbatim path: no Anthropic-style cache counters.
         cache_creation_tokens: 0,
         cache_read_tokens: 0,
+        // Carried across by the caller (`drain_responses_sse_frames`), which
+        // measured these before this terminal frame arrived.
+        upstream_ttft_ms: 0,
+        downstream_latency_ms: 0,
     })
 }
 
@@ -2186,10 +2210,27 @@ fn responses_sse_usage(bytes: &[u8]) -> Option<ResponseUsage> {
 /// an incomplete trailing frame is left in `buf` for the next chunk. Reuses
 /// the shared SSE framing helpers from the `/v1/messages` passthrough so the
 /// two surfaces parse identically.
+/// Whether this SSE event carries generated output. Mirrors the set
+/// `SseTextCapture::observe` accumulates, so TTFT lands on the same frame
+/// the capture considers the first real token.
+fn is_responses_content_delta(json: &Value) -> bool {
+    matches!(
+        json.get("type").and_then(Value::as_str),
+        Some(
+            "response.output_text.delta"
+                | "response.function_call_arguments.delta"
+                | "response.mcp_call_arguments.delta"
+                | "response.custom_tool_call_input.delta"
+        )
+    )
+}
+
 fn drain_responses_sse_frames(
     buf: &mut Vec<u8>,
     acc: &mut Option<ResponseUsage>,
     mut capture: Option<&mut SseTextCapture>,
+    attempt_started: Instant,
+    first_token_seen: &mut bool,
 ) {
     while let Some(end) = crate::messages::find_frame_end(buf) {
         let frame: Vec<u8> = buf.drain(..end).collect();
@@ -2198,8 +2239,26 @@ fn drain_responses_sse_frames(
                 continue;
             }
             if let Ok(json) = serde_json::from_slice::<Value>(data) {
+                // First content delta → upstream TTFT. `/v1/responses`
+                // reported none at all before this, so codex-class clients
+                // showed a blank figure.
+                if !*first_token_seen && is_responses_content_delta(&json) {
+                    *first_token_seen = true;
+                    acc.get_or_insert_with(Default::default).upstream_ttft_ms =
+                        attempt_started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                }
                 if let Some(u) = parse_responses_terminal_usage(&json) {
-                    *acc = Some(u);
+                    // The terminal frame replaces the token counters; carry
+                    // the two latency figures measured before it across.
+                    let (ttft, down) = acc
+                        .as_ref()
+                        .map(|a| (a.upstream_ttft_ms, a.downstream_latency_ms))
+                        .unwrap_or_default();
+                    *acc = Some(ResponseUsage {
+                        upstream_ttft_ms: ttft,
+                        downstream_latency_ms: down,
+                        ..u
+                    });
                 }
                 if let Some(c) = capture.as_deref_mut() {
                     c.observe(&json);
@@ -2384,6 +2443,10 @@ impl EosOutputScan {
 /// SSE wire shape.
 fn build_responses_passthrough_stream<S, F>(
     upstream: S,
+    // Request clock — what the CALLER waited for.
+    started: Instant,
+    // Attempt clock — how the UPSTREAM behaved.
+    attempt_started: Instant,
     content_cap: Option<u32>,
     eos_scan: Option<EosOutputScan>,
     on_complete: F,
@@ -2427,13 +2490,20 @@ where
         };
         futures::pin_mut!(upstream);
         let mut buf: Vec<u8> = Vec::new();
+        let mut first_token_seen = false;
         while let Some(item) = upstream.next().await {
             if let Ok(bytes) = &item {
                 // Side-channel parse: copy into the frame buffer (the original
                 // `bytes` is yielded unchanged below) and drain complete frames.
                 buf.extend_from_slice(bytes);
                 let (usage_acc, capture) = guard.parts();
-                drain_responses_sse_frames(&mut buf, usage_acc, capture);
+                drain_responses_sse_frames(
+                    &mut buf,
+                    usage_acc,
+                    capture,
+                    attempt_started,
+                    &mut first_token_seen,
+                );
                 // Bound the frame buffer: the happy path drains complete frames
                 // above so `buf` only holds a partial trailing frame. A
                 // non-conformant upstream streaming bytes without a blank-line
@@ -2450,6 +2520,15 @@ where
                 }
             }
             // Forward the original item verbatim (Ok bytes OR a mid-stream Err).
+            // The first successful forward is what the caller waited for.
+            if item.is_ok() {
+                let (usage_acc, _) = guard.parts();
+                let acc = usage_acc.get_or_insert_with(Default::default);
+                if acc.downstream_latency_ms == 0 {
+                    acc.downstream_latency_ms =
+                        started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                }
+            }
             yield item;
         }
         // Clean end-of-stream: run the monitor observation (needs async, so
@@ -2686,7 +2765,9 @@ fn emit_usage_event(
         cache_creation_tokens: usage.cache_creation_tokens,
         cache_read_tokens: usage.cache_read_tokens,
         usage_estimated: usage.usage_estimated,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        upstream_ttft_ms: usage.upstream_ttft_ms,
+        downstream_latency_ms: usage.downstream_latency_ms,
         status_code,
         inbound_protocol: "openai".to_string(),
         attempt_index: attempt.index,
@@ -2769,7 +2850,7 @@ fn emit_zero_token_event(
         requested_model: requested_model.to_string(),
         redacted_entity_counts,
         guardrail_monitor_hits,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         status_code,
         inbound_protocol: "openai".to_string(),
         attempt_index: attempt.index,
diff --git a/crates/aisix-proxy/src/responses_bridge.rs b/crates/aisix-proxy/src/responses_bridge.rs
index a26d23fa..33fa738e 100644
--- a/crates/aisix-proxy/src/responses_bridge.rs
+++ b/crates/aisix-proxy/src/responses_bridge.rs
@@ -891,7 +891,11 @@ pub struct ResponsesStreamCompletion {
     pub cache_creation_tokens: u32,
     pub cache_read_tokens: u32,
     pub finish_reason: String,
-    pub ttft_ms: u32,
+    /// Attempt-scoped time to the upstream's first generated chunk.
+    pub upstream_ttft_ms: u32,
+    /// Request-scoped time until the caller got its first response bytes.
+    /// Trails `upstream_ttft_ms` by any hold-back guardrail scan.
+    pub downstream_latency_ms: u32,
     /// Set when an output guardrail blocked the streamed response (a content
     /// block or a fail-closed buffer overflow). The upstream still billed, so
     /// the usage event carries the tokens but is marked blocked — matching
@@ -983,7 +987,10 @@ impl<F: FnOnce(ResponsesStreamCompletion)> Drop for CompleteOnDrop<F> {
 pub fn build_responses_bridge_stream(
     upstream: ChatChunkStream,
     encoder: ResponsesSseEncoder,
+    // Request clock — what the CALLER waited for.
     started: Instant,
+    // Attempt clock — how the UPSTREAM behaved on this call.
+    attempt_started: Instant,
     output_guardrail: Option<Arc<aisix_guardrails::GuardrailChain>>,
     hold_back: bool,
     max_buffer_bytes: usize,
@@ -1004,6 +1011,16 @@ pub fn build_responses_bridge_stream(
             slot: Some((on_complete, ResponsesStreamCompletion::default())),
             estimator,
         };
+        // Stamped on the first bytes that actually leave for the client —
+        // under hold-back that is the release, not the upstream chunk.
+        macro_rules! downstream_mark {
+            () => {
+                if guard.comp().downstream_latency_ms == 0 {
+                    guard.comp().downstream_latency_ms =
+                        started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                }
+            };
+        }
         let mut upstream = upstream;
         let mut first_chunk_seen = false;
         let buffering = output_guardrail.is_some() && hold_back;
@@ -1019,8 +1036,8 @@ pub fn build_responses_bridge_stream(
                         && (chunk.delta.content.is_some() || chunk.delta.tool_calls.is_some())
                     {
                         first_chunk_seen = true;
-                        guard.comp().ttft_ms =
-                            started.elapsed().as_millis().min(u32::MAX as u128) as u32;
+                        guard.comp().upstream_ttft_ms =
+                            attempt_started.elapsed().as_millis().min(u32::MAX as u128) as u32;
                     }
                     {
                         let comp = guard.comp();
@@ -1087,6 +1104,7 @@ pub fn build_responses_bridge_stream(
                             }
                             held.push(b);
                         } else {
+                            downstream_mark!();
                             yield Ok::<_, std::io::Error>(b);
                         }
                     }
@@ -1116,6 +1134,7 @@ pub fn build_responses_bridge_stream(
                     }
                     held.push(b);
                 } else {
+                    downstream_mark!();
                     yield Ok(b);
                 }
             }
@@ -1282,12 +1301,14 @@ pub fn build_responses_bridge_stream(
                     &mut guard.comp().redacted_entity_counts,
                     counts,
                 );
+                downstream_mark!();
                 yield Ok(bytes::Bytes::from(rewritten));
                 return;
             }
         }
         // Release the held events verbatim.
         for b in held {
+            downstream_mark!();
             yield Ok(b);
         }
     };
diff --git a/crates/aisix-proxy/src/videos.rs b/crates/aisix-proxy/src/videos.rs
index 876b1384..be57108d 100644
--- a/crates/aisix-proxy/src/videos.rs
+++ b/crates/aisix-proxy/src/videos.rs
@@ -1938,7 +1938,10 @@ fn emit_submit_usage_event(
         api_key_id: api_key_id.to_string(),
         requested_model: requested_model.to_string(),
         status_code,
-        latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        // Single-attempt endpoint: the attempt spans the whole request, so
+        // the upstream figure and what the caller waited for coincide.
+        upstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
+        downstream_latency_ms: elapsed.as_millis().min(u32::MAX as u128) as u32,
         inbound_protocol: "openai".to_string(),
         applied_guardrails: applied_guardrails.to_vec(),
         guardrail_monitor_hits,
diff --git a/crates/aisix-server/src/telemetry.rs b/crates/aisix-server/src/telemetry.rs
index 1da2e787..4b9ad283 100644
--- a/crates/aisix-server/src/telemetry.rs
+++ b/crates/aisix-server/src/telemetry.rs
@@ -312,7 +312,7 @@ mod tests {
             api_key_id: "ak-uuid".into(),
             prompt_tokens: 10,
             completion_tokens: 20,
-            latency_ms: 30,
+            upstream_latency_ms: 30,
             status_code: 200,
             cost_usd: 0.001,
             guardrail_blocked: false,
diff --git a/tests/e2e/src/cases/latency-guardrail-holdback-e2e.test.ts b/tests/e2e/src/cases/latency-guardrail-holdback-e2e.test.ts
new file mode 100644
index 00000000..2bf55dee
--- /dev/null
+++ b/tests/e2e/src/cases/latency-guardrail-holdback-e2e.test.ts
@@ -0,0 +1,270 @@
+// E2E: a MASKING output guardrail puts the streamed response on the
+// hold-back path — nothing reaches the client until the whole response
+// scans clean. The caller therefore waits materially longer than the
+// upstream took to produce its first chunk, and the two figures must
+// show that:
+//
+//   upstream_ttft_ms      unchanged — the guardrail does not slow the
+//                         upstream leg down.
+//   downstream_latency_ms covers the whole wait, scan included.
+//
+// Measuring the caller-facing figure where the upstream chunk arrives
+// (rather than where bytes are handed to the client) collapses it to the
+// first number and hides the guardrail's cost entirely.
+//
+// This lives apart from `latency-upstream-downstream-e2e` because the
+// guardrail is env-scoped: seeding it in that file would silently push
+// its other cases onto the hold-back path too.
+
+import { createHash } from "node:crypto";
+import { createServer, type Server } from "node:http";
+import { afterAll, beforeAll, describe, expect, test } from "vitest";
+import {
+  EtcdClient,
+  SeedClient,
+  pickFreePort,
+  spawnApp,
+  startOpenAiUpstream,
+  waitConfigPropagation,
+  type OpenAiUpstream,
+  type SpawnedApp,
+} from "../harness/index.js";
+
+// E2E: the UsageEvent carries two latency families measured against
+// different clocks (see `UsageEvent` in aisix-obs/src/usage.rs):
+//
+//   upstream_ttft_ms      attempt-scoped — when the UPSTREAM produced its
+//                         first generated chunk.
+//   downstream_latency_ms request-scoped — when the CALLER got its first
+//                         usable bytes. Includes everything the gateway
+//                         did in between.
+//
+// Two properties pin the split:
+//
+//  1. With no output guardrail the gateway forwards chunks straight
+//     through, so the two figures nearly coincide.
+// The hold-back counterpart — where a masking output guardrail makes the
+// two diverge — needs an env-scoped guardrail, which would leak into
+// every later request here, so it lives in
+// `latency-guardrail-holdback-e2e.test.ts` with its own gateway.
+//
+// It also covers a bug fixed alongside: `/v1/responses` never recorded a
+// TTFT at all, so codex-class clients showed a blank figure.
+
+const CALLER_PLAINTEXT = "sk-latency-holdback-caller";
+const CALLER_KEY_HASH = createHash("sha256")
+  .update(CALLER_PLAINTEXT)
+  .digest("hex");
+
+/** Inter-chunk gap; the hold-back case pays the whole stream before releasing. */
+const CHUNK_GAP_MS = 250;
+const CHUNK_COUNT = 4;
+/** Total time the upstream spends streaming after its first chunk. */
+const STREAM_TAIL_MS = CHUNK_GAP_MS * (CHUNK_COUNT - 1);
+
+interface OtlpReceiver {
+  url: string;
+  spans: Array<Record<string, string>>;
+  close(): Promise<void>;
+}
+
+async function startOtlpReceiver(): Promise<OtlpReceiver> {
+  const spans: Array<Record<string, string>> = [];
+  const server: Server = createServer((req, res) => {
+    let raw = "";
+    req.on("data", (c: Buffer) => (raw += c.toString("utf8")));
+    req.on("end", () => {
+      try {
+        const body = JSON.parse(raw);
+        for (const rs of body.resourceSpans ?? []) {
+          for (const ss of rs.scopeSpans ?? []) {
+            for (const span of ss.spans ?? []) {
+              const attrs: Record<string, string> = {};
+              for (const a of span.attributes ?? []) {
+                const v = a.value ?? {};
+                attrs[a.key] =
+                  v.stringValue ?? String(v.intValue ?? v.boolValue ?? "");
+              }
+              spans.push(attrs);
+            }
+          }
+        }
+      } catch {
+        // ignore malformed bodies — assertions fail on missing spans
+      }
+      res.statusCode = 200;
+      res.end("{}");
+    });
+  });
+  const port = await pickFreePort();
+  await new Promise<void>((resolve) => server.listen(port, "127.0.0.1", resolve));
+  return {
+    url: `http://127.0.0.1:${port}/v1/traces`,
+    spans,
+    async close() {
+      await new Promise<void>((resolve, reject) => {
+        server.close((err) => (err ? reject(err) : resolve()));
+      });
+    },
+  };
+}
+
+async function waitForSpan(
+  recv: OtlpReceiver,
+  requestId: string,
+  timeoutMs = 10_000,
+): Promise<Record<string, string>> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    const hit = recv.spans.find((a) => a["aisix.request_id"] === requestId);
+    if (hit) return hit;
+    await new Promise((r) => setTimeout(r, 50));
+  }
+  throw new Error(`no usage span for request_id=${requestId}`);
+}
+
+/** OpenAI-shape SSE chunks whose text is benign (nothing for the guardrail to mask). */
+function chatChunks(): string[] {
+  const events = Array.from({ length: CHUNK_COUNT }, (_, i) =>
+    JSON.stringify({
+      id: "chatcmpl-latency-split",
+      object: "chat.completion.chunk",
+      created: 1,
+      model: "gpt-4o-mini",
+      choices: [{ index: 0, delta: { content: `part${i} ` }, finish_reason: null }],
+    }),
+  );
+  events.push(
+    JSON.stringify({
+      id: "chatcmpl-latency-split",
+      object: "chat.completion.chunk",
+      created: 1,
+      model: "gpt-4o-mini",
+      choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
+      usage: { prompt_tokens: 5, completion_tokens: 8, total_tokens: 13 },
+    }),
+    "[DONE]",
+  );
+  return events;
+}
+
+describe("streamed latency under a hold-back output guardrail", () => {
+  let etcdReachable = false;
+  let app: SpawnedApp | undefined;
+  let seed: SeedClient | undefined;
+  let otlp: OtlpReceiver | undefined;
+  const upstreams: OpenAiUpstream[] = [];
+
+  beforeAll(async () => {
+    const etcd = new EtcdClient();
+    etcdReachable = await etcd.ping();
+    if (!etcdReachable) return;
+    app = await spawnApp();
+    seed = new SeedClient(etcd, app.etcdPrefix);
+    otlp = await startOtlpReceiver();
+    await seed.createObservabilityExporter({
+      name: "latency-split-otlp",
+      enabled: true,
+      kind: "otlp_http",
+      endpoint: otlp.url,
+    });
+    await seed.createApiKey({
+      key_hash: CALLER_KEY_HASH,
+      allowed_models: ["*"],
+    });
+  });
+
+  afterAll(async () => {
+    await app?.exit();
+    await Promise.all(upstreams.map((u) => u.close()));
+    await otlp?.close();
+  });
+
+  async function createModel(
+    displayName: string,
+    upstream: OpenAiUpstream,
+  ): Promise<void> {
+    if (!seed) throw new Error("seed client not initialized");
+    const pk = await seed.createProviderKey({
+      display_name: `${displayName}-pk`,
+      secret: "sk-mock",
+      api_base: `${upstream.baseUrl}/v1`,
+    });
+    await seed.createModel({
+      display_name: displayName,
+      provider: "openai",
+      model_name: "gpt-4o-mini",
+      provider_key_id: pk.id,
+    });
+  }
+
+  /** Seed a throwaway key AFTER the config under test, then poll until it authenticates. */
+  async function awaitPropagation(tag: string): Promise<void> {
+    const canary = `sk-canary-${tag}-${Date.now()}`;
+    await seed!.createApiKey({
+      key_hash: createHash("sha256").update(canary).digest("hex"),
+      allowed_models: ["*"],
+    });
+    await waitConfigPropagation(async () => {
+      const res = await fetch(`${app!.proxyUrl}/v1/models`, {
+        headers: { authorization: `Bearer ${canary}` },
+      });
+      return res.status === 200;
+    });
+  }
+
+  test("a masking output guardrail holds the stream back, and only the downstream figure shows it", async (ctx) => {
+    if (!etcdReachable || !app || !seed || !otlp) {
+      ctx.skip();
+      return;
+    }
+
+    const upstream = await startOpenAiUpstream({
+      streamEvents: chatChunks(),
+      eventDelayMs: CHUNK_GAP_MS,
+    });
+    upstreams.push(upstream);
+    await createModel("latency-masked", upstream);
+    // A masking detector puts the streamed-output policy into whole-response
+    // hold-back: nothing reaches the client until the scan clears.
+    await seed.createGuardrail({
+      name: "latency-split-mask",
+      enabled: true,
+      hook_point: "output",
+      kind: "pii",
+      detectors: [{ type: "email", action: "mask" }],
+    });
+    await awaitPropagation("masked");
+
+    const res = await fetch(`${app.proxyUrl}/v1/chat/completions`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${CALLER_PLAINTEXT}`,
+        "content-type": "application/json",
+      },
+      body: JSON.stringify({
+        model: "latency-masked",
+        messages: [{ role: "user", content: "stream please" }],
+        stream: true,
+      }),
+    });
+    expect(res.status).toBe(200);
+    const requestId = res.headers.get("x-aisix-call-id");
+    expect(requestId).toBeTruthy();
+    await res.text();
+
+    const span = await waitForSpan(otlp, requestId!);
+    const upstreamTtft = Number(span["aisix.upstream_ttft_ms"]);
+    const downstream = Number(span["aisix.downstream_latency_ms"]);
+
+    // The upstream still delivered its first chunk promptly — the guardrail
+    // does not slow the upstream leg down.
+    expect(upstreamTtft).toBeLessThan(STREAM_TAIL_MS);
+    // But the caller waited for the entire stream plus the scan. Measuring
+    // this off the upstream chunk (as the pre-split telemetry did) would
+    // report the small figure above and hide the guardrail's cost.
+    expect(downstream).toBeGreaterThanOrEqual(STREAM_TAIL_MS);
+    expect(downstream).toBeGreaterThan(upstreamTtft);
+  });
+
+});
diff --git a/tests/e2e/src/cases/latency-upstream-downstream-e2e.test.ts b/tests/e2e/src/cases/latency-upstream-downstream-e2e.test.ts
new file mode 100644
index 00000000..eb2b35a9
--- /dev/null
+++ b/tests/e2e/src/cases/latency-upstream-downstream-e2e.test.ts
@@ -0,0 +1,292 @@
+import { createHash } from "node:crypto";
+import { createServer, type Server } from "node:http";
+import { afterAll, beforeAll, describe, expect, test } from "vitest";
+import {
+  EtcdClient,
+  SeedClient,
+  pickFreePort,
+  spawnApp,
+  startOpenAiUpstream,
+  waitConfigPropagation,
+  type OpenAiUpstream,
+  type SpawnedApp,
+} from "../harness/index.js";
+
+// E2E: the UsageEvent carries two latency families measured against
+// different clocks (see `UsageEvent` in aisix-obs/src/usage.rs):
+//
+//   upstream_ttft_ms      attempt-scoped — when the UPSTREAM produced its
+//                         first generated chunk.
+//   downstream_latency_ms request-scoped — when the CALLER got its first
+//                         usable bytes. Includes everything the gateway
+//                         did in between.
+//
+// Two properties pin the split:
+//
+//  1. With no output guardrail the gateway forwards chunks straight
+//     through, so the two figures nearly coincide.
+// The hold-back counterpart — where a masking output guardrail makes the
+// two diverge — needs an env-scoped guardrail, which would leak into
+// every later request here, so it lives in
+// `latency-guardrail-holdback-e2e.test.ts` with its own gateway.
+//
+// It also covers a bug fixed alongside: `/v1/responses` never recorded a
+// TTFT at all, so codex-class clients showed a blank figure.
+
+const CALLER_PLAINTEXT = "sk-latency-split-caller";
+const CALLER_KEY_HASH = createHash("sha256")
+  .update(CALLER_PLAINTEXT)
+  .digest("hex");
+
+/** Inter-chunk gap; the hold-back case pays the whole stream before releasing. */
+const CHUNK_GAP_MS = 250;
+const CHUNK_COUNT = 4;
+/** Total time the upstream spends streaming after its first chunk. */
+const STREAM_TAIL_MS = CHUNK_GAP_MS * (CHUNK_COUNT - 1);
+
+interface OtlpReceiver {
+  url: string;
+  spans: Array<Record<string, string>>;
+  close(): Promise<void>;
+}
+
+async function startOtlpReceiver(): Promise<OtlpReceiver> {
+  const spans: Array<Record<string, string>> = [];
+  const server: Server = createServer((req, res) => {
+    let raw = "";
+    req.on("data", (c: Buffer) => (raw += c.toString("utf8")));
+    req.on("end", () => {
+      try {
+        const body = JSON.parse(raw);
+        for (const rs of body.resourceSpans ?? []) {
+          for (const ss of rs.scopeSpans ?? []) {
+            for (const span of ss.spans ?? []) {
+              const attrs: Record<string, string> = {};
+              for (const a of span.attributes ?? []) {
+                const v = a.value ?? {};
+                attrs[a.key] =
+                  v.stringValue ?? String(v.intValue ?? v.boolValue ?? "");
+              }
+              spans.push(attrs);
+            }
+          }
+        }
+      } catch {
+        // ignore malformed bodies — assertions fail on missing spans
+      }
+      res.statusCode = 200;
+      res.end("{}");
+    });
+  });
+  const port = await pickFreePort();
+  await new Promise<void>((resolve) => server.listen(port, "127.0.0.1", resolve));
+  return {
+    url: `http://127.0.0.1:${port}/v1/traces`,
+    spans,
+    async close() {
+      await new Promise<void>((resolve, reject) => {
+        server.close((err) => (err ? reject(err) : resolve()));
+      });
+    },
+  };
+}
+
+async function waitForSpan(
+  recv: OtlpReceiver,
+  requestId: string,
+  timeoutMs = 10_000,
+): Promise<Record<string, string>> {
+  const deadline = Date.now() + timeoutMs;
+  while (Date.now() < deadline) {
+    const hit = recv.spans.find((a) => a["aisix.request_id"] === requestId);
+    if (hit) return hit;
+    await new Promise((r) => setTimeout(r, 50));
+  }
+  throw new Error(`no usage span for request_id=${requestId}`);
+}
+
+/** OpenAI-shape SSE chunks whose text is benign (nothing for the guardrail to mask). */
+function chatChunks(): string[] {
+  const events = Array.from({ length: CHUNK_COUNT }, (_, i) =>
+    JSON.stringify({
+      id: "chatcmpl-latency-split",
+      object: "chat.completion.chunk",
+      created: 1,
+      model: "gpt-4o-mini",
+      choices: [{ index: 0, delta: { content: `part${i} ` }, finish_reason: null }],
+    }),
+  );
+  events.push(
+    JSON.stringify({
+      id: "chatcmpl-latency-split",
+      object: "chat.completion.chunk",
+      created: 1,
+      model: "gpt-4o-mini",
+      choices: [{ index: 0, delta: {}, finish_reason: "stop" }],
+      usage: { prompt_tokens: 5, completion_tokens: 8, total_tokens: 13 },
+    }),
+    "[DONE]",
+  );
+  return events;
+}
+
+describe("upstream vs downstream latency split", () => {
+  let etcdReachable = false;
+  let app: SpawnedApp | undefined;
+  let seed: SeedClient | undefined;
+  let otlp: OtlpReceiver | undefined;
+  const upstreams: OpenAiUpstream[] = [];
+
+  beforeAll(async () => {
+    const etcd = new EtcdClient();
+    etcdReachable = await etcd.ping();
+    if (!etcdReachable) return;
+    app = await spawnApp();
+    seed = new SeedClient(etcd, app.etcdPrefix);
+    otlp = await startOtlpReceiver();
+    await seed.createObservabilityExporter({
+      name: "latency-split-otlp",
+      enabled: true,
+      kind: "otlp_http",
+      endpoint: otlp.url,
+    });
+    await seed.createApiKey({
+      key_hash: CALLER_KEY_HASH,
+      allowed_models: ["*"],
+    });
+  });
+
+  afterAll(async () => {
+    await app?.exit();
+    await Promise.all(upstreams.map((u) => u.close()));
+    await otlp?.close();
+  });
+
+  async function createModel(
+    displayName: string,
+    upstream: OpenAiUpstream,
+  ): Promise<void> {
+    if (!seed) throw new Error("seed client not initialized");
+    const pk = await seed.createProviderKey({
+      display_name: `${displayName}-pk`,
+      secret: "sk-mock",
+      api_base: `${upstream.baseUrl}/v1`,
+    });
+    await seed.createModel({
+      display_name: displayName,
+      provider: "openai",
+      model_name: "gpt-4o-mini",
+      provider_key_id: pk.id,
+    });
+  }
+
+  /** Seed a throwaway key AFTER the config under test, then poll until it authenticates. */
+  async function awaitPropagation(tag: string): Promise<void> {
+    const canary = `sk-canary-${tag}-${Date.now()}`;
+    await seed!.createApiKey({
+      key_hash: createHash("sha256").update(canary).digest("hex"),
+      allowed_models: ["*"],
+    });
+    await waitConfigPropagation(async () => {
+      const res = await fetch(`${app!.proxyUrl}/v1/models`, {
+        headers: { authorization: `Bearer ${canary}` },
+      });
+      return res.status === 200;
+    });
+  }
+
+  test("without an output guardrail the two figures nearly coincide", async (ctx) => {
+    if (!etcdReachable || !app || !seed || !otlp) {
+      ctx.skip();
+      return;
+    }
+
+    const upstream = await startOpenAiUpstream({
+      streamEvents: chatChunks(),
+      eventDelayMs: CHUNK_GAP_MS,
+    });
+    upstreams.push(upstream);
+    await createModel("latency-plain", upstream);
+    await awaitPropagation("plain");
+
+    const res = await fetch(`${app.proxyUrl}/v1/chat/completions`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${CALLER_PLAINTEXT}`,
+        "content-type": "application/json",
+      },
+      body: JSON.stringify({
+        model: "latency-plain",
+        messages: [{ role: "user", content: "stream please" }],
+        stream: true,
+      }),
+    });
+    expect(res.status).toBe(200);
+    const requestId = res.headers.get("x-aisix-call-id");
+    expect(requestId).toBeTruthy();
+    await res.text();
+
+    const span = await waitForSpan(otlp, requestId!);
+    const upstreamTtft = Number(span["aisix.upstream_ttft_ms"]);
+    const downstream = Number(span["aisix.downstream_latency_ms"]);
+
+    expect(Number.isFinite(upstreamTtft)).toBe(true);
+    expect(Number.isFinite(downstream)).toBe(true);
+    // Live-forward: the client gets the first chunk as it arrives, so the
+    // caller-facing figure sits just above the upstream's TTFT — and well
+    // below the point where the whole stream has finished.
+    expect(downstream).toBeGreaterThanOrEqual(upstreamTtft);
+    expect(downstream - upstreamTtft).toBeLessThan(STREAM_TAIL_MS);
+  });
+
+  test("/v1/responses streaming records a TTFT (it previously reported none)", async (ctx) => {
+    if (!etcdReachable || !app || !seed || !otlp) {
+      ctx.skip();
+      return;
+    }
+
+    const upstream = await startOpenAiUpstream({
+      streamEvents: [
+        JSON.stringify({ type: "response.created", response: { id: "resp_lat" } }),
+        JSON.stringify({ type: "response.output_text.delta", delta: "hello " }),
+        JSON.stringify({ type: "response.output_text.delta", delta: "there" }),
+        JSON.stringify({
+          type: "response.completed",
+          response: {
+            id: "resp_lat",
+            status: "completed",
+            usage: { input_tokens: 6, output_tokens: 9 },
+          },
+        }),
+        "[DONE]",
+      ],
+      eventDelayMs: CHUNK_GAP_MS,
+    });
+    upstreams.push(upstream);
+    await createModel("latency-responses", upstream);
+    await awaitPropagation("responses");
+
+    const res = await fetch(`${app.proxyUrl}/v1/responses`, {
+      method: "POST",
+      headers: {
+        authorization: `Bearer ${CALLER_PLAINTEXT}`,
+        "content-type": "application/json",
+      },
+      body: JSON.stringify({
+        model: "latency-responses",
+        input: "measure ttft",
+        stream: true,
+      }),
+    });
+    expect(res.status).toBe(200);
+    const requestId = res.headers.get("x-aisix-request-id");
+    expect(requestId).toBeTruthy();
+    await res.text();
+
+    const span = await waitForSpan(otlp, requestId!);
+    // The regression: this attribute was absent entirely on /v1/responses.
+    expect(span["aisix.upstream_ttft_ms"]).toBeDefined();
+    expect(Number(span["aisix.upstream_ttft_ms"])).toBeGreaterThan(0);
+    expect(Number(span["aisix.downstream_latency_ms"])).toBeGreaterThan(0);
+  });
+});