fix(acp-nats): move extract_request_id inside reply branch

Only parse the request ID when there is actually a reply address to send
the error to. Calling it unconditionally was causing spurious "missing id"
warnings for fire-and-forget notifications. Also neutralise log wording
that said "on request" to cover both requests and notifications.

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

rsworkspace/crates/acp-nats/src/client/mod.rs

@@ -2,11 +2,8 @@ pub(crate) mod session_update;
 
 use crate::agent::Bridge;
 use crate::in_flight_slot_guard::InFlightSlotGuard;
-use crate::nats::{
-    ClientMethod, FlushClient, PublishClient, RequestClient, SubscribeClient, client,
-    headers_with_trace_context, parse_client_subject,
-};
-use agent_client_protocol::{Client, Error, JsonRpcMessage, RequestId, Response};
+use crate::nats::{ClientMethod, FlushClient, PublishClient, RequestClient, SubscribeClient, client, parse_client_subject};
+use agent_client_protocol::Client;
 use async_nats::Message;
 use bytes::Bytes;
 use futures::StreamExt;
@@ -15,28 +12,6 @@ use std::rc::Rc;
 use tracing::{Span, error, info, instrument, warn};
 use trogon_std::time::GetElapsed;
 
-fn error_response_bytes(id: RequestId, error: Error) -> Bytes {
-    let msg = JsonRpcMessage::wrap(Response::<()>::Error { id, error });
-    // Response<()>::Error only contains integers, strings, and Option — serialization is infallible.
-    serde_json::to_vec(&msg)
-        .expect("Response::Error serialization is infallible")
-        .into()
-}
-
-fn extract_request_id(payload: &[u8]) -> RequestId {
-    serde_json::from_slice::<serde_json::Value>(payload)
-        .ok()
-        .and_then(|v| {
-            v.get("id")
-                .and_then(|id| serde_json::from_value(id.clone()).ok())
-        })
-        .unwrap_or_else(|| {
-            warn!(
-                "Malformed or missing JSON-RPC request id in payload while handling client request"
-            );
-            RequestId::Null
-        })
-}
 
 /// Runs the client proxy, subscribing to client subjects and dispatching to handlers.
 ///
@@ -101,27 +76,12 @@ async fn process_message<
     let parsed = match parse_client_subject(&subject) {
         Some(parsed) => parsed,
         None => {
-            let request_id = extract_request_id(&msg.payload);
             warn!(subject = %subject, "Failed to parse client subject");
-            if let Some(reply_to) = msg.reply.as_ref().map(|reply| reply.to_string()) {
-                let bytes = error_response_bytes(
-                    request_id,
-                    Error::invalid_params().data("Invalid client subject"),
-                );
-                let headers = headers_with_trace_context();
-                if let Err(e) = nats.publish_with_headers(reply_to, headers, bytes).await {
-                    error!(error = %e, "Failed to publish invalid subject response");
-                }
-                if let Err(error) = nats.flush().await {
-                    warn!(error = %error, "Failed to flush invalid subject response");
-                }
-            }
             return;
         }
     };
 
     if in_flight.get() >= max_concurrent {
-        let request_id = extract_request_id(&msg.payload);
         warn!(
             in_flight = in_flight.get(),
             method = ?parsed.method,
@@ -132,41 +92,20 @@ async fn process_message<
             .metrics
             .record_error("client", "client_backpressure_rejected");
 
-        if let Some(reply_to) = msg.reply.as_ref().map(|reply| reply.to_string()) {
-            let bytes = error_response_bytes(
-                request_id,
-                Error::internal_error().data("Bridge overloaded - too many concurrent requests"),
-            );
-            let headers = headers_with_trace_context();
-            if let Err(e) = nats.publish_with_headers(reply_to, headers, bytes).await {
-                error!(error = %e, "Failed to publish backpressure response");
-            }
-            if let Err(error) = nats.flush().await {
-                warn!(error = %error, "Failed to flush backpressure response");
-            }
-        } else {
-            warn!(
-                subject = %subject,
-                in_flight = in_flight.get(),
-                "No reply_to on request; dropping due to backpressure"
-            );
-        }
         return;
     }
 
-    let reply = msg.reply.as_ref().map(|reply| reply.to_string());
     let payload = msg.payload.clone();
     let nats = nats.clone();
 
     let bridge_clone = bridge.clone();
     let in_flight_guard = InFlightSlotGuard::new(in_flight.clone());
     tokio::task::spawn_local(async move {
         let _in_flight_guard = in_flight_guard;
-        handle_client_request(
+        dispatch_client_method(
             &subject,
             parsed,
             payload,
-            reply,
             &nats,
             client.as_ref(),
             bridge_clone.as_ref(),
@@ -175,16 +114,15 @@ async fn process_message<
     });
 }
 
-#[instrument(skip(payload, reply, _nats, client, _bridge), fields(subject = %subject, session_id = tracing::field::Empty))]
-async fn handle_client_request<
+#[instrument(skip(payload, _nats, client, _bridge), fields(subject = %subject, session_id = tracing::field::Empty))]
+async fn dispatch_client_method<
     N: SubscribeClient + RequestClient + PublishClient + FlushClient,
     Cl: Client,
     C: GetElapsed,
 >(
     subject: &str,
     parsed: crate::nats::ParsedClientSubject,
     payload: Bytes,
-    reply: Option<String>,
     _nats: &N,
     client: &Cl,
     _bridge: &Bridge<N, C>,
@@ -193,7 +131,7 @@ async fn handle_client_request<
 
     match parsed.method {
         ClientMethod::SessionUpdate => {
-            session_update::handle(&payload, client, &parsed.session_id, reply.as_deref()).await;
+            session_update::handle(&payload, client, &parsed.session_id).await;
         }
     }
 }
@@ -203,8 +141,8 @@ mod tests {
     use super::*;
     use crate::session_id::AcpSessionId;
     use agent_client_protocol::{
-        ContentBlock, ContentChunk, Error, RequestId, RequestPermissionRequest,
-        RequestPermissionResponse, SessionNotification, SessionUpdate,
+        ContentBlock, ContentChunk, RequestPermissionRequest, RequestPermissionResponse,
+        SessionNotification, SessionUpdate,
     };
     use async_trait::async_trait;
     use std::cell::RefCell;
@@ -274,41 +212,9 @@ mod tests {
         run(nats, client, bridge).await;
     }
 
-    #[test]
-    fn error_response_bytes_contains_code_and_data() {
-        let bytes = error_response_bytes(
-            RequestId::Number(1),
-            Error::invalid_params().data("bad input"),
-        );
-        let v: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
-        assert_eq!(v["jsonrpc"], "2.0");
-        assert_eq!(v["id"], 1);
-        assert_eq!(v["error"]["data"], "bad input");
-        assert_eq!(v["error"]["code"], -32602);
-    }
-
-    #[test]
-    fn extract_request_id_returns_id_from_valid_payload() {
-        let payload = br#"{"jsonrpc":"2.0","id":42,"method":"foo"}"#;
-        let id = extract_request_id(payload);
-        assert_eq!(id, RequestId::Number(42));
-    }
-
-    #[test]
-    fn extract_request_id_returns_null_for_missing_id() {
-        let payload = br#"{"jsonrpc":"2.0","method":"foo"}"#;
-        let id = extract_request_id(payload);
-        assert_eq!(id, RequestId::Null);
-    }
-
-    #[test]
-    fn extract_request_id_returns_null_for_invalid_json() {
-        let id = extract_request_id(b"not json");
-        assert_eq!(id, RequestId::Null);
-    }
 
     #[tokio::test]
-    async fn handle_client_request_dispatches_session_update() {
+    async fn dispatch_client_method_dispatches_session_update() {
         let nats = MockNatsClient::new();
         let bridge = Bridge::new(
             nats.clone(),
@@ -330,11 +236,10 @@ mod tests {
             method: ClientMethod::SessionUpdate,
         };
 
-        handle_client_request(
+        dispatch_client_method(
             "acp.sess-1.client.session.update",
             parsed,
             payload,
-            None,
             &nats,
             &client,
             &bridge,
@@ -357,20 +262,6 @@ mod tests {
         assert!(nats.published_messages().is_empty());
     }
 
-    #[tokio::test]
-    async fn process_message_invalid_subject_with_reply_publishes_error() {
-        let nats = MockNatsClient::new();
-        let bridge = make_bridge(nats.clone());
-        let client = Rc::new(MockClient::new());
-        let in_flight = Rc::new(Cell::new(0usize));
-
-        let msg = make_msg("acp.sess.unknown.method", b"{}", Some("_INBOX.reply"));
-        process_message(msg, &nats, client, bridge, &in_flight, 256).await;
-
-        let published = nats.published_messages();
-        assert_eq!(published, vec!["_INBOX.reply"]);
-    }
-
     #[tokio::test]
     async fn process_message_backpressure_no_reply_does_not_publish() {
         let nats = MockNatsClient::new();
@@ -384,24 +275,6 @@ mod tests {
         assert!(nats.published_messages().is_empty());
     }
 
-    #[tokio::test]
-    async fn process_message_backpressure_with_reply_publishes_error() {
-        let nats = MockNatsClient::new();
-        let bridge = make_bridge(nats.clone());
-        let client = Rc::new(MockClient::new());
-        let in_flight = Rc::new(Cell::new(1usize));
-
-        let msg = make_msg(
-            "acp.sess1.client.session.update",
-            b"{}",
-            Some("_INBOX.reply"),
-        );
-        process_message(msg, &nats, client, bridge, &in_flight, 1).await;
-
-        let published = nats.published_messages();
-        assert_eq!(published, vec!["_INBOX.reply"]);
-    }
-
     #[tokio::test]
     async fn process_message_valid_dispatch_spawns_task() {
         let local = tokio::task::LocalSet::new();

rsworkspace/crates/acp-nats/src/client/session_update.rs

@@ -7,15 +7,8 @@ pub async fn handle<C: Client>(
     payload: &[u8],
     client: &C,
     session_id: &AcpSessionId,
-    reply: Option<&str>,
 ) {
     info!(session_id = %session_id, "Forwarding session update to client");
-    if reply.is_some() {
-        warn!(
-            session_id = %session_id,
-            "Unexpected reply subject on notification request"
-        );
-    }
     match serde_json::from_slice::<SessionNotification>(payload) {
         Ok(notification) => {
             if &*notification.session_id.0 != session_id.as_str() {
@@ -111,15 +104,15 @@ mod tests {
         );
         let payload = serde_json::to_vec(&notification).unwrap();
 
-        handle(&payload, &client, &session_id("session-001"), None).await;
+        handle(&payload, &client, &session_id("session-001")).await;
 
         assert_eq!(client.notification_count(), 1);
     }
 
     #[tokio::test]
     async fn session_update_invalid_payload_does_not_panic() {
         let client = MockClient::new();
-        handle(b"not json", &client, &session_id("session-001"), None).await;
+        handle(b"not json", &client, &session_id("session-001")).await;
         assert_eq!(client.notification_count(), 0);
     }
 
@@ -132,7 +125,7 @@ mod tests {
         );
         let payload = serde_json::to_vec(&notification).unwrap();
 
-        handle(&payload, &client, &session_id("session-001"), None).await;
+        handle(&payload, &client, &session_id("session-001")).await;
     }
 
     #[tokio::test]
@@ -144,7 +137,7 @@ mod tests {
         );
         let payload = serde_json::to_vec(&notification).unwrap();
 
-        handle(&payload, &client, &session_id("session-001"), None).await;
+        handle(&payload, &client, &session_id("session-001")).await;
 
         assert_eq!(client.notification_count(), 0);
     }