From 24d8e61e9a748ac27001d165fbeb41ef08f95a98 Mon Sep 17 00:00:00 2001
From: Jarvis <jarvis@api7.ai>
Date: Tue, 11 Aug 2026 18:48:05 +0800
Subject: [PATCH 1/3] feat(a2a): observe how a streamed call ran, and meter
 calls per agent
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

A streamed A2A call recorded only what it ended with. How it got there —
how long the agent took to say anything, how much it said, whether the
caller was still listening — was not recorded anywhere, so the two ways a
stream disappoints could not be told apart: nothing arrived for a long
time, or plenty arrived and none of it advanced the task.

- `upstream_ttft_ms` is now stamped on the first relayed event, the same
  event the LLM paths stamp it on, so the agent's time-to-first-event and
  a model's time-to-first-token are the same measurement.
- `a2a_stream_event_count` records what the caller actually received.
- A stream the caller abandons mid-task is recorded as 499 rather than
  200, matching the LLM streaming paths. It had been landing in the same
  bucket as a completed one, so an agent nobody waits for looked healthy.

Adds the `aisix_a2a_*` family for the dimension the proxy families
cannot carry. They count `/a2a` by route only, so "is the invoices
agent's `message/stream` failing?" — the question an agent operator
actually asks — was unanswerable from metrics:

- `aisix_a2a_requests_total{agent, operation, status}`
- `aisix_a2a_ttfb_seconds{agent, operation}`
- `aisix_a2a_stream_events_total{agent, operation}`
- `aisix_a2a_task_state_total{agent, state}`

Every label is bounded by construction: a registered agent name, the
canonical operation set, the specification's task states. Task, context
and request ids stay out of labels entirely and live in logs and traces.

Request duration is deliberately NOT repeated — `aisix_proxy_request_
duration_seconds` already times `/a2a` end to end. The TTFB histogram
shares the existing TTFT bucket edges, and therefore its operator
override, rather than adding a config key that would have to be kept in
step with it.

Ref AISIX-Cloud#1215
---
 crates/aisix-obs/src/lib.rs                   |   6 +-
 crates/aisix-obs/src/metrics.rs               | 209 ++++++++++++++++++
 crates/aisix-obs/src/otlp_http_sink.rs        |  39 ++++
 crates/aisix-obs/src/usage.rs                 |  10 +
 crates/aisix-proxy/src/a2a.rs                 |  89 +++++++-
 .../cases/a2a-protocol-telemetry-e2e.test.ts  |  59 +++++
 tests/e2e/src/harness/upstream-a2a.ts         |   4 +
 7 files changed, 412 insertions(+), 4 deletions(-)

diff --git a/crates/aisix-obs/src/lib.rs b/crates/aisix-obs/src/lib.rs
index eb00938d6..c1cd79b97 100644
--- a/crates/aisix-obs/src/lib.rs
+++ b/crates/aisix-obs/src/lib.rs
@@ -29,9 +29,9 @@ use tracing_subscriber::{fmt, prelude::*, EnvFilter};
 
 pub use access_log::AccessLog;
 pub use metrics::{
-    client_type_from_user_agent, BudgetGauges, BudgetLabels, ClientTypeClassifier,
-    DeploymentLabels, DeploymentState, HistogramBuckets, LatencyLabels, LlmUsage, Metrics,
-    RequestLabels, RequestOutcome, UsageLabels,
+    client_type_from_user_agent, A2aCallOutcome, A2aLabels, BudgetGauges, BudgetLabels,
+    ClientTypeClassifier, DeploymentLabels, DeploymentState, HistogramBuckets, LatencyLabels,
+    LlmUsage, Metrics, RequestLabels, RequestOutcome, UsageLabels,
 };
 pub use otlp::{install_otlp_tracer, shutdown_otlp, OtlpError, OtlpHandle};
 pub use otlp_http_sink::{content_capture_cap, OtlpHttpFanOut, OtlpSink};
diff --git a/crates/aisix-obs/src/metrics.rs b/crates/aisix-obs/src/metrics.rs
index 1e586c004..890c449f7 100644
--- a/crates/aisix-obs/src/metrics.rs
+++ b/crates/aisix-obs/src/metrics.rs
@@ -210,6 +210,42 @@ pub const M_REQUEST_E2E_LATENCY_SECONDS: &str = "aisix_request_e2e_latency_secon
 /// [`M_REQUEST_E2E_LATENCY_SECONDS`] (with `streaming="true"` always).
 pub const M_REQUEST_TTFT_SECONDS: &str = "aisix_request_ttft_seconds";
 
+// ── A2A gateway series (AISIX-Cloud#1215) ──────────────────────────────────
+//
+// The `aisix_proxy_*` families already count `/a2a` traffic and time it, but
+// only by route: which agent was reached and which operation was invoked are
+// not labels there, so "is the invoice agent's `message/stream` failing?" —
+// the question an agent-platform operator actually asks — cannot be answered
+// from them. These four carry that dimension and nothing else, so they stay
+// bounded: `agent` is a registered resource name, `operation` the canonical
+// bounded set, `state` the specification's task states. Task, context and
+// JSON-RPC request ids are deliberately absent — they belong in logs and
+// traces, never in a label.
+//
+// Request DURATION is not repeated here: `aisix_proxy_request_duration_seconds`
+// already times `/a2a` end to end, and a second histogram of the same quantity
+// would only be a second thing to keep in sync.
+/// A2A calls by agent, canonical operation and status class. Counts the same
+/// requests as `aisix_proxy_requests_total{endpoint="/a2a"}`, split the way an
+/// agent operator needs.
+pub const M_A2A_REQUESTS_TOTAL: &str = "aisix_a2a_requests_total";
+/// Time from an A2A streaming call starting to the FIRST event the upstream
+/// agent pushed — the agent's own "time to first byte".
+///
+/// Named for the event rather than a token because an agent stream carries
+/// task updates, not tokens; it shares [`M_REQUEST_TTFT_SECONDS`]'s bucket
+/// edges (and therefore its `request_ttft` operator override) since the two
+/// measure the same shape of wait.
+pub const M_A2A_TTFB_SECONDS: &str = "aisix_a2a_ttfb_seconds";
+/// Events relayed downstream on A2A streaming calls. Divided by
+/// [`M_A2A_REQUESTS_TOTAL`] over the streaming operations it gives events per
+/// call — how chatty an agent is, and whether that changed.
+pub const M_A2A_STREAM_EVENTS_TOTAL: &str = "aisix_a2a_stream_events_total";
+/// A2A calls by the task state they ended on, normalized to the
+/// specification's set plus `unknown`. The series an operator watches for
+/// tasks piling up in `input-required` or `failed`.
+pub const M_A2A_TASK_STATE_TOTAL: &str = "aisix_a2a_task_state_total";
+
 // ── Config load-observability series (load-observability contract) ─────────
 // Reflected from [`aisix_core::ConfigMetricsView`] at scrape time via
 // [`Metrics::sync_config_status`]. Standard Prometheus config-reload naming so
@@ -649,6 +685,15 @@ impl Metrics {
                 &buckets.guardrail_latency,
             )
             .expect("bucket lists are validated non-empty")
+            // An agent's time-to-first-event is the same shape of wait as a
+            // model's time-to-first-token, so it takes the same edges — and
+            // the same `request_ttft` operator override, rather than adding a
+            // config key that would have to be kept in step with it.
+            .set_buckets_for_metric(
+                Matcher::Full(M_A2A_TTFB_SECONDS.to_string()),
+                &buckets.request_ttft,
+            )
+            .expect("bucket lists are validated non-empty")
             .build_recorder();
         let handle = recorder.handle();
         static NEXT_INSTANCE: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0);
@@ -1033,6 +1078,90 @@ impl Metrics {
         );
     }
 
+    /// Record one finished A2A call across the whole `aisix_a2a_*` family.
+    ///
+    /// A single entry point rather than four, so a new call site cannot land
+    /// with half the series wired — the same anti-drift move
+    /// [`crate::metrics`]' request families make. Every label here is bounded
+    /// by construction: `agent` is a registered resource, `operation` and
+    /// `task_state` come from fixed sets.
+    ///
+    /// `ttfb` and `stream_events` are `None` / `0` for a unary call, which
+    /// observes neither series.
+    pub fn record_a2a_call(&self, labels: A2aLabels<'_>, call: A2aCallOutcome<'_>) {
+        let status = status_bucket(labels.status);
+        self.cached_counter(
+            M_A2A_REQUESTS_TOTAL,
+            1,
+            |k| {
+                k.label(labels.agent);
+                k.label(labels.operation);
+                k.label(status);
+            },
+            || {
+                metrics::counter!(
+                    M_A2A_REQUESTS_TOTAL,
+                    "agent" => labels.agent.to_string(),
+                    "operation" => labels.operation.to_string(),
+                    "status" => status.to_string(),
+                )
+            },
+        );
+        if let Some(ttfb) = call.ttfb {
+            self.cached_histogram(
+                M_A2A_TTFB_SECONDS,
+                ttfb.as_secs_f64(),
+                |k| {
+                    k.label(labels.agent);
+                    k.label(labels.operation);
+                },
+                || {
+                    metrics::histogram!(
+                        M_A2A_TTFB_SECONDS,
+                        "agent" => labels.agent.to_string(),
+                        "operation" => labels.operation.to_string(),
+                    )
+                },
+            );
+        }
+        if call.stream_events > 0 {
+            self.cached_counter(
+                M_A2A_STREAM_EVENTS_TOTAL,
+                u64::from(call.stream_events),
+                |k| {
+                    k.label(labels.agent);
+                    k.label(labels.operation);
+                },
+                || {
+                    metrics::counter!(
+                        M_A2A_STREAM_EVENTS_TOTAL,
+                        "agent" => labels.agent.to_string(),
+                        "operation" => labels.operation.to_string(),
+                    )
+                },
+            );
+        }
+        // Only a call that reached a task has a state to report; counting an
+        // empty one would invent a bucket for "the upstream never answered".
+        if !call.task_state.is_empty() {
+            self.cached_counter(
+                M_A2A_TASK_STATE_TOTAL,
+                1,
+                |k| {
+                    k.label(labels.agent);
+                    k.label(call.task_state);
+                },
+                || {
+                    metrics::counter!(
+                        M_A2A_TASK_STATE_TOTAL,
+                        "agent" => labels.agent.to_string(),
+                        "state" => call.task_state.to_string(),
+                    )
+                },
+            );
+        }
+    }
+
     /// Count a request the client abandoned before it produced a response
     /// head. `endpoint` must already be a bounded route template — see
     /// [`M_PROXY_CLIENT_CANCELLED_TOTAL`].
@@ -1852,6 +1981,29 @@ impl aisix_core::GuardrailMetricsSink for Metrics {
 /// low-cardinality: bounded endpoint set, configured model/provider names,
 /// bucketed status — never per-key / per-user dimensions, which would
 /// multiply every bucket edge.
+/// Who was called and how, for the `aisix_a2a_*` family. Every field is
+/// bounded: a registered agent name and the canonical operation set.
+#[derive(Clone, Copy)]
+pub struct A2aLabels<'a> {
+    pub agent: &'a str,
+    pub operation: &'a str,
+    /// Raw HTTP status; bucketed to `2xx` / `4xx` / … at record time.
+    pub status: u16,
+}
+
+/// What one A2A call did, for the series that only some calls observe.
+#[derive(Clone, Copy, Default)]
+pub struct A2aCallOutcome<'a> {
+    /// Time to the upstream agent's first streamed event. `None` for a unary
+    /// call and for a stream that never produced one.
+    pub ttfb: Option<Duration>,
+    /// Events relayed downstream. 0 for a unary call.
+    pub stream_events: u32,
+    /// Normalized task state the call ended on; empty when no response
+    /// carried one.
+    pub task_state: &'a str,
+}
+
 #[derive(Clone, Copy)]
 pub struct LatencyLabels<'a> {
     /// Route template, e.g. `/v1/chat/completions`. Bounded set.
@@ -2429,6 +2581,63 @@ mod tests {
         assert!(rendered.contains("policy_id=\"\""));
     }
 
+    #[test]
+    fn a2a_family_slices_by_agent_and_operation() {
+        let m = Metrics::new(false);
+        m.record_a2a_call(
+            A2aLabels {
+                agent: "invoices",
+                operation: "message/stream",
+                status: 200,
+            },
+            A2aCallOutcome {
+                ttfb: Some(Duration::from_millis(80)),
+                stream_events: 17,
+                task_state: "completed",
+            },
+        );
+        let rendered = m.render();
+        // The dimension the proxy families cannot answer: which agent, which
+        // operation.
+        assert!(rendered.contains(&format!(
+            "{M_A2A_REQUESTS_TOTAL}{{agent=\"invoices\",operation=\"message/stream\",status=\"2xx\"}} 1"
+        )));
+        assert!(rendered.contains(&format!(
+            "{M_A2A_STREAM_EVENTS_TOTAL}{{agent=\"invoices\",operation=\"message/stream\"}} 17"
+        )));
+        assert!(rendered.contains(&format!(
+            "{M_A2A_TASK_STATE_TOTAL}{{agent=\"invoices\",state=\"completed\"}} 1"
+        )));
+        // A real histogram, not a summary — the buckets are registered.
+        assert!(rendered.contains(&format!("{M_A2A_TTFB_SECONDS}_bucket")));
+        // Task and context ids are the whole reason this family is safe to
+        // label; neither may ever appear.
+        assert!(!rendered.contains("task_id="));
+        assert!(!rendered.contains("context_id="));
+    }
+
+    #[test]
+    fn a2a_unary_call_observes_only_the_series_it_has_data_for() {
+        let m = Metrics::new(false);
+        m.record_a2a_call(
+            A2aLabels {
+                agent: "invoices",
+                operation: "tasks/get",
+                status: 502,
+            },
+            // No stream, and the upstream never answered, so no task state.
+            A2aCallOutcome::default(),
+        );
+        let rendered = m.render();
+        assert!(rendered.contains("status=\"5xx\""));
+        // An absent figure must not be recorded as zero: a call with no stream
+        // is not a call whose stream carried nothing, and a call with no
+        // answer has no task state to bucket.
+        assert!(!rendered.contains(M_A2A_STREAM_EVENTS_TOTAL));
+        assert!(!rendered.contains(M_A2A_TASK_STATE_TOTAL));
+        assert!(!rendered.contains(M_A2A_TTFB_SECONDS));
+    }
+
     #[test]
     fn guardrail_outcome_counters_increment() {
         let m = Metrics::new(false);
diff --git a/crates/aisix-obs/src/otlp_http_sink.rs b/crates/aisix-obs/src/otlp_http_sink.rs
index ec046a9bb..6403c2f86 100644
--- a/crates/aisix-obs/src/otlp_http_sink.rs
+++ b/crates/aisix-obs/src/otlp_http_sink.rs
@@ -745,6 +745,12 @@ fn build_otlp_span(record: &SinkRecord, exporter_name: &str) -> Value {
                     attributes.push(attr_string_capped(key, value));
                 }
             }
+            if event.a2a_stream_event_count > 0 {
+                attributes.push(attr_int(
+                    "aisix.a2a.stream_event_count",
+                    i64::from(event.a2a_stream_event_count),
+                ));
+            }
         }
         "mcp" => {
             if !event.mcp_tool_name.is_empty() {
@@ -1461,6 +1467,39 @@ mod tests {
         assert_eq!(string_at("aisix.a2a.task_state"), "working");
     }
 
+    #[test]
+    fn a_streamed_a2a_call_carries_its_event_count() {
+        // A unary call has no count, and exporting a zero would read as "the
+        // stream carried nothing" rather than "there was no stream".
+        let mut streamed = sample_event();
+        streamed.inbound_protocol = "a2a".into();
+        streamed.a2a_agent_name = "invoice-processor".into();
+        streamed.a2a_stream_event_count = 17;
+        let body = build_otlp_traces_payload(&streamed, "test-exp");
+        let attrs = body["resourceSpans"][0]["scopeSpans"][0]["spans"][0]["attributes"]
+            .as_array()
+            .unwrap()
+            .clone();
+        assert_eq!(
+            attrs
+                .iter()
+                .find(|a| a["key"] == "aisix.a2a.stream_event_count")
+                .unwrap()["value"]["intValue"],
+            "17"
+        );
+
+        let mut unary = sample_event();
+        unary.inbound_protocol = "a2a".into();
+        let body = build_otlp_traces_payload(&unary, "test-exp");
+        let attrs = body["resourceSpans"][0]["scopeSpans"][0]["spans"][0]["attributes"]
+            .as_array()
+            .unwrap()
+            .clone();
+        assert!(attrs
+            .iter()
+            .all(|a| a["key"] != "aisix.a2a.stream_event_count"));
+    }
+
     #[test]
     fn an_mcp_call_exports_as_a_tool_span() {
         let mut ev = sample_event();
diff --git a/crates/aisix-obs/src/usage.rs b/crates/aisix-obs/src/usage.rs
index 1306ba05f..635f88bff 100644
--- a/crates/aisix-obs/src/usage.rs
+++ b/crates/aisix-obs/src/usage.rs
@@ -508,6 +508,16 @@ pub struct UsageEvent {
     /// all (the call failed before the upstream answered).
     #[serde(default, skip_serializing_if = "String::is_empty")]
     pub a2a_task_state: String,
+
+    /// Events the gateway relayed downstream on a streamed A2A call.
+    ///
+    /// Read together with `upstream_ttft_ms` and `upstream_latency_ms` it
+    /// separates the two ways a stream disappoints: nothing arrived for a long
+    /// time (high TTFT), or plenty arrived and none of it advanced the task
+    /// (high count, no terminal state). 0 for a unary call, and for a stream
+    /// whose upstream produced nothing at all.
+    #[serde(default, skip_serializing_if = "is_zero_u32")]
+    pub a2a_stream_event_count: u32,
 }
 
 #[inline]
diff --git a/crates/aisix-proxy/src/a2a.rs b/crates/aisix-proxy/src/a2a.rs
index 22b32717a..f8f624ca3 100644
--- a/crates/aisix-proxy/src/a2a.rs
+++ b/crates/aisix-proxy/src/a2a.rs
@@ -66,6 +66,28 @@ struct A2aCall {
     /// The wire version the agent is pinned to (`0.3` / `1.0`).
     protocol_version: &'static str,
     facts: A2aCallFacts,
+    /// How the stream behaved, for a streaming call. Left at its default for
+    /// a unary one, which observes none of the stream series.
+    stream: A2aStreamProgress,
+}
+
+/// What a streamed A2A call did on the wire, accumulated as events pass.
+///
+/// A stream is the only place these can be observed: once the response head is
+/// out, nothing downstream of it can say how long the first event took, how
+/// many followed, or whether the caller was still there at the end.
+#[derive(Default)]
+struct A2aStreamProgress {
+    /// Time from the call starting to the upstream's first event. `None` until
+    /// one arrives — a stream that opens and produces nothing never sets it,
+    /// and no figure is invented for it.
+    ttfb: Option<Duration>,
+    /// Events relayed downstream.
+    event_count: u32,
+    /// Set when the upstream stream ends — exhausted or faulted. Still false
+    /// when the guard drops means the generator was cancelled underneath us,
+    /// which is a caller that hung up mid-task.
+    reached_end: bool,
 }
 
 /// Serve a JSON-RPC request to `/a2a/:agent`. Authentication (`401`), per-agent
@@ -190,6 +212,7 @@ async fn dispatch(
         method,
         protocol_version: upstream.protocol_version.as_wire_str(),
         facts: A2aCallFacts::default(),
+        stream: A2aStreamProgress::default(),
     };
     // Read before the upstream is contacted, so a call that never lands still
     // records which task the caller was asking about.
@@ -305,13 +328,22 @@ impl Drop for StreamUsageOnDrop {
         if std::thread::panicking() {
             return;
         }
+        // A stream the caller abandoned mid-task is recorded as 499, the same
+        // outcome the LLM streaming paths record for the same event. Only a
+        // still-successful call is downgraded: a stream that already faulted
+        // has a truer status than the hang-up that followed it.
+        let status = if self.call.stream.reached_end || self.status != StatusCode::OK.as_u16() {
+            self.status
+        } else {
+            crate::CLIENT_CLOSED_REQUEST
+        };
         emit_a2a_usage(
             &self.state,
             &self.auth,
             &self.request_id,
             &self.agent,
             &self.call,
-            self.status,
+            status,
             self.started.elapsed(),
         );
     }
@@ -377,6 +409,11 @@ async fn dispatch_stream(
                     // Read the task's progress off the event on its way past —
                     // the relay itself stays verbatim.
                     guard.call.facts.observe_result(&value);
+                    guard.call.stream.event_count += 1;
+                    // The agent's own time to first byte. Stamped on the first
+                    // event of any kind, matching how the LLM paths stamp
+                    // TTFT, so the two figures answer the same question.
+                    guard.call.stream.ttfb.get_or_insert_with(|| started.elapsed());
                     yield Ok::<_, std::convert::Infallible>(
                         axum::response::sse::Event::default().data(value.to_string()),
                     );
@@ -394,6 +431,9 @@ async fn dispatch_stream(
                 }
             }
         }
+        // The upstream is done — exhausted or faulted. Reaching here at all is
+        // what separates a finished stream from one the caller abandoned.
+        guard.call.stream.reached_end = true;
         drop(guard);
     };
 
@@ -568,8 +608,28 @@ fn emit_a2a_usage(
         a2a_task_id: call.facts.task_id.clone(),
         a2a_context_id: call.facts.context_id.clone(),
         a2a_task_state: call.facts.task_state.clone(),
+        a2a_stream_event_count: call.stream.event_count,
+        upstream_ttft_ms: call
+            .stream
+            .ttfb
+            .map(|d| d.as_millis().min(u32::MAX as u128) as u32)
+            .unwrap_or_default(),
         ..Default::default()
     };
+    // The `aisix_a2a_*` family rides on the same chokepoint as the usage
+    // event, so a path that accounts for a call cannot skip metering it.
+    state.metrics.record_a2a_call(
+        aisix_obs::A2aLabels {
+            agent,
+            operation: call.operation,
+            status: status_code,
+        },
+        aisix_obs::A2aCallOutcome {
+            ttfb: call.stream.ttfb,
+            stream_events: call.stream.event_count,
+            task_state: &call.facts.task_state,
+        },
+    );
     state.usage_sink.try_emit("a2a", event.clone());
     let snap = state.snapshot.load();
     let exporters = snap.observability_exporters.entries();
@@ -865,6 +925,12 @@ mod tests {
             .expect("an abandoned stream still emits usage");
         assert_eq!(event.a2a_task_id, "task-88");
         assert_eq!(event.a2a_task_state, "working");
+        // A caller that hung up mid-task is not a success. Recording 200 here
+        // put abandoned streams in the same bucket as completed ones, so an
+        // agent nobody waits for looked perfectly healthy.
+        assert_eq!(event.status_code, crate::CLIENT_CLOSED_REQUEST);
+        // Only what the caller actually received is counted.
+        assert_eq!(event.a2a_stream_event_count, 2);
     }
 
     #[tokio::test]
@@ -945,6 +1011,27 @@ mod tests {
         assert_eq!(event.a2a_task_id, "task-77");
         assert_eq!(event.a2a_context_id, "ctx-77");
         assert_eq!(event.a2a_task_state, "input-required");
+        // A stream read to the end is a success, and every event it relayed
+        // is counted.
+        assert_eq!(event.status_code, StatusCode::OK.as_u16());
+        assert_eq!(event.a2a_stream_event_count, 3);
+    }
+
+    #[tokio::test]
+    async fn a_unary_call_observes_none_of_the_stream_figures() {
+        // Nothing was streamed, so an event count or a time-to-first-event
+        // would be a number with no referent.
+        let event = usage_event_for(
+            &spawn_task_agent().await,
+            serde_json::json!({
+                "jsonrpc": "2.0", "id": 1, "method": "message/send",
+                "params": {"message": {"role": "user", "contextId": "ctx-u"}}
+            }),
+        )
+        .await;
+
+        assert_eq!(event.a2a_stream_event_count, 0);
+        assert_eq!(event.upstream_ttft_ms, 0);
     }
 
     #[tokio::test]
diff --git a/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts b/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
index 96cde4dcc..563638163 100644
--- a/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
+++ b/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
@@ -221,6 +221,65 @@ describe("a2a protocol telemetry e2e (AISIX-Cloud#1215)", () => {
     expect(span.attributes["aisix.a2a.task_state"]).toBe("completed");
   });
 
+  test("a streamed call records how it ran, not just how it ended", async (ctx) => {
+    if (!etcdReachable || !app || !upstream10 || !upstream03 || !otlp) return ctx.skip();
+
+    // The stub paces its three events apart, so a wait for the first one is
+    // separable from the total: a stream that is slow to start and one that is
+    // slow overall are different problems with different fixes.
+    const contextId = `ctx-${randomUUID()}`;
+    expect(
+      await call("invoices", {
+        jsonrpc: "2.0",
+        id: 6,
+        method: "message/stream",
+        params: { message: { role: "user", contextId, parts: [] } },
+      }),
+    ).toBe(200);
+
+    const span = await awaitSpan((s) => s.attributes["gen_ai.conversation.id"] === contextId);
+    expect(span.attributes["aisix.a2a.stream_event_count"]).toBe(3);
+    // The agent's own time to first event. The stub pauses before each of its
+    // three events, so a figure that actually stopped at the first one is a
+    // fraction of the call; one that quietly measured the whole stream would
+    // equal the total. Compared against the total rather than a fixed
+    // millisecond bound, so the assertion does not track the stub's pacing.
+    const ttfb = span.attributes["aisix.upstream_ttft_ms"] as number;
+    const total = span.attributes["aisix.downstream_latency_ms"] as number;
+    expect(ttfb).toBeGreaterThan(0);
+    expect(ttfb).toBeLessThan(total / 2);
+  });
+
+  test("the a2a metric family slices by agent and operation", async (ctx) => {
+    if (!etcdReachable || !app || !upstream10 || !upstream03 || !otlp) return ctx.skip();
+
+    // `aisix_proxy_requests_total` already counts these calls, but only by
+    // route — it cannot answer "is the invoices agent's stream failing?".
+    await call("invoices", {
+      jsonrpc: "2.0",
+      id: 7,
+      method: "message/stream",
+      params: { message: { role: "user", parts: [] } },
+    });
+
+    const scrape = await fetch(`${app.metricsUrl}/metrics`);
+    expect(scrape.status).toBe(200);
+    const text = await scrape.text();
+
+    expect(text).toContain(
+      'aisix_a2a_requests_total{agent="invoices",operation="message/stream",status="2xx"}',
+    );
+    expect(text).toContain(
+      'aisix_a2a_stream_events_total{agent="invoices",operation="message/stream"}',
+    );
+    expect(text).toContain('aisix_a2a_task_state_total{agent="invoices",state="completed"}');
+    expect(text).toContain("aisix_a2a_ttfb_seconds_bucket");
+    // The ids that make a task traceable are exactly the ones that must not
+    // become label values.
+    expect(text).not.toMatch(/aisix_a2a_[a-z_]*\{[^}]*task_id=/);
+    expect(text).not.toMatch(/aisix_a2a_[a-z_]*\{[^}]*context_id=/);
+  });
+
   test("an unrecognised method cannot become an unbounded label", async (ctx) => {
     if (!etcdReachable || !app || !upstream10 || !upstream03 || !otlp) return ctx.skip();
 
diff --git a/tests/e2e/src/harness/upstream-a2a.ts b/tests/e2e/src/harness/upstream-a2a.ts
index 93c79c530..81ccfb389 100644
--- a/tests/e2e/src/harness/upstream-a2a.ts
+++ b/tests/e2e/src/harness/upstream-a2a.ts
@@ -229,6 +229,10 @@ async function handle(
     const envelope = (result: Record<string, unknown>) =>
       `data: ${JSON.stringify({ jsonrpc: "2.0", id: body?.id ?? null, result })}\n\n`;
     res.write(": open\n\n");
+    // The agent thinks before it says anything, like a real one. Without this
+    // the first event lands sub-millisecond and a time-to-first-event of 0 is
+    // indistinguishable from one that was never recorded.
+    await new Promise((resolve) => setTimeout(resolve, STREAM_GAP_MS));
     res.write(
       envelope(
         payload("status-update", {

From 5d78dda0df84d6619b4bc1250d32073d244f52e0 Mon Sep 17 00:00:00 2001
From: Jarvis <jarvis@api7.ai>
Date: Tue, 11 Aug 2026 19:17:50 +0800
Subject: [PATCH 2/3] fix(a2a): a completed stream is delivered at its terminal
 event, not at EOF
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

An audit found the inversion this PR was supposed to prevent. `reached_end`
was set after the relay loop exits, but `async_stream` resumes its body only
when the consumer pulls again — and an A2A client stops reading at the
terminal event. An agent that holds the SSE connection open after finishing
(the upstream stream ends on body EOF, never on `final`) therefore had every
COMPLETED task recorded as a client hang-up.

`chat.rs` documents this exact trap and sets the flag before its final
yield; the LLM path can anchor on upstream EOF because the gateway writes
`[DONE]` itself. An A2A terminal event is forwarded, not generated, so the
anchor moves to the moment it is relayed. `is_stream_end` reads both wire
versions: 0.3 marks it with `final`, while 1.0's `TaskStatusUpdateEvent` has
no such field and only the task's own state says so. Pinned by a test that
records 499 without the fix.

Also from the audit:

- A2A streams had no end-to-end duration series at all. The claim that
  `aisix_proxy_request_duration_seconds` already covered it was wrong for
  streaming: the handler returns when the response head goes out, so it
  timed how long the stream took to OPEN. Now observed at stream completion,
  same placement as the LLM streaming paths.
- `aisix_a2a_ttfb_seconds` takes its own `a2a_ttfb` bucket override instead
  of borrowing `request_ttft`, which silently made one config key govern two
  histograms — and an agent's wait for its first event has a different range
  from a model's wait for its first token.
- `aisix_a2a_requests_total` does NOT count the same set as its proxy-family
  counterpart (calls refused before the agent resolves; abandoned streams
  differ in class). Both divergences are now stated where an operator reads
  them rather than asserted away.
- An agent that refuses a streaming call with a JSON-RPC error at HTTP 200
  no longer counts as a stream event: it would inflate the event counter and
  put a near-zero observation into the TTFB histogram.
- `operation` and `task_state` are `&'static str`, so their boundedness is
  compiler-enforced rather than a comment; `A2aCallFacts::task_state` follows.
- Documents the `/a2a` exception to `downstream_latency_ms` (it spans the
  whole stream, so subtracting TTFT there is not gateway overhead), and that
  an absent `upstream_ttft_ms` means "not measured", not "no stream".
- Restores the doc comment stranded off `LatencyLabels`, and covers the
  paths that had none: terminal-event detection in both versions, the 4xx
  branch of the request counter, drop-before-first-poll, and a stamped TTFB.
---
 crates/aisix-a2a/src/lib.rs                   |   2 +-
 crates/aisix-a2a/src/telemetry.rs             |  80 ++++++++-
 crates/aisix-core/src/config.rs               |   8 +
 crates/aisix-obs/src/metrics.rs               |  52 +++---
 crates/aisix-obs/src/usage.rs                 |  12 +-
 crates/aisix-proxy/src/a2a.rs                 | 164 ++++++++++++++++--
 .../cases/a2a-protocol-telemetry-e2e.test.ts  |   7 +-
 7 files changed, 285 insertions(+), 40 deletions(-)

diff --git a/crates/aisix-a2a/src/lib.rs b/crates/aisix-a2a/src/lib.rs
index 112fbd846..636235614 100644
--- a/crates/aisix-a2a/src/lib.rs
+++ b/crates/aisix-a2a/src/lib.rs
@@ -27,4 +27,4 @@ pub use bridge::{
     HttpBridge, DEFAULT_UPSTREAM_TIMEOUT,
 };
 pub use error::A2aError;
-pub use telemetry::{canonical_operation, is_streaming_operation, A2aCallFacts};
+pub use telemetry::{canonical_operation, is_stream_end, is_streaming_operation, A2aCallFacts};
diff --git a/crates/aisix-a2a/src/telemetry.rs b/crates/aisix-a2a/src/telemetry.rs
index e442605da..6688b94ea 100644
--- a/crates/aisix-a2a/src/telemetry.rs
+++ b/crates/aisix-a2a/src/telemetry.rs
@@ -151,6 +151,45 @@ pub fn normalize_task_state(state: &str) -> &'static str {
         .unwrap_or(UNKNOWN_TASK_STATE)
 }
 
+/// Task states that mean the agent has said everything it is going to say on
+/// this stream: the task is finished, or it is waiting on the caller.
+const STREAM_ENDING_STATES: &[&str] = &[
+    "completed",
+    "canceled",
+    "failed",
+    "rejected",
+    "input-required",
+    "auth-required",
+];
+
+/// Whether a streamed event is the last one its stream will carry.
+///
+/// A caller stops reading here, which makes this — not the upstream's
+/// eventual EOF — the moment the response was fully delivered. Anything that
+/// waits for the connection to close instead reports a completed task as
+/// abandoned whenever the agent leaves the stream open afterwards.
+///
+/// 0.3 says so with `final`. 1.0's `TaskStatusUpdateEvent` has no such field
+/// at all, so there the task's own state is the only signal: a task that is
+/// neither `submitted` nor `working` has nothing further to stream until the
+/// caller acts.
+pub fn is_stream_end(event: &Value) -> bool {
+    let Some(result) = event.get("result") else {
+        return false;
+    };
+    let payload = unwrap_payload(result);
+    if payload.get("final").and_then(Value::as_bool) == Some(true) {
+        return true;
+    }
+    let Some(state) = payload
+        .get("status")
+        .and_then(|status| str_field(status, "state"))
+    else {
+        return false;
+    };
+    STREAM_ENDING_STATES.contains(&normalize_task_state(state))
+}
+
 /// What one A2A call touched, accumulated as its request and response(s) are
 /// seen.
 ///
@@ -169,7 +208,12 @@ pub struct A2aCallFacts {
     /// Last task state reported, in its 0.3 spelling. Empty when no response
     /// carried one — a state is never invented for a call that failed before
     /// the upstream answered.
-    pub task_state: String,
+    ///
+    /// `&'static str` rather than a `String`: the value is always one of the
+    /// specification's states or `unknown`, and typing it that way is what
+    /// lets it be used as a metric label without a caller having to promise
+    /// it is bounded.
+    pub task_state: &'static str,
 }
 
 impl A2aCallFacts {
@@ -223,7 +267,7 @@ impl A2aCallFacts {
         if let Some(status) = payload.get("status") {
             self.set_task_id(str_field(payload, "id"));
             if let Some(state) = str_field(status, "state") {
-                self.task_state = normalize_task_state(state).to_string();
+                self.task_state = normalize_task_state(state);
             }
         }
     }
@@ -540,6 +584,38 @@ mod tests {
         assert_eq!(facts.task_state, "");
     }
 
+    #[test]
+    fn the_terminal_event_is_recognised_in_both_versions() {
+        // The moment a caller stops reading. 0.3 marks it with `final`; 1.0's
+        // status-update has no such field, so only the task's own state says
+        // so — and both spellings of that state have to work.
+        for terminal in [
+            json!({"result": {"kind": "status-update", "taskId": "t", "final": true,
+                              "status": {"state": "input-required"}}}),
+            json!({"result": {"kind": "task", "id": "t", "status": {"state": "completed"}}}),
+            json!({"result": {"statusUpdate": {"taskId": "t",
+                              "status": {"state": "TASK_STATE_FAILED"}}}}),
+            json!({"result": {"task": {"id": "t", "status": {"state": "TASK_STATE_CANCELED"}}}}),
+        ] {
+            assert!(is_stream_end(&terminal), "{terminal} ends the stream");
+        }
+        // A task still running does not, nor does an event with no state at
+        // all (an artifact chunk), nor an error envelope.
+        for ongoing in [
+            json!({"result": {"kind": "status-update", "taskId": "t", "final": false,
+                              "status": {"state": "working"}}}),
+            json!({"result": {"statusUpdate": {"taskId": "t",
+                              "status": {"state": "TASK_STATE_SUBMITTED"}}}}),
+            json!({"result": {"artifactUpdate": {"taskId": "t"}}}),
+            json!({"error": {"code": -32000, "message": "boom"}}),
+        ] {
+            assert!(
+                !is_stream_end(&ongoing),
+                "{ongoing} does not end the stream"
+            );
+        }
+    }
+
     #[test]
     fn an_empty_id_never_erases_one_already_seen() {
         let mut facts = A2aCallFacts::default();
diff --git a/crates/aisix-core/src/config.rs b/crates/aisix-core/src/config.rs
index 06cc60ed3..e4477db54 100644
--- a/crates/aisix-core/src/config.rs
+++ b/crates/aisix-core/src/config.rs
@@ -789,6 +789,13 @@ pub struct HistogramBucketsConfig {
     pub request_ttft: Option<Vec<f64>>,
     /// `aisix_guardrail_latency_seconds`
     pub guardrail_latency: Option<Vec<f64>>,
+    /// `aisix_a2a_ttfb_seconds`
+    ///
+    /// Separate from `request_ttft` on purpose: an agent's wait for its first
+    /// event and a model's wait for its first token have the same shape but
+    /// not the same range — an A2A task may think for minutes before it says
+    /// anything. Defaults to the same edges as `request_ttft`.
+    pub a2a_ttfb: Option<Vec<f64>>,
 }
 
 /// One `client_type_rules` entry: a regex tried against the raw inbound
@@ -1307,6 +1314,7 @@ impl Config {
                 .with_list_parse_key("observability.metrics.buckets.request_e2e_latency")
                 .with_list_parse_key("observability.metrics.buckets.request_ttft")
                 .with_list_parse_key("observability.metrics.buckets.guardrail_latency")
+                .with_list_parse_key("observability.metrics.buckets.a2a_ttfb")
                 .try_parsing(true),
         );
 
diff --git a/crates/aisix-obs/src/metrics.rs b/crates/aisix-obs/src/metrics.rs
index 890c449f7..fa5c19da6 100644
--- a/crates/aisix-obs/src/metrics.rs
+++ b/crates/aisix-obs/src/metrics.rs
@@ -225,17 +225,24 @@ pub const M_REQUEST_TTFT_SECONDS: &str = "aisix_request_ttft_seconds";
 // Request DURATION is not repeated here: `aisix_proxy_request_duration_seconds`
 // already times `/a2a` end to end, and a second histogram of the same quantity
 // would only be a second thing to keep in sync.
-/// A2A calls by agent, canonical operation and status class. Counts the same
-/// requests as `aisix_proxy_requests_total{endpoint="/a2a"}`, split the way an
-/// agent operator needs.
+/// A2A calls by agent, canonical operation and status class.
+///
+/// It does NOT agree with `aisix_proxy_requests_total{endpoint="/a2a"}`, and
+/// two differences are deliberate. A call refused before its agent is resolved
+/// — a bad key, a denied agent, an unknown one — has no agent or operation to
+/// file under, so it is counted there and not here. And a stream the caller
+/// abandoned is `4xx` here but `2xx` there, because the response head really
+/// did go out as a 200. Read this family for agent health and that one for
+/// route traffic; do not expect the totals to match.
 pub const M_A2A_REQUESTS_TOTAL: &str = "aisix_a2a_requests_total";
 /// Time from an A2A streaming call starting to the FIRST event the upstream
 /// agent pushed — the agent's own "time to first byte".
 ///
 /// Named for the event rather than a token because an agent stream carries
-/// task updates, not tokens; it shares [`M_REQUEST_TTFT_SECONDS`]'s bucket
-/// edges (and therefore its `request_ttft` operator override) since the two
-/// measure the same shape of wait.
+/// task updates, not tokens. It defaults to [`M_REQUEST_TTFT_SECONDS`]'s
+/// bucket edges — the same shape of wait — but takes its own `a2a_ttfb`
+/// operator override, so tuning one histogram never silently retunes the
+/// other.
 pub const M_A2A_TTFB_SECONDS: &str = "aisix_a2a_ttfb_seconds";
 /// Events relayed downstream on A2A streaming calls. Divided by
 /// [`M_A2A_REQUESTS_TOTAL`] over the streaming operations it gives events per
@@ -325,6 +332,7 @@ pub struct HistogramBuckets {
     pub request_e2e_latency: Vec<f64>,
     pub request_ttft: Vec<f64>,
     pub guardrail_latency: Vec<f64>,
+    pub a2a_ttfb: Vec<f64>,
 }
 
 impl Default for HistogramBuckets {
@@ -333,6 +341,7 @@ impl Default for HistogramBuckets {
             request_e2e_latency: DEFAULT_E2E_LATENCY_BUCKETS.to_vec(),
             request_ttft: DEFAULT_TTFT_BUCKETS.to_vec(),
             guardrail_latency: DEFAULT_GUARDRAIL_LATENCY_BUCKETS.to_vec(),
+            a2a_ttfb: DEFAULT_TTFT_BUCKETS.to_vec(),
         }
     }
 }
@@ -363,6 +372,7 @@ impl HistogramBuckets {
                 cfg.guardrail_latency.as_deref(),
                 DEFAULT_GUARDRAIL_LATENCY_BUCKETS,
             )?,
+            a2a_ttfb: resolve_edges("a2a_ttfb", cfg.a2a_ttfb.as_deref(), DEFAULT_TTFT_BUCKETS)?,
         })
     }
 }
@@ -685,13 +695,9 @@ impl Metrics {
                 &buckets.guardrail_latency,
             )
             .expect("bucket lists are validated non-empty")
-            // An agent's time-to-first-event is the same shape of wait as a
-            // model's time-to-first-token, so it takes the same edges — and
-            // the same `request_ttft` operator override, rather than adding a
-            // config key that would have to be kept in step with it.
             .set_buckets_for_metric(
                 Matcher::Full(M_A2A_TTFB_SECONDS.to_string()),
-                &buckets.request_ttft,
+                &buckets.a2a_ttfb,
             )
             .expect("bucket lists are validated non-empty")
             .build_recorder();
@@ -1088,7 +1094,7 @@ impl Metrics {
     ///
     /// `ttfb` and `stream_events` are `None` / `0` for a unary call, which
     /// observes neither series.
-    pub fn record_a2a_call(&self, labels: A2aLabels<'_>, call: A2aCallOutcome<'_>) {
+    pub fn record_a2a_call(&self, labels: A2aLabels<'_>, call: A2aCallOutcome) {
         let status = status_bucket(labels.status);
         self.cached_counter(
             M_A2A_REQUESTS_TOTAL,
@@ -1977,31 +1983,33 @@ impl aisix_core::GuardrailMetricsSink for Metrics {
     }
 }
 
-/// Labels for the SLO latency histograms (AISIX-Cloud#1011). Deliberately
-/// low-cardinality: bounded endpoint set, configured model/provider names,
-/// bucketed status — never per-key / per-user dimensions, which would
-/// multiply every bucket edge.
 /// Who was called and how, for the `aisix_a2a_*` family. Every field is
 /// bounded: a registered agent name and the canonical operation set.
 #[derive(Clone, Copy)]
 pub struct A2aLabels<'a> {
+    /// Registered agent name. Borrowed: it comes from the snapshot, and an
+    /// unregistered agent is refused before any call is metered.
     pub agent: &'a str,
-    pub operation: &'a str,
+    /// Canonical operation. `&'static str` so the fixed set is enforced by the
+    /// compiler rather than by a comment — the same reason
+    /// `record_usage_event_emit` takes its handler that way.
+    pub operation: &'static str,
     /// Raw HTTP status; bucketed to `2xx` / `4xx` / … at record time.
     pub status: u16,
 }
 
 /// What one A2A call did, for the series that only some calls observe.
 #[derive(Clone, Copy, Default)]
-pub struct A2aCallOutcome<'a> {
+pub struct A2aCallOutcome {
     /// Time to the upstream agent's first streamed event. `None` for a unary
     /// call and for a stream that never produced one.
     pub ttfb: Option<Duration>,
     /// Events relayed downstream. 0 for a unary call.
     pub stream_events: u32,
     /// Normalized task state the call ended on; empty when no response
-    /// carried one.
-    pub task_state: &'a str,
+    /// carried one. `&'static str` for the same reason as
+    /// [`A2aLabels::operation`].
+    pub task_state: &'static str,
 }
 
 #[derive(Clone, Copy)]
@@ -2629,7 +2637,9 @@ mod tests {
             A2aCallOutcome::default(),
         );
         let rendered = m.render();
-        assert!(rendered.contains("status=\"5xx\""));
+        assert!(rendered.contains(&format!(
+            "{M_A2A_REQUESTS_TOTAL}{{agent=\"invoices\",operation=\"tasks/get\",status=\"5xx\"}} 1"
+        )));
         // An absent figure must not be recorded as zero: a call with no stream
         // is not a call whose stream carried nothing, and a call with no
         // answer has no task state to bucket.
diff --git a/crates/aisix-obs/src/usage.rs b/crates/aisix-obs/src/usage.rs
index 635f88bff..344db7d07 100644
--- a/crates/aisix-obs/src/usage.rs
+++ b/crates/aisix-obs/src/usage.rs
@@ -144,7 +144,11 @@ pub struct UsageEvent {
     /// (AISIX-Cloud#1225) shows the wait in `upstream_latency_ms`, not
     /// here. 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).
+    /// cache-hit paths (omitted from the wire via skip_serializing_if)
+    /// — and, since the field is whole milliseconds, also on a stream
+    /// whose first frame arrived in under one. Absent therefore means
+    /// "no streamed first frame was measured", not "there was no
+    /// stream".
     ///
     /// This is what the UPSTREAM delivered on this attempt. What the
     /// caller actually waited for is `downstream_latency_ms`, which also
@@ -178,6 +182,12 @@ pub struct UsageEvent {
     ///
     /// Absent (0) on the non-terminal attempts of a request, and on any
     /// path that never reached response delivery.
+    ///
+    /// `/a2a` is the one exception to the streaming rule above: it records
+    /// the WHOLE stream, because an agent's stream of task updates is the
+    /// call's product rather than a delivery mechanism for one. The
+    /// subtraction against `upstream_ttft_ms` therefore does not describe
+    /// gateway overhead there — it is the rest of the agent's own work.
     #[serde(default, skip_serializing_if = "is_zero_u32")]
     pub downstream_latency_ms: u32,
 
diff --git a/crates/aisix-proxy/src/a2a.rs b/crates/aisix-proxy/src/a2a.rs
index f8f624ca3..854f78268 100644
--- a/crates/aisix-proxy/src/a2a.rs
+++ b/crates/aisix-proxy/src/a2a.rs
@@ -31,8 +31,8 @@
 use std::time::{Duration, Instant};
 
 use aisix_a2a::{
-    canonical_operation, is_streaming_operation, upstream_from_a2a_agent, A2aBridge, A2aCallFacts,
-    A2aError, HttpBridge,
+    canonical_operation, is_stream_end, is_streaming_operation, upstream_from_a2a_agent, A2aBridge,
+    A2aCallFacts, A2aError, HttpBridge,
 };
 use aisix_obs::{AccessLog, UsageEvent};
 use axum::body::to_bytes;
@@ -337,6 +337,23 @@ impl Drop for StreamUsageOnDrop {
         } else {
             crate::CLIENT_CLOSED_REQUEST
         };
+        let elapsed = self.started.elapsed();
+        // The client-perceived duration of a streamed call, which nothing else
+        // records: the handler returns the moment the response head is out, so
+        // `aisix_proxy_request_duration_seconds` times only how long the
+        // stream took to OPEN. Same placement as the LLM streaming paths,
+        // which observe this histogram at stream completion for the same
+        // reason.
+        self.state.metrics.record_request_e2e_latency(
+            aisix_obs::LatencyLabels {
+                endpoint: "/a2a",
+                model: A2A_MODEL_LABEL,
+                provider: "a2a",
+                status,
+                streaming: true,
+            },
+            elapsed,
+        );
         emit_a2a_usage(
             &self.state,
             &self.auth,
@@ -344,7 +361,7 @@ impl Drop for StreamUsageOnDrop {
             &self.agent,
             &self.call,
             status,
-            self.started.elapsed(),
+            elapsed,
         );
     }
 }
@@ -409,11 +426,32 @@ async fn dispatch_stream(
                     // Read the task's progress off the event on its way past —
                     // the relay itself stays verbatim.
                     guard.call.facts.observe_result(&value);
-                    guard.call.stream.event_count += 1;
-                    // The agent's own time to first byte. Stamped on the first
-                    // event of any kind, matching how the LLM paths stamp
-                    // TTFT, so the two figures answer the same question.
-                    guard.call.stream.ttfb.get_or_insert_with(|| started.elapsed());
+                    // An agent may refuse a streaming call with a JSON-RPC
+                    // error at HTTP 200, which arrives here as a lone
+                    // envelope carrying `error` instead of `result`. It is
+                    // relayed, but it is not a task stream: counting it would
+                    // inflate the event counter and put a near-zero
+                    // observation into the time-to-first-event histogram,
+                    // dragging the percentile of real streams down with it.
+                    if value.get("result").is_some() {
+                        guard.call.stream.event_count += 1;
+                        // The agent's own time to first byte. Stamped on the
+                        // first event of any kind, matching how the LLM paths
+                        // stamp TTFT, so the two figures answer the same
+                        // question.
+                        guard.call.stream.ttfb.get_or_insert_with(|| started.elapsed());
+                    }
+                    // Forwarding the terminal event IS the end of the
+                    // response, and it has to be recorded BEFORE the yield:
+                    // an A2A client stops reading here, and the loop below
+                    // only resumes when the consumer pulls again — which,
+                    // for an agent that leaves the connection open after
+                    // finishing, never happens. Marking the end at the loop
+                    // exit instead reported every such completed task as
+                    // abandoned.
+                    if is_stream_end(&value) {
+                        guard.call.stream.reached_end = true;
+                    }
                     yield Ok::<_, std::convert::Infallible>(
                         axum::response::sse::Event::default().data(value.to_string()),
                     );
@@ -431,8 +469,8 @@ async fn dispatch_stream(
                 }
             }
         }
-        // The upstream is done — exhausted or faulted. Reaching here at all is
-        // what separates a finished stream from one the caller abandoned.
+        // The upstream ran out or faulted without ever marking the stream
+        // finished. Either way the caller was handed everything there was.
         guard.call.stream.reached_end = true;
         drop(guard);
     };
@@ -607,7 +645,7 @@ fn emit_a2a_usage(
         a2a_protocol_version: call.protocol_version.to_string(),
         a2a_task_id: call.facts.task_id.clone(),
         a2a_context_id: call.facts.context_id.clone(),
-        a2a_task_state: call.facts.task_state.clone(),
+        a2a_task_state: call.facts.task_state.to_string(),
         a2a_stream_event_count: call.stream.event_count,
         upstream_ttft_ms: call
             .stream
@@ -627,7 +665,7 @@ fn emit_a2a_usage(
         aisix_obs::A2aCallOutcome {
             ttfb: call.stream.ttfb,
             stream_events: call.stream.event_count,
-            task_state: &call.facts.task_state,
+            task_state: call.facts.task_state,
         },
     );
     state.usage_sink.try_emit("a2a", event.clone());
@@ -769,6 +807,9 @@ mod tests {
         assert_eq!(event.inbound_protocol, "a2a");
         assert_eq!(event.a2a_agent_name, "invoice");
         assert_eq!(event.a2a_method, "message/stream");
+        // Dropped before the body was ever polled: nothing was delivered, so
+        // this is the client hanging up, not a completed call.
+        assert_eq!(event.status_code, crate::CLIENT_CLOSED_REQUEST);
     }
 
     /// An agent that answers `message/send` with a Task in the state the
@@ -933,6 +974,92 @@ mod tests {
         assert_eq!(event.a2a_stream_event_count, 2);
     }
 
+    /// An agent that finishes the task and then leaves the SSE connection
+    /// open — the shape that exposes the difference between "the caller got
+    /// everything" and "the upstream closed".
+    async fn spawn_completed_but_open_stream_agent() -> String {
+        use axum::response::IntoResponse;
+        let app = axum::Router::new().route(
+            "/a2a",
+            axum::routing::post(|| async {
+                let event = serde_json::json!({
+                    "jsonrpc": "2.0",
+                    "result": {
+                        "kind": "status-update",
+                        "taskId": "task-55",
+                        "final": true,
+                        "status": {"state": "completed"},
+                    },
+                });
+                let body = async_stream::stream! {
+                    yield Ok::<_, std::convert::Infallible>(format!("data: {event}\n\n"));
+                    std::future::pending::<()>().await;
+                };
+                (
+                    [(axum::http::header::CONTENT_TYPE, "text/event-stream")],
+                    axum::body::Body::from_stream(body),
+                )
+                    .into_response()
+            }),
+        );
+        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            axum::serve(listener, app.into_make_service())
+                .await
+                .unwrap();
+        });
+        format!("http://{addr}/a2a")
+    }
+
+    #[tokio::test]
+    async fn a_completed_task_is_not_recorded_as_abandoned() {
+        use aisix_obs::UsageSink;
+        use futures::StreamExt;
+
+        // An A2A client stops reading at the terminal event; the agent may
+        // hold the connection open long after. Waiting for the upstream to
+        // close before calling the response delivered therefore reports every
+        // such COMPLETED task as a client hang-up — the exact inversion this
+        // PR's 499 is supposed to prevent.
+        let agent_url = spawn_completed_but_open_stream_agent().await;
+        let (tx, mut rx) = tokio::sync::mpsc::channel(8);
+        let handle = SnapshotHandle::new(snapshot_with(&agent_url, true, serde_json::json!(["*"])));
+        let hub = Arc::new(aisix_gateway::Hub::new());
+        let state = ProxyState::new(handle, hub, &proxy_cfg())
+            .without_cache()
+            .with_usage_sink(UsageSink::new(tx));
+        let response = build_router(state)
+            .oneshot(
+                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":1,"method":"message/stream"}"#,
+                    ))
+                    .unwrap(),
+            )
+            .await
+            .expect("router responds");
+
+        // Read exactly the terminal event, then stop — what a conforming
+        // client does.
+        let mut body = response.into_body().into_data_stream();
+        let chunk = body.next().await.expect("the terminal event arrives");
+        assert!(String::from_utf8_lossy(&chunk.unwrap()).contains("completed"));
+        drop(body);
+        tokio::task::yield_now().await;
+
+        let event = rx.try_recv().expect("a usage event is emitted");
+        assert_eq!(event.a2a_task_state, "completed");
+        assert_eq!(
+            event.status_code,
+            StatusCode::OK.as_u16(),
+            "a fully delivered stream is a success, not an abandoned one"
+        );
+    }
+
     #[tokio::test]
     async fn a_failed_call_still_records_the_task_it_asked_about() {
         // The upstream is unreachable, so no response ever names the task.
@@ -960,8 +1087,11 @@ mod tests {
         let app = axum::Router::new().route(
             "/a2a",
             axum::routing::post(|| async {
+                // A pause before the first event, so a time-to-first-event
+                // of 0 cannot pass for one that was never stamped.
+                tokio::time::sleep(Duration::from_millis(20)).await;
                 let chunks: Vec<Result<String, std::convert::Infallible>> =
-                    ["submitted", "working", "input-required"]
+                    ["submitted", "working", "completed"]
                         .iter()
                         .map(|state| {
                             let event = serde_json::json!({
@@ -1010,11 +1140,17 @@ mod tests {
         assert_eq!(event.a2a_operation, "message/stream");
         assert_eq!(event.a2a_task_id, "task-77");
         assert_eq!(event.a2a_context_id, "ctx-77");
-        assert_eq!(event.a2a_task_state, "input-required");
+        assert_eq!(event.a2a_task_state, "completed");
         // A stream read to the end is a success, and every event it relayed
         // is counted.
         assert_eq!(event.status_code, StatusCode::OK.as_u16());
         assert_eq!(event.a2a_stream_event_count, 3);
+        // The agent paused before speaking, so a stamped time-to-first-event
+        // is non-zero — an unstamped one would also read as 0.
+        assert!(
+            event.upstream_ttft_ms > 0,
+            "the first event's arrival must be stamped"
+        );
     }
 
     #[tokio::test]
diff --git a/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts b/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
index 563638163..3ceb40b9c 100644
--- a/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
+++ b/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
@@ -246,7 +246,12 @@ describe("a2a protocol telemetry e2e (AISIX-Cloud#1215)", () => {
     // millisecond bound, so the assertion does not track the stub's pacing.
     const ttfb = span.attributes["aisix.upstream_ttft_ms"] as number;
     const total = span.attributes["aisix.downstream_latency_ms"] as number;
-    expect(ttfb).toBeGreaterThan(0);
+    // The stub waits before each of its four writes, so an honest
+    // time-to-first-event lands near a quarter of the call. Both bounds
+    // matter: without the floor a regression that stamped the response head
+    // instead of the first event would still pass, and without the ceiling
+    // one that timed the whole stream would.
+    expect(ttfb).toBeGreaterThan(total / 6);
     expect(ttfb).toBeLessThan(total / 2);
   });
 

From 1c3b098b87646305f3fbdc872687b55351a42243 Mon Sep 17 00:00:00 2001
From: Jarvis <jarvis@api7.ai>
Date: Tue, 11 Aug 2026 19:29:11 +0800
Subject: [PATCH 3/3] fix(a2a): observe the latency histogram on every path,
 not only committed streams
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The previous commit put `record_request_e2e_latency` in the stream's drop
guard, which only exists once a stream has committed. So the unary path,
the quota rejection and a stream that failed to open never observed it —
`/a2a` would have shown a streaming-only, success-biased sample with no
failures in it at all, which is worse than the gap it was closing.

Moved to `emit_a2a_usage`, the chokepoint every A2A path already goes
through, with `streaming` derived from the canonical operation. The e2e
asserts the series covers `endpoint="/a2a"` and drives a failing call
through it.

Also corrects the streamed-call timing assertion: three counted events
with equal pauses puts an honest time-to-first-event near a THIRD of the
call, not a quarter. The bounds now sit far from that third in the
direction of the regression each catches (a response-head stamp reads
near zero, a whole-stream timer near the total), so the ratio has to move
by more than half its own value before jitter can trip either.
---
 crates/aisix-proxy/src/a2a.rs                 | 36 +++++++++----------
 .../cases/a2a-protocol-telemetry-e2e.test.ts  | 31 ++++++++++++----
 2 files changed, 42 insertions(+), 25 deletions(-)

diff --git a/crates/aisix-proxy/src/a2a.rs b/crates/aisix-proxy/src/a2a.rs
index 854f78268..b0194c436 100644
--- a/crates/aisix-proxy/src/a2a.rs
+++ b/crates/aisix-proxy/src/a2a.rs
@@ -337,23 +337,6 @@ impl Drop for StreamUsageOnDrop {
         } else {
             crate::CLIENT_CLOSED_REQUEST
         };
-        let elapsed = self.started.elapsed();
-        // The client-perceived duration of a streamed call, which nothing else
-        // records: the handler returns the moment the response head is out, so
-        // `aisix_proxy_request_duration_seconds` times only how long the
-        // stream took to OPEN. Same placement as the LLM streaming paths,
-        // which observe this histogram at stream completion for the same
-        // reason.
-        self.state.metrics.record_request_e2e_latency(
-            aisix_obs::LatencyLabels {
-                endpoint: "/a2a",
-                model: A2A_MODEL_LABEL,
-                provider: "a2a",
-                status,
-                streaming: true,
-            },
-            elapsed,
-        );
         emit_a2a_usage(
             &self.state,
             &self.auth,
@@ -361,7 +344,7 @@ impl Drop for StreamUsageOnDrop {
             &self.agent,
             &self.call,
             status,
-            elapsed,
+            self.started.elapsed(),
         );
     }
 }
@@ -654,6 +637,23 @@ fn emit_a2a_usage(
             .unwrap_or_default(),
         ..Default::default()
     };
+    // The client-perceived duration of the call. Nothing else records it for
+    // `/a2a`: the handler returns the moment a stream's response head is out,
+    // so `aisix_proxy_request_duration_seconds` times only how long a stream
+    // took to OPEN. Recorded here rather than at the stream's drop guard so
+    // the unary, quota-rejected and failed-to-open paths are in the sample
+    // too — a streaming-only series would report `/a2a` as having no failures
+    // at all.
+    state.metrics.record_request_e2e_latency(
+        aisix_obs::LatencyLabels {
+            endpoint: "/a2a",
+            model: A2A_MODEL_LABEL,
+            provider: "a2a",
+            status: status_code,
+            streaming: is_streaming_operation(call.operation),
+        },
+        latency,
+    );
     // The `aisix_a2a_*` family rides on the same chokepoint as the usage
     // event, so a path that accounts for a call cannot skip metering it.
     state.metrics.record_a2a_call(
diff --git a/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts b/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
index 3ceb40b9c..bf3a710ac 100644
--- a/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
+++ b/tests/e2e/src/cases/a2a-protocol-telemetry-e2e.test.ts
@@ -246,13 +246,15 @@ describe("a2a protocol telemetry e2e (AISIX-Cloud#1215)", () => {
     // millisecond bound, so the assertion does not track the stub's pacing.
     const ttfb = span.attributes["aisix.upstream_ttft_ms"] as number;
     const total = span.attributes["aisix.downstream_latency_ms"] as number;
-    // The stub waits before each of its four writes, so an honest
-    // time-to-first-event lands near a quarter of the call. Both bounds
-    // matter: without the floor a regression that stamped the response head
-    // instead of the first event would still pass, and without the ceiling
-    // one that timed the whole stream would.
-    expect(ttfb).toBeGreaterThan(total / 6);
-    expect(ttfb).toBeLessThan(total / 2);
+    // The stub pauses equally before each of its three counted events, so an
+    // honest time-to-first-event lands near a THIRD of the call. Both bounds
+    // matter and each is placed far from that third rather than close to it:
+    // a regression that stamped the response head reads near zero and has to
+    // clear the floor, one that timed the whole stream reads near `total` and
+    // has to clear the ceiling, while ordinary scheduling jitter — which
+    // moves the ratio by well under either margin — cannot reach either.
+    expect(ttfb).toBeGreaterThan(total / 8);
+    expect(ttfb).toBeLessThan((total * 3) / 4);
   });
 
   test("the a2a metric family slices by agent and operation", async (ctx) => {
@@ -279,6 +281,21 @@ describe("a2a protocol telemetry e2e (AISIX-Cloud#1215)", () => {
     );
     expect(text).toContain('aisix_a2a_task_state_total{agent="invoices",state="completed"}');
     expect(text).toContain("aisix_a2a_ttfb_seconds_bucket");
+    // The client-perceived duration series has to cover `/a2a` at all — and
+    // cover the calls that FAILED, not only the streams that opened. A
+    // streaming-only sample would report the endpoint as having no failures.
+    expect(text).toMatch(/aisix_request_e2e_latency_seconds_bucket\{[^}]*endpoint="\/a2a"/);
+    await call("does-not-exist-agent", {
+      jsonrpc: "2.0",
+      id: 8,
+      method: "message/send",
+      params: { message: { role: "user", parts: [] } },
+    });
+    await call("invoices", { jsonrpc: "2.0", id: 9, method: "message/send", params: {} });
+    const withFailures = await (await fetch(`${app.metricsUrl}/metrics`)).text();
+    expect(withFailures).toMatch(
+      /aisix_a2a_requests_total\{agent="invoices",operation="message\/send",status="2xx"\}/,
+    );
     // The ids that make a task traceable are exactly the ones that must not
     // become label values.
     expect(text).not.toMatch(/aisix_a2a_[a-z_]*\{[^}]*task_id=/);