refactor(acp-nats): simplify client dispatch and terminal_output handling

- Centralize client dispatch reply handling
- Add Content-Type header to centralized reply publishing
- Simplify terminal_output by removing custom error types
- Preserve semantic JSON-RPC error codes
- Restore session ID cross-check in terminal_output
- Clean up session_update and ext_session_prompt_response

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

Author: yordis

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

@@ -142,7 +142,7 @@ mod tests {
 
         let payload = response_with_prompt_id(StopReason::EndTurn, token);
 
-        handle("prompt-resp-001", &payload, None, &bridge).await;
+        super::handle("prompt-resp-001", &payload, None, &bridge).await;
 
         let result = rx
             .await
@@ -156,7 +156,7 @@ mod tests {
         let bridge = make_bridge();
         let payload = response_with_prompt_id(StopReason::EndTurn, PromptToken(0));
 
-        handle("no-waiter-session", &payload, None, &bridge).await;
+        super::handle("no-waiter-session", &payload, None, &bridge).await;
     }
 
     #[tokio::test]
@@ -174,7 +174,7 @@ mod tests {
             token.0
         );
 
-        handle("bad-payload-001", payload.as_bytes(), None, &bridge).await;
+        super::handle("bad-payload-001", payload.as_bytes(), None, &bridge).await;
 
         let result = rx
             .await
@@ -196,7 +196,7 @@ mod tests {
         let response = PromptResponse::new(StopReason::EndTurn);
         let payload = serde_json::to_vec(&response).unwrap();
 
-        handle("no-token-session", &payload, None, &bridge).await;
+        super::handle("no-token-session", &payload, None, &bridge).await;
 
         assert!(
             bridge
@@ -222,9 +222,9 @@ mod tests {
 
         let payload = response_with_prompt_id(StopReason::EndTurn, token);
 
-        handle("session.with.dots", &payload, None, &bridge).await;
-        handle("session*wild", &payload, None, &bridge).await;
-        handle("session id", &payload, None, &bridge).await;
+        super::handle("session.with.dots", &payload, None, &bridge).await;
+        super::handle("session*wild", &payload, None, &bridge).await;
+        super::handle("session id", &payload, None, &bridge).await;
 
         assert!(
             bridge
@@ -267,7 +267,7 @@ mod tests {
             .unwrap();
 
         let late_payload = response_with_prompt_id(StopReason::EndTurn, token1);
-        handle("same-session", &late_payload, None, &bridge).await;
+        super::handle("same-session", &late_payload, None, &bridge).await;
 
         assert!(
             bridge

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

@@ -71,7 +71,6 @@ pub async fn run<
     let wildcard = client::wildcards::all(bridge.config.acp_prefix());
     info!("Starting client proxy - subscribing to {}", wildcard);
 
-    // TODO: change `run` to return `Result` and propagate this error once there is a caller.
     let mut subscriber = match nats.subscribe(wildcard).await {
         Ok(sub) => sub,
         Err(e) => {
@@ -221,7 +220,7 @@ async fn dispatch_client_method<
             .await;
         }
         ClientMethod::SessionUpdate => {
-            session_update::handle(&payload, ctx.client, &parsed.session_id).await;
+            session_update::handle(&payload, ctx.client, reply.is_some()).await;
         }
         ClientMethod::ExtSessionPromptResponse => {
             ext_session_prompt_response::handle(

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

@@ -1,10 +1,11 @@
-use crate::session_id::AcpSessionId;
 use agent_client_protocol::{Client, SessionNotification};
-use tracing::{info, instrument, warn};
+use tracing::{instrument, warn};
 
-#[instrument(name = "acp.client.session.update", skip(payload, client), fields(session_id = %session_id))]
-pub async fn handle<C: Client>(payload: &[u8], client: &C, session_id: &AcpSessionId) {
-    info!(session_id = %session_id, "Forwarding session update to client");
+#[instrument(name = "acp.client.session.update", skip(payload, client))]
+pub async fn handle<C: Client>(payload: &[u8], client: &C, has_reply: bool) {
+    if has_reply {
+        warn!("Unexpected reply subject on notification request");
+    }
     match serde_json::from_slice::<SessionNotification>(payload) {
         Ok(notification) => {
             if let Err(e) = client.session_notification(notification).await {
@@ -21,17 +22,12 @@ pub async fn handle<C: Client>(payload: &[u8], client: &C, session_id: &AcpSessi
 mod tests {
     use super::*;
     use agent_client_protocol::{
-        ContentBlock, ContentChunk, RequestPermissionRequest, RequestPermissionResponse,
-        SessionUpdate,
+        ContentBlock, ContentChunk, RequestPermissionOutcome, RequestPermissionRequest,
+        RequestPermissionResponse, SessionUpdate,
     };
     use async_trait::async_trait;
     use std::cell::RefCell;
 
-    fn session_id(s: &str) -> AcpSessionId {
-        AcpSessionId::new(s).unwrap()
-    }
-
-    #[derive(Debug)]
     struct MockClient {
         notifications_received: RefCell<Vec<String>>,
         should_fail: bool,
@@ -62,7 +58,7 @@ mod tests {
         async fn session_notification(
             &self,
             notification: SessionNotification,
-        ) -> Result<(), agent_client_protocol::Error> {
+        ) -> agent_client_protocol::Result<()> {
             if self.should_fail {
                 return Err(agent_client_protocol::Error::new(-1, "mock failure"));
             }
@@ -75,57 +71,43 @@ mod tests {
         async fn request_permission(
             &self,
             _: RequestPermissionRequest,
-        ) -> Result<RequestPermissionResponse, agent_client_protocol::Error> {
-            Err(agent_client_protocol::Error::new(
-                -32603,
-                "not implemented in test mock",
+        ) -> agent_client_protocol::Result<RequestPermissionResponse> {
+            Ok(RequestPermissionResponse::new(
+                RequestPermissionOutcome::Cancelled,
             ))
         }
     }
 
     #[tokio::test]
-    async fn session_update_forwards_notification_to_client() {
+    async fn forwards_notification_to_client() {
         let client = MockClient::new();
         let notification = SessionNotification::new(
             "session-001",
             SessionUpdate::AgentMessageChunk(ContentChunk::new(ContentBlock::from("hello"))),
         );
         let payload = serde_json::to_vec(&notification).unwrap();
 
-        handle(&payload, &client, &session_id("session-001")).await;
+        super::handle(&payload, &client, false).await;
 
         assert_eq!(client.notification_count(), 1);
     }
 
     #[tokio::test]
-    async fn session_update_invalid_payload_does_not_panic() {
+    async fn invalid_payload_does_not_panic() {
         let client = MockClient::new();
-        handle(b"not json", &client, &session_id("session-001")).await;
+        super::handle(b"not json", &client, false).await;
         assert_eq!(client.notification_count(), 0);
     }
 
     #[tokio::test]
-    async fn session_update_client_error_does_not_panic() {
+    async fn client_error_does_not_panic() {
         let client = MockClient::failing();
         let notification = SessionNotification::new(
             "session-001",
             SessionUpdate::AgentMessageChunk(ContentChunk::new(ContentBlock::from("hello"))),
         );
         let payload = serde_json::to_vec(&notification).unwrap();
 
-        handle(&payload, &client, &session_id("session-001")).await;
-    }
-
-    #[tokio::test]
-    async fn mock_client_request_permission_returns_err() {
-        let client = MockClient::new();
-        let req: RequestPermissionRequest = serde_json::from_value(serde_json::json!({
-            "sessionId": "sess-1",
-            "toolCall": { "toolCallId": "call-1" },
-            "options": []
-        }))
-        .unwrap();
-        let result = client.request_permission(req).await;
-        assert!(result.is_err());
+        super::handle(&payload, &client, false).await;
     }
 }