fix(telemetry): write a streamed request's access-log line when the stream ends, not when its head does

crates/aisix-obs/src/access_log.rs

@@ -5,46 +5,52 @@
 //!
 //! # When the line is written, and what that costs
 //!
-//! WHEN differs by path, and it decides which fields can be filled at all.
-//! Four cases, and only the first is "at the end of the request":
+//! Exactly one line per request, whatever the outcome — but WHEN it is
+//! written differs by path, and that decides which fields can be filled at
+//! all. Four cases:
 //!
 //! - **Non-streamed response** — from the handler, on its way out, with
 //!   everything it resolved available.
-//! - **Streamed response** — from the handler too, but when the SSE body is
-//!   handed to the server, BEFORE a single frame is polled. The upstream has
-//!   produced nothing yet, so the token counts and `provider_request_id` are
-//!   necessarily absent, and `status` is the response-OPEN status: a stream
-//!   that later aborts, or whose consumer walks away, still logged `200`.
-//!   `latency` is time-to-first-token for the same reason, NOT how long the
-//!   stream ran — the two differ by the whole length of the stream, which
-//!   for an LLM is routinely minutes. A stream's real end is only on its
-//!   `UsageEvent`; reading this line as the end of the request is how a
-//!   long-running stream gets mistaken for a connection sitting idle
-//!   (AISIX-Cloud#1394).
+//! - **Streamed response** — NOT when the SSE head is handed to the server.
+//!   The handler defers the line to the request's attribution cell
+//!   (`attribution::defer_access_log`) and it goes out beside the request's
+//!   TERMINAL usage event, at the point the stream's outcome is known:
+//!   fully consumed, abandoned mid-stream, or dropped before its first
+//!   poll. It therefore reports the same `status`, `error_kind` and `error`
+//!   as that event — a stream whose consumer walked away reads `499` /
+//!   `client_disconnected` on both — and it can carry the token counts and
+//!   `provider_request_id`, which only exist once the upstream has answered
+//!   (AISIX-Cloud#1571).
 //! - **`/v1/realtime`** — the opposite extreme. The handler returns the
 //!   WebSocket upgrade immediately; the line is written by `run_session` on
 //!   a detached task once the session closes, so it carries the close status
 //!   and the session's real token totals.
-//! - **Caller hung up before the response was delivered** — written from
+//! - **Caller hung up before the response head was written** — from
 //!   `ClientCancelGuard::drop`, with no handler involved. Status is `499`,
 //!   and the fields it can fill are the ones the request published to its
 //!   attribution cell as it resolved: `model`, `provider`, and the
 //!   dispatched target (`upstream_model` + `provider_key_id`). The
 //!   handler-side figures — tokens, `provider_request_id`, the routing
 //!   counts — stay `None`, because the future was dropped before it could
 //!   produce them. Such a request also emits a `499` usage event carrying
-//!   the same identities (AISIX-Cloud#1571), keyed by this `request_id`.
-//!   Only the no-response-head case writes one: a request whose head DID go
-//!   out already has its handler's line, and a second one under a second
-//!   status would make one request read as two. That request's `499` lives
-//!   on its usage event alone.
+//!   the same identities, keyed by this `request_id`.
 //!
 //! So do not add a field whose value only exists once the upstream has
 //! responded and expect it on every line: it is silently empty on the
-//! streamed and cancelled ones. A streamed request's completion-time figures
-//! live on the per-attempt `UsageEvent` (and, for the provider response id,
-//! on the `provider call completed` line `UsageSink::try_emit` writes),
-//! keyed by the same `request_id`.
+//! cancelled ones, where the request never got that far.
+//!
+//! # `latency` and `duration` answer two different questions
+//!
+//! - `latency_ms` is what the CALLER waited for: the first token forwarded
+//!   downstream on a streamed response, the complete response on a buffered
+//!   one. It is the same figure the request's terminal `UsageEvent` reports
+//!   as `downstream_latency_ms`. Deliberately not the length of the stream
+//!   (AISIX-Cloud#1394) — reading a minutes-long stream's wait as its
+//!   time-to-first-token is what makes a working stream look like a
+//!   connection sitting idle.
+//! - `duration_ms` is how long the request occupied the gateway, arrival to
+//!   last byte out. On a non-streamed request the two coincide; on a
+//!   streamed one they differ by the whole length of the stream.
 
 use std::time::Duration;
 
@@ -58,7 +64,22 @@ pub struct AccessLog<'a> {
     pub method: &'a str,
     pub path: &'a str,
     pub status: u16,
+    /// What the caller waited for — see the module docs. On a streamed
+    /// response this is time-to-first-token, NOT how long the stream ran.
     pub latency: Duration,
+    /// How long the request occupied the gateway: from arrival to the point
+    /// this line is written. Equal to `latency` on everything that is not
+    /// streamed.
+    ///
+    /// "The point this line is written" is the request's end on every
+    /// surface whose record is written at completion — which is all of them
+    /// except the two that meter at their handler tail and relay an
+    /// open-ended body afterwards (`/v1/audio/speech`, billed per input
+    /// character, and `/v1/videos/{id}/content`, metered by the
+    /// submission). Those two have no completion-time emitter to carry a
+    /// line, so theirs ends at the response head and does not span the
+    /// relay.
+    pub duration: Duration,
     pub provider: Option<&'a str>,
     /// The model name the CALLER addressed — for a routing group, the group
     /// itself, never the target it dispatched to. See `upstream_model`
@@ -85,13 +106,14 @@ pub struct AccessLog<'a> {
     ///
     /// `None` whenever no id exists by the time this line is written:
     /// the request never reached an upstream (guardrail block,
-    /// pre-dispatch error), it was served from cache, the endpoint's
+    /// pre-dispatch error), it was served from cache, or the endpoint's
     /// provider response carries no id at all (embeddings / audio /
-    /// images / count_tokens), or the response is **streamed** — there the
-    /// id arrives in the first frame, after this line. Streamed and
-    /// mid-stream-failed-over calls are covered instead by the per-attempt
-    /// `provider call completed` line (see `UsageSink::try_emit`), which
-    /// shares this `request_id`.
+    /// images / count_tokens). A **streamed** response does carry it —
+    /// the id arrives in the first frame and the line is written at the
+    /// stream's end (AISIX-Cloud#1571) — unless the caller walked away
+    /// before that frame. Mid-stream-failed-over calls are covered by the
+    /// per-attempt `provider call completed` line (see
+    /// `UsageSink::try_emit`), which shares this `request_id`.
     pub provider_request_id: Option<&'a str>,
     /// Routing target that ultimately served the request (the winning
     /// attempt's display name). `None` for direct models / cache hits.
@@ -147,6 +169,7 @@ impl AccessLog<'_> {
             path = self.path,
             status = self.status,
             latency_ms = self.latency.as_millis() as u64,
+            duration_ms = self.duration.as_millis() as u64,
             provider = self.provider,
             model = self.model,
             upstream_model = self.upstream_model,
@@ -220,6 +243,7 @@ mod tests {
                 path: "/v1/chat/completions",
                 status: 200,
                 latency: Duration::from_millis(42),
+                duration: Duration::from_millis(9_000),
                 provider: Some("openai"),
                 model: Some("my-gpt4"),
                 upstream_model: Some("gpt-4o"),
@@ -245,6 +269,12 @@ mod tests {
         assert!(out.contains("method=\"POST\"") || out.contains("method=POST"));
         assert!(out.contains("status=200"));
         assert!(out.contains("latency_ms=42"));
+        // AISIX-Cloud#1571: the two figures are separate fields because on
+        // a streamed line they are separate questions — what the caller
+        // waited for, and how long the request held the gateway. This one
+        // is deliberately the longer of the two, so transposing them at the
+        // emit site cannot pass.
+        assert!(out.contains("duration_ms=9000"), "{out}");
         assert!(out.contains("provider=\"openai\"") || out.contains("provider=openai"));
         assert!(out.contains("total_tokens=3"));
         assert!(out.contains("request_id=\"req-abc\"") || out.contains("request_id=req-abc"));
@@ -297,6 +327,7 @@ mod tests {
                 path: "/v1/messages",
                 status: 504,
                 latency: Duration::from_millis(7167),
+                duration: Duration::from_millis(7167),
                 provider: None,
                 model: Some("claude-sonnet-4"),
                 upstream_model: None,
@@ -349,6 +380,7 @@ mod tests {
                 path: "/v1/chat/completions",
                 status: 401,
                 latency: Duration::from_millis(1),
+                duration: Duration::from_millis(1),
                 provider: None,
                 model: None,
                 upstream_model: None,

crates/aisix-proxy/AGENTS.md

@@ -200,6 +200,20 @@ nothing, and nothing errors: the caller gets a correct status while the gateway
 keeps no record of the request, which is indistinguishable from the request never
 arriving.
 
+One exception, and it is the whole of it: **a STREAMED response's line is not
+the handler's to write.** Its tail runs when the head goes out, which is not
+when the request ends — so it parks the line on the attribution cell
+(`attribution::PendingAccessLog`) and the line goes out from
+`usage_attr::emit_usage` with the request's TERMINAL usage event, whichever of
+the stream's endings produced it. That is what makes the line and the row agree
+on `status`, `error_class` and `error_message` by construction; writing the line
+at the tail instead reported every abandoned stream as a `200` beside its own
+`499` row (AISIX-Cloud#1571). Two consequences for a new streaming family: park
+the line whenever the response IS a stream (a "telemetry already emitted" flag
+is NOT the same predicate — chat's buffered ensemble sets one), and make sure
+the stream really does emit a terminal usage event, because that emit is now
+the only thing that writes the line.
+
 Two shapes give up early, and both must answer through
 `reject::reject_before_dispatch` (it renders the envelope *and* emits the
 telemetry, so the two can't drift apart):

crates/aisix-proxy/src/a2a.rs

@@ -164,35 +164,53 @@ pub async fn a2a_endpoint(
 
     let elapsed = started.elapsed();
     let status = response.status().as_u16();
-    let target = crate::attribution::AccessLogTarget::current();
-    AccessLog {
-        method: http_method.as_str(),
-        path: "/a2a",
-        status,
-        latency: elapsed,
-        provider: Some("a2a"),
-        model: None,
-        upstream_model: target.upstream_model(),
-        provider_key_id: target.provider_key_id(),
-        api_key_id: Some(&api_key_id),
-        // Counted inside `dispatch`, which hands back only a rendered
-        // `Response` — and for a stream, not until its drop guard fires, long
-        // after this line. The usage event carries them.
-        prompt_tokens: None,
-        completion_tokens: None,
-        total_tokens: None,
-        request_id: &request_id,
-        // Same as `/mcp`: `dispatch` returns an already-rendered `Response`,
-        // so no typed error reaches this point.
-        error_kind: None,
-        error: None,
-        provider_request_id: None,
-        served_by_model: None,
-        routing_attempt_count: None,
-        routing_fallback_count: None,
-        mcp: None,
+    if crate::attribution::stream_owns_access_log() {
+        // A streamed call ends when the agent's last event is relayed or
+        // the caller walks away, both of which are below this frame and
+        // minutes away. Park the line; `StreamUsageOnDrop` writes it beside
+        // the usage event that already reports that ending
+        // (AISIX-Cloud#1571).
+        crate::attribution::defer_access_log(
+            crate::attribution::PendingAccessLog::new(
+                http_method.as_str(),
+                "/a2a",
+                &request_id,
+                &api_key_id,
+                started,
+            )
+            .with_model("a2a", ""),
+        );
+    } else {
+        let target = crate::attribution::AccessLogTarget::current();
+        AccessLog {
+            method: http_method.as_str(),
+            path: "/a2a",
+            status,
+            latency: elapsed,
+            duration: elapsed,
+            provider: Some("a2a"),
+            model: None,
+            upstream_model: target.upstream_model(),
+            provider_key_id: target.provider_key_id(),
+            api_key_id: Some(&api_key_id),
+            // Counted inside `dispatch`, which hands back only a rendered
+            // `Response`. The usage event carries them.
+            prompt_tokens: None,
+            completion_tokens: None,
+            total_tokens: None,
+            request_id: &request_id,
+            // Same as `/mcp`: `dispatch` returns an already-rendered
+            // `Response`, so no typed error reaches this point.
+            error_kind: None,
+            error: None,
+            provider_request_id: None,
+            served_by_model: None,
+            routing_attempt_count: None,
+            routing_fallback_count: None,
+            mcp: None,
+        }
+        .emit();
     }
-    .emit();
     crate::request_metrics::record(
         &state,
         "/a2a",
@@ -631,6 +649,12 @@ async fn dispatch_stream(
         drop(guard);
     });
 
+    // The endpoint tail below owns this request's access-log line, and from
+    // there the response is opaque — it cannot tell this stream from a
+    // rendered error. Say so here, so the tail parks the line for the guard
+    // above to write at the call's real end (AISIX-Cloud#1571).
+    crate::attribution::note_stream_owns_access_log();
+
     let mut response = axum::response::Sse::new(sse);
     if let Some(interval) = crate::sse_keepalive::interval() {
         response = response.keep_alive(axum::response::sse::KeepAlive::new().interval(interval));
@@ -2142,4 +2166,41 @@ mod tests {
         assert_eq!(card["version"], "2.1.0");
         assert_eq!(card["skills"][0]["id"], "extract");
     }
+
+    /// A streamed `/a2a` call writes ONE access-log line, at the task's end
+    /// rather than when the head went out, so each of the three endings
+    /// reports its own outcome (AISIX-Cloud#1571). The tail that writes this
+    /// family's line sees only an opaque `Response`, so the streaming branch
+    /// tells it to park the line instead.
+    ///
+    /// `latency_ms` is deliberately NOT asserted to be a time-to-first-event
+    /// here: an agent's stream of task updates is the call's product rather
+    /// than a delivery mechanism, so `/a2a` records the WHOLE stream as what
+    /// the caller waited for — and the line reports the same figure its
+    /// usage event does.
+    #[tokio::test]
+    async fn a_streamed_call_writes_one_line_per_stream_ending() {
+        let agent_url = spawn_progressing_stream_agent().await;
+        let handle = SnapshotHandle::new(snapshot_with(&agent_url, true, serde_json::json!(["*"])));
+        let hub = Arc::new(aisix_gateway::Hub::new());
+        let router = build_router(ProxyState::new(handle, hub, &proxy_cfg()).without_cache());
+
+        let endings = crate::test_log::three_stream_endings(router, || {
+            HttpRequest::post("/a2a/invoice")
+                .header("host", "gw.example.com")
+                .header("content-type", "application/json")
+                .header("authorization", format!("Bearer {TOKEN}"))
+                .body(Body::from(
+                    r#"{"jsonrpc":"2.0","id":"s","method":"message/stream"}"#,
+                ))
+                .unwrap()
+        })
+        .await;
+        crate::test_log::assert_one_line_per_ending(&endings, "/a2a", "ak-1");
+        assert_eq!(
+            endings.delivered.field("provider").as_deref(),
+            Some("a2a"),
+            "the line must keep naming the family it belongs to",
+        );
+    }
 }