fix(a2a): let one JSON-RPC envelope define the wire at every edge

docs/.vitepress/config.mts

@@ -62,6 +62,10 @@ export default async () => {
             { text: "Bring Your Own OpenBao", link: "/how-to/bring-your-own-openbao" },
             { text: "Migrate Between Key Backends", link: "/how-to/migrate-key-backends" },
             { text: "Handle Unusable Keys", link: "/how-to/handle-unusable-keys" },
+            {
+              text: "Retire the ACP Notifications Stream",
+              link: "/how-to/retire-acp-notifications-stream",
+            },
           ],
         },
         {

docs/adr/0055-nats-subject-design-jsonrpc-bindings.md

@@ -282,7 +282,7 @@ the configured prefix, and no ACP or A2A code reads a protocol version at the
 transport layer today. Record the rule now; skip the implementation.
 
 **Revisit trigger.** The first time a durable stream must survive a protocol
-version bump. ACP's `COMMANDS`, `RESPONSES`, and `NOTIFICATIONS` streams are
+version bump. ACP's `COMMANDS`, `RESPONSES`, and `CLIENT_OPS` streams are
 `Limits` retention with `max_age`, so stored messages outlive their connection.
 If a payload shape changes across an ACP version, a replaying consumer has no
 way to pick the right schema, and `v{major}` cannot help because it versions
@@ -309,9 +309,16 @@ MUST NOT mint a per-request subject such as `...response.{req_id}`.
 
 ACP terminal durable results use one subject shape:
 `{prefix}.v{major}.session.{session_id}.agent.response`. The historical
-`...agent.prompt.response.{req_id}` path collapses into that subject; mid-flight
-prompt progress remains on `...agent.update`. Partitioning terminal results by
-method is unnecessary once demux is on the correlation token.
+`...agent.prompt.response.{req_id}` path collapses into that subject. Partitioning
+terminal results by method is unnecessary once demux is on the correlation token.
+
+Mid-flight progress is *not* a durable agent response. `session/update` is a
+client-directed notification under the ACP method surface, so it takes the
+client-op terminal `{prefix}.v{major}.session.{session_id}.client.session.update`
+alongside every other agent-to-client call, and rides the `CLIENT_OPS` stream.
+Routing it under `...agent.update` instead would fork one method across two role
+segments and scope progress to prompts only, which drops the `session/load`
+replay that emits the same notification.
 
 ### Streams
 
@@ -369,21 +376,27 @@ this ADR's cardinality, `v{major}`, limits, and left-prefix rules. They use an
 - Entity tokens stay long-lived (agent, caller, task). Per-request identifiers
   remain forbidden.
 
-#### Audit subject defects (current code)
+#### Audit subject defects
 
-Today's audit emitters are non-conformant and MUST be corrected as part of
-baseline alignment:
+Audit subjects MUST be entity-scoped with fixed terminals, for example
+`a2a.v1.audit.{agent_id}.{outcome}` (method in the payload or a header), not
+method-led growth without an entity token. Two defects motivated the rule:
 
 - `a2a.audit.{outcome}.{method}` grows with the method set and embeds the method
   as a subject token, defeating the fixed-terminal rule.
-- The emitter accepts `agent_id` and then drops it (`let _ = agent_id`), so audit
-  traffic cannot be filtered per agent.
+- An emitter that accepts `agent_id` and then drops it (`let _ = agent_id`)
+  leaves audit traffic unfilterable per agent.
+
+Exact terminals are an implementation detail under this profile; the two defects
+above are the conformance requirement.
 
-Target shape is entity-scoped, for example
-`a2a.v1.audit.{agent_id}.{outcome}` (fixed outcome terminals; method in the
-payload or a header), not method-led growth without an entity token. Exact
-terminals are an implementation detail under this profile; the defects above are
-the conformance requirement.
+The A2A emitter (`a2a-nats::audit::emitter`) now conforms, publishing
+`{prefix}.v1.audit.{agent_id}.{ok|err}` and
+`{prefix}.v1.audit.{agent_id}.lifecycle`. The gateway's ingress audit builder
+(`a2a-gateway::audit_ingress::ingress_audit_subject`) still emits
+`{prefix}.a2a.audit.{outcome}.ingress.{skill}` — duplicated root, no `v{major}`,
+skill as a growing terminal, no entity token. It has no production caller today,
+so it is corrected when the ingress audit path is wired rather than ahead of it.
 
 ### Limits
 
@@ -437,10 +450,13 @@ non-conformant with this standard; they are not the standard:
   prefix.
 - **Request id in ACP response and update subjects**
   (`...agent.response.{req_id}`, `...agent.update.{req_id}`) and the prompt-
-  specific `...agent.prompt.response.{req_id}`. Replace with entity-scoped
-  `...agent.response` and `...agent.update`, correlating on ACP's
-  transport-minted JSON-RPC `id`. Collapse the prompt-specific terminal into
-  `...agent.response`.
+  specific `...agent.prompt.response.{req_id}`. Replace with the entity-scoped
+  `...agent.response`, correlating on ACP's transport-minted JSON-RPC `id`, and
+  collapse the prompt-specific terminal into it.
+- **Duplicate ACP progress subject.** `...agent.update` and
+  `...client.session.update` both claimed `session/update`, and only the latter
+  had a publisher. Remove `...agent.update` and its `NOTIFICATIONS` stream;
+  progress rides the client-op subtree.
 - **Request id in A2A task event subjects**
   (`...tasks.{task_id}.events.{req_id}`). Replace with
   `...tasks.{task_id}.events` plus `Trogon-Req-Id` (A2A ids are peer-supplied). This

docs/adr/0056-canonical-jsonrpc-bodies-over-nats.md

@@ -34,7 +34,7 @@ now needs:
    message. This is *envelope* reconstruction and is unrelated to
    [ADR#0021](./0021-typed-decode-over-passthrough-forwarding.md), which rejected
    *payload* passthrough (forwarding `params` as an untyped `Value` instead of
-   decoding it). [ADR#0021](0021-typed-decode-over-passthrough-forwarding.md) is unaffected by this ADR; see §7.
+   decoding it). [ADR#0021](0021-typed-decode-over-passthrough-forwarding.md) is unaffected by this ADR; see §8.
 
 [ADR#0041](./0041-canonical-mcp-jsonrpc-bodies-over-nats.md) (draft) already
 moves MCP to canonical full-envelope bodies with non-authoritative header
@@ -148,7 +148,34 @@ This ADR adds only that the `id` is application-level, travels authoritatively i
 the body, and is projected to `Jsonrpc-Id`. `Nats-Msg-Id` stays reserved for
 JetStream deduplication.
 
-### 5. MCP-specific rules carried from [ADR#0041](0041-canonical-mcp-jsonrpc-bodies-over-nats.md)
+### 5. A streaming chunk is a response, not a notification
+
+A2A streams `message/stream` and `tasks/resubscribe` as a sequence of JSON-RPC
+**success responses that repeat the request id**, terminated by the one whose
+result is final. Each chunk on the NATS leg carries that complete object:
+
+```json
+{"jsonrpc": "2.0", "id": "<request id>", "result": { "statusUpdate": { } }}
+```
+
+This is not a stylistic choice. A server push carrying both an `id` and a
+`method` is a *request* under JSON-RPC, and one carrying a `method` without an
+`id` is a notification the caller cannot correlate to the subscription it opened.
+Only the response shape says "this belongs to the call you made" without
+inventing a member the specification does not define.
+
+Because the chunk is complete at the point of publish, every hop after it
+forwards the bytes: the gateway egress pump, the HTTP bridge, the SSE facade,
+and the stdio bridge each emit the body verbatim rather than each inventing an
+envelope. An edge that owns the caller's id (a bridge whose caller minted its own
+`id`) rewrites only that member.
+
+ACP is the opposite case and stays so: its mid-turn `session/update` really is a
+notification (id-less, method-bearing) and the terminal `PromptResponse` is the
+only message answering the request id. The difference is in the upstream
+protocols, not in this binding.
+
+### 6. MCP-specific rules carried from [ADR#0041](0041-canonical-mcp-jsonrpc-bodies-over-nats.md)
 
 These remain in force for MCP and are unchanged by the ACP/A2A migration:
 
@@ -159,14 +186,14 @@ These remain in force for MCP and are unchanged by the ACP/A2A migration:
   `Mcp-Name`, `Mcp-Param-*`. `Mcp-Session-Id` and unrelated HTTP headers are not
   forwarded. Those headers remain derived metadata validated against the body.
 
-### 6. Shared package; legacy content mode removed after migration
+### 7. Shared package; legacy content mode removed after migration
 
 The shared `jsonrpc-nats` package owns encode/decode. After ACP and A2A migrate
 to the canonical APIs, the legacy content-mode `encode`/`decode` are removed (or
 deprecated with a removal milestone). A bridge that hand-builds an envelope
 forwards the body unmodified instead, where that assembly is redundant.
 
-### 7. What this ADR does not change
+### 8. What this ADR does not change
 
 - **Typed payload decode stays.**
   [ADR#0021](./0021-typed-decode-over-passthrough-forwarding.md) (accepted) keeps
@@ -198,6 +225,8 @@ forwards the body unmodified instead, where that assembly is redundant.
   subject terminal that is not the body method's projection under the binding's
   method-to-terminal mapping.
 - `"jsonrpc":"2.0"` travels in the body; it is not duplicated into a header.
+- Every A2A stream chunk deserializes as a JSON-RPC success response bearing the
+  request id, on the NATS leg and on every edge that forwards it.
 - MCP `params._meta` and allowlisted `Mcp-*` / `MCP-Protocol-Version` headers
   survive an HTTP proxy to NATS round trip.
 

docs/how-to/retire-acp-notifications-stream.md

@@ -0,0 +1,81 @@
+# Retire the ACP Notifications Stream
+
+`session/update` reaches the client through the client-op proxy for every
+operation. The prompt-scoped notification path that used the
+`<PREFIX>_NOTIFICATIONS` stream was a second delivery path for the same
+updates, so it was removed, and the ACP provisioner no longer creates that
+stream.
+
+Removing a stream from the provisioner does not remove it from a deployment
+that already ran an earlier release. That deployment still has the stream, its
+stored messages, and its storage bill. This page is how you retire it.
+
+## When to use this
+
+Use this procedure once per ACP deployment, after every process has been
+upgraded to a release whose provisioner no longer lists the stream. Check the
+deployed process versions to establish that, then use this to see whether the
+retired stream is still there:
+
+```shell
+nats stream ls
+```
+
+You are retiring a stream whose name matches `retired_stream_names` for your
+prefix. For the default `acp` prefix that is `ACP_NOTIFICATIONS`; for a prefix
+of `my.multi.part` it is `MY_MULTI_PART_NOTIFICATIONS`.
+
+Do not use this procedure while any process is still running the previous
+release. Those processes publish to `<prefix>.v1.session.*.agent.update`, and
+deleting the stream underneath them drops those messages.
+
+## Preconditions
+
+- Every ACP agent and client process runs a release that does not provision
+  the stream. A mixed fleet is the one case where this procedure loses data.
+- The stream has no consumers with unacknowledged messages you still need.
+  `nats consumer ls <PREFIX>_NOTIFICATIONS` lists them; an empty list is the
+  state you want before deleting.
+- You have a JetStream account credential with delete authority on the
+  stream.
+
+## Steps
+
+1. Confirm the stream is idle. Its message count should stop growing:
+
+   ```shell
+   nats stream info <PREFIX>_NOTIFICATIONS
+   ```
+
+   A message count that still climbs means something is publishing to
+   `<prefix>.v1.session.*.agent.update`. Find it and upgrade it before
+   continuing.
+
+2. Delete the remaining consumers. Deleting the stream removes them anyway,
+   but doing it first makes a still-attached reader fail visibly here rather
+   than silently later:
+
+   ```shell
+   nats consumer rm <PREFIX>_NOTIFICATIONS <CONSUMER>
+   ```
+
+3. Delete the stream:
+
+   ```shell
+   nats stream rm <PREFIX>_NOTIFICATIONS
+   ```
+
+## What this does not do
+
+The provisioner does not perform any of the above. It creates and reconciles
+the streams it declares and touches nothing else, so a stream delete is never
+a side effect of a boot. That is deliberate: a stream delete is
+unrecoverable, it races an operator who may still be draining the stream, and
+a rollback to the previous release would re-create the stream empty and hide
+the fact that its history is gone.
+
+## Rollback
+
+There is none. A deleted stream and its messages do not come back. If you
+roll back to a release that still provisions the stream, the provisioner
+creates it again with no history.

rsworkspace/crates/a2a/a2a-bridge/src/constants.rs

@@ -1 +1,4 @@
 pub(crate) const AGENT_ID_HEADER: &str = "x-a2a-agent-id";
+
+/// JSON-RPC 2.0 reserved code for a server-side failure.
+pub(crate) const INTERNAL_ERROR: i32 = -32603;

rsworkspace/crates/a2a/a2a-bridge/src/inbound.rs

@@ -19,8 +19,10 @@ use axum::{
 use bytes::Bytes;
 use futures_util::StreamExt;
 use futures_util::stream::{self, BoxStream, Stream};
-use serde_json::{Value, json};
+use serde_json::{Map, Value, json};
+use tracing::warn;
 
+use a2a_nats::RequestId;
 use a2a_nats::constants::{GATEWAY_CALLER_ID_HEADER, GATEWAY_CALLER_ID_HTTP, REQ_ID_HEADER};
 use a2a_nats::jetstream::consumers::{PullConfig, resubscribe_consumer, stream_events_consumer};
 use a2a_nats::jetstream::streams::events_stream_name;
@@ -29,7 +31,7 @@ use a2a_nats::{A2aPrefix, A2aTaskId, ReqId};
 use a2a_auth_callout::{CALLER_JWT_HEADER_NAME, CallerJwtHeaderValue, MintedUserJwt};
 
 use crate::auth::AuthCalloutClient;
-use crate::constants::AGENT_ID_HEADER;
+use crate::constants::{AGENT_ID_HEADER, INTERNAL_ERROR};
 use crate::error::BridgeError;
 use crate::identity::{BridgeAgentId, BridgeUserJwt, CallerHttpsAuth};
 
@@ -475,32 +477,73 @@ impl TaskJetStreamPort for ScriptedTaskJetstream {
     }
 }
 
-fn sse_gateway_line(body: &[u8]) -> Event {
-    Event::default()
-        .event("gateway-bootstrap")
-        .data(String::from_utf8_lossy(body))
+/// Every A2A SSE frame is an unnamed `data:` line carrying a JSON-RPC response
+/// that repeats the caller's request id. Naming the events instead would put the
+/// bootstrap and the task chunks on distinct SSE event types, which a spec client
+/// never subscribes to, and the bodies already say which is which.
+///
+/// The id is restamped rather than forwarded, because the hops behind this edge
+/// correlate on their own transport id (`Trogon-Req-Id`, or a minted one when the
+/// caller sent no id at all). Only the caller's own id is meaningful to the
+/// caller. A payload that is not a JSON-RPC response leaves as a correlated
+/// error frame: a client parses every `data:` line as a JSON-RPC response, so
+/// anything else there is a parse failure rather than the diagnostic it looks
+/// like.
+///
+/// An event an older agent published carries no envelope of its own, so it is
+/// given one here instead of being restamped: forwarding it as-is would put a
+/// `data:` line on the wire that no spec client can parse as a response.
+fn sse_data_line(body: &[u8], caller_id: &Value) -> Event {
+    if let Some(response) = a2a_nats::task_event::legacy_event_as_response(body, caller_id) {
+        return Event::default().data(response.to_string());
+    }
+
+    let Some(mut envelope) = serde_json::from_slice::<Value>(body).ok().and_then(response_envelope) else {
+        warn!(
+            payload = %String::from_utf8_lossy(body),
+            "stream payload is not a JSON-RPC envelope"
+        );
+        return sse_error_frame(caller_id, "stream payload is not a JSON-RPC envelope".to_owned());
+    };
+    envelope.insert("id".to_owned(), caller_id.clone());
+    Event::default().data(Value::Object(envelope).to_string())
 }
 
-fn sse_task_line(body: &Bytes) -> Event {
-    Event::default()
-        .event("task-event")
-        .data(String::from_utf8_lossy(body.as_ref()))
+/// A JSON-RPC response carries exactly one of `result` and `error`. An object
+/// holding neither is a request, an empty body, or something else entirely, and
+/// stamping the caller's id onto it would emit a `data:` line that answers
+/// nothing.
+fn response_envelope(body: Value) -> Option<Map<String, Value>> {
+    let Value::Object(envelope) = body else {
+        return None;
+    };
+    (envelope.contains_key("result") != envelope.contains_key("error")).then_some(envelope)
+}
+
+fn sse_error_line(caller_id: &Value, err: &BridgeError) -> Event {
+    sse_error_frame(caller_id, err.to_string())
 }
 
-fn sse_error_line(err: &BridgeError) -> Event {
-    Event::default().event("error").data(err.to_string())
+fn sse_error_frame(caller_id: &Value, message: String) -> Event {
+    let envelope = serde_json::json!({
+        "jsonrpc": "2.0",
+        "id": caller_id,
+        "error": { "code": INTERNAL_ERROR, "message": message },
+    });
+    Event::default().data(envelope.to_string())
 }
 
 fn sse_from_bootstrap_and_payloads(
     bootstrap_owned: Vec<u8>,
     tail: Pin<Box<dyn Stream<Item = Result<Bytes, BridgeError>> + Send>>,
+    caller_id: Value,
 ) -> BoxStream<'static, Result<Event, Infallible>> {
-    let head_event = sse_gateway_line(&bootstrap_owned);
+    let head_event = sse_data_line(&bootstrap_owned, &caller_id);
     let head = futures_util::stream::once(futures_util::future::ready(Ok::<Event, Infallible>(head_event)));
-    let tail_mapped = tail.map(|item| {
+    let tail_mapped = tail.map(move |item| {
         Ok::<Event, Infallible>(match item {
-            Ok(chunk) => sse_task_line(&chunk),
-            Err(ref err) => sse_error_line(err),
+            Ok(chunk) => sse_data_line(chunk.as_ref(), &caller_id),
+            Err(ref err) => sse_error_line(&caller_id, err),
         })
     });
 
@@ -715,6 +758,13 @@ pub async fn handle_jsonrpc(headers: HeaderMap, body: bytes::Bytes, state: &AppS
     let req_id = json_rpc_corr_id(&v);
 
     if is_sse_jsonrpc_method(method) {
+        // Every SSE frame repeats this id, so a streaming request that carries
+        // none has no reply the caller can correlate. Rejecting here keeps the
+        // gateway from doing the work of a stream nobody can read.
+        let stream_caller_id = v
+            .get("id")
+            .and_then(|id| serde_json::from_value::<RequestId>(id.clone()).ok())
+            .ok_or(BridgeError::MissingJsonRpcId)?;
         // The unary publish comes BEFORE the JetStream consumer. Task event
         // subjects are scoped to the task (ADR#0055), and the bootstrap reply is
         // where the task id comes from. Nothing is lost in the gap, because
@@ -744,7 +794,7 @@ pub async fn handle_jsonrpc(headers: HeaderMap, body: bytes::Bytes, state: &AppS
             }
             None => Box::pin(futures_util::stream::empty()),
         };
-        let merged = sse_from_bootstrap_and_payloads(unary_reply.to_vec(), payloads);
+        let merged = sse_from_bootstrap_and_payloads(unary_reply.to_vec(), payloads, stream_caller_id.to_json());
         return Ok(Sse::new(merged).keep_alive(KeepAlive::default()).into_response());
     }