test(acp-nats): cover Display and Error::source for all TerminalWaitForExitError variants

Author: yordis

.gitignore

@@ -40,6 +40,7 @@ Thumbs.db
 
 # Coverage
 lcov.info
+*.lcov
 coverage.xml
 coverage-*.xml
 *.profraw

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

@@ -14,7 +14,7 @@ use crate::in_flight_slot_guard::InFlightSlotGuard;
 use crate::jsonrpc::extract_request_id;
 use crate::nats::{
     ClientMethod, FlushClient, PublishClient, RequestClient, SubscribeClient, client,
-    headers_with_trace_context, parse_client_subject,
+    parse_client_subject,
 };
 use agent_client_protocol::{Client, ErrorCode};
 use async_nats::Message;
@@ -26,32 +26,6 @@ use tracing::{Span, error, info, instrument, warn};
 use trogon_std::JsonSerialize;
 use trogon_std::time::GetElapsed;
 
-fn jsonrpc_error_response(
-    request_id: agent_client_protocol::RequestId,
-    code: ErrorCode,
-    message: &str,
-) -> Bytes {
-    let id_value = serde_json::to_value(&request_id).unwrap_or(serde_json::Value::Null);
-    let response = serde_json::json!({
-        "jsonrpc": "2.0",
-        "id": id_value,
-        "error": {
-            "code": i32::from(code),
-            "message": message
-        }
-    });
-    serde_json::to_vec(&response)
-        .unwrap_or_else(|e| {
-            format!(
-                "{{\"jsonrpc\":\"2.0\",\"id\":null,\"error\":{{\"code\":{},\"message\":\"failed to serialize error response: {}\"}}}}",
-                i32::from(code),
-                e
-            )
-            .into_bytes()
-        })
-        .into()
-}
-
 async fn publish_backpressure_error_reply<N: PublishClient + FlushClient, S: JsonSerialize>(
     nats: &N,
     payload: &[u8],
@@ -281,28 +255,25 @@ async fn dispatch_client_method<
             .await;
         }
         ClientMethod::TerminalWaitForExit => {
-            let result = terminal_wait_for_exit::handle(
-                &payload,
-                ctx.client,
-                parsed.session_id.as_str(),
-                ctx.bridge.config.operation_timeout(),
-            )
-            .await;
             if let Some(reply_to) = &reply {
+                let result = terminal_wait_for_exit::handle(
+                    &payload,
+                    ctx.client,
+                    parsed.session_id.as_str(),
+                    ctx.bridge.config.operation_timeout(),
+                )
+                .await;
                 let request_id = extract_request_id(&payload);
                 match result {
                     Ok(bytes) => {
-                        let headers = headers_with_trace_context();
-                        if let Err(e) = ctx
-                            .nats
-                            .publish_with_headers(reply_to.clone(), headers, bytes.into())
-                            .await
-                        {
-                            error!(error = %e, "Failed to publish reply");
-                        }
-                        if let Err(e) = ctx.nats.flush().await {
-                            warn!(error = %e, "Failed to flush response");
-                        }
+                        rpc_reply::publish_reply(
+                            ctx.nats,
+                            reply_to,
+                            bytes.into(),
+                            rpc_reply::CONTENT_TYPE_JSON,
+                            "terminal/wait_for_exit reply",
+                        )
+                        .await;
                     }
                     Err(e) => {
                         let (code, message) = terminal_wait_for_exit::error_code_and_message(&e);
@@ -312,20 +283,27 @@ async fn dispatch_client_method<
                             session_id = %parsed.session_id,
                             "Failed to handle terminal/wait_for_exit"
                         );
-                        let bytes = jsonrpc_error_response(request_id, code, &message);
-                        let headers = headers_with_trace_context();
-                        if let Err(e) = ctx
-                            .nats
-                            .publish_with_headers(reply_to.clone(), headers, bytes)
-                            .await
-                        {
-                            error!(error = %e, "Failed to publish error reply");
-                        }
-                        if let Err(e) = ctx.nats.flush().await {
-                            warn!(error = %e, "Failed to flush error reply");
-                        }
+                        let (bytes, content_type) = rpc_reply::error_response_bytes(
+                            ctx.serializer,
+                            request_id,
+                            code,
+                            &message,
+                        );
+                        rpc_reply::publish_reply(
+                            ctx.nats,
+                            reply_to,
+                            bytes,
+                            content_type,
+                            "terminal/wait_for_exit error reply",
+                        )
+                        .await;
                     }
                 }
+            } else {
+                warn!(
+                    session_id = %parsed.session_id,
+                    "terminal/wait_for_exit requires reply subject; ignoring message"
+                );
             }
         }
     }
@@ -840,6 +818,44 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn dispatch_client_method_terminal_wait_for_exit_reply_none_skips_handler() {
+        let nats = MockNatsClient::new();
+        let client = MockClient::new();
+        let session_id = AcpSessionId::new("sess-1").unwrap();
+
+        let envelope = Request {
+            id: RequestId::Number(1),
+            method: std::sync::Arc::from("terminal/wait_for_exit"),
+            params: Some(WaitForTerminalExitRequest::new("sess-1", "term-001")),
+        };
+        let payload = bytes::Bytes::from(serde_json::to_vec(&envelope).unwrap());
+
+        let parsed = crate::nats::ParsedClientSubject {
+            session_id,
+            method: ClientMethod::TerminalWaitForExit,
+        };
+
+        let bridge = make_bridge(nats.clone());
+        let ctx = DispatchContext {
+            nats: &nats,
+            client: &client,
+            bridge: &bridge,
+            serializer: &StdJsonSerialize,
+        };
+        dispatch_client_method(
+            "acp.sess-1.client.terminal.wait_for_exit",
+            parsed,
+            payload,
+            None,
+            &ctx,
+        )
+        .await;
+
+        assert!(nats.published_messages().is_empty());
+        assert_eq!(client.wait_for_terminal_exit_call_count(), 0);
+    }
+
     #[tokio::test]
     async fn dispatch_client_method_dispatches_terminal_wait_for_exit() {
         let nats = MockNatsClient::new();

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

@@ -365,6 +365,18 @@ mod tests {
     fn terminal_wait_for_exit_error_display() {
         let err = TerminalWaitForExitError::TimedOut;
         assert!(err.to_string().contains("Timed out"));
+        let json_err =
+            TerminalWaitForExitError::MalformedJson(serde_json::from_str::<()>("{").unwrap_err());
+        assert!(json_err.to_string().contains("malformed JSON"));
+        let params_err = TerminalWaitForExitError::InvalidParams(
+            agent_client_protocol::Error::new(-32602, "bad params"),
+        );
+        assert!(params_err.to_string().contains("invalid params"));
+        let client_err = TerminalWaitForExitError::ClientError(agent_client_protocol::Error::new(
+            -32603,
+            "client failed",
+        ));
+        assert!(client_err.to_string().contains("client error"));
     }
 
     #[test]
@@ -374,5 +386,14 @@ mod tests {
         let json_err =
             TerminalWaitForExitError::MalformedJson(serde_json::from_str::<()>("{").unwrap_err());
         assert!(json_err.source().is_some());
+        let params_err = TerminalWaitForExitError::InvalidParams(
+            agent_client_protocol::Error::new(-32602, "bad params"),
+        );
+        assert!(params_err.source().is_some());
+        let client_err = TerminalWaitForExitError::ClientError(agent_client_protocol::Error::new(
+            -32603,
+            "client failed",
+        ));
+        assert!(client_err.source().is_some());
     }
 }