style(acp-nats): apply rustfmt formatting

Run cargo fmt to fix formatting in client/mod.rs (match arm indentation
for long type annotation) and client/terminal_output.rs (closure
formatting and line length).

Author: yordis

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

@@ -16,7 +16,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;
@@ -28,30 +28,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],
@@ -207,11 +183,9 @@ async fn dispatch_client_method<
     reply: Option<String>,
     ctx: &DispatchContext<'_, N, Cl, C, S>,
 ) {
-    let request_id = extract_request_id(&payload);
-
     Span::current().record("session_id", parsed.session_id.as_str());
 
-    let response_bytes: Option<Result<Vec<u8>, Box<dyn std::error::Error + Send + Sync>>> = match parsed.method {
+    match parsed.method {
         ClientMethod::FsReadTextFile => {
             fs_read_text_file::handle(
                 &payload,
@@ -222,7 +196,6 @@ async fn dispatch_client_method<
                 ctx.serializer,
             )
             .await;
-            None
         }
         ClientMethod::FsWriteTextFile => {
             fs_write_text_file::handle(
@@ -234,7 +207,6 @@ async fn dispatch_client_method<
                 ctx.serializer,
             )
             .await;
-            None
         }
         ClientMethod::SessionRequestPermission => {
             request_permission::handle(
@@ -246,18 +218,9 @@ async fn dispatch_client_method<
                 ctx.serializer,
             )
             .await;
-            None
         }
         ClientMethod::SessionUpdate => {
-            if reply.is_some() {
-                warn!(
-                    session_id = %parsed.session_id,
-                    method = ?parsed.method,
-                    "Unexpected reply subject on notification request"
-                );
-            }
-            session_update::handle(&payload, ctx.client).await;
-            None
+            session_update::handle(&payload, ctx.client, reply.is_some()).await;
         }
         ClientMethod::ExtSessionPromptResponse => {
             ext_session_prompt_response::handle(
@@ -267,7 +230,6 @@ async fn dispatch_client_method<
                 ctx.bridge,
             )
             .await;
-            None
         }
         ClientMethod::TerminalCreate => {
             terminal_create::handle(
@@ -279,7 +241,6 @@ async fn dispatch_client_method<
                 ctx.serializer,
             )
             .await;
-            None
         }
         ClientMethod::TerminalKill => {
             terminal_kill::handle(
@@ -291,12 +252,17 @@ async fn dispatch_client_method<
                 ctx.serializer,
             )
             .await;
-            None
         }
         ClientMethod::TerminalOutput => {
-            Some(
-                terminal_output::handle(&payload, ctx.client, parsed.session_id.as_str()).await,
+            terminal_output::handle(
+                &payload,
+                ctx.client,
+                reply.as_deref(),
+                ctx.nats,
+                parsed.session_id.as_str(),
+                ctx.serializer,
             )
+            .await;
         }
         ClientMethod::TerminalRelease => {
             terminal_release::handle(
@@ -308,7 +274,6 @@ async fn dispatch_client_method<
                 ctx.serializer,
             )
             .await;
-            None
         }
         ClientMethod::TerminalWaitForExit => {
             terminal_wait_for_exit::handle(
@@ -321,51 +286,6 @@ async fn dispatch_client_method<
                 ctx.serializer,
             )
             .await;
-            None
-        }
-    };
-
-    if let (Some(reply_to), Some(result)) = (reply, response_bytes) {
-        match result {
-            Ok(bytes) => {
-                let mut headers = headers_with_trace_context();
-                headers.insert("Content-Type", rpc_reply::CONTENT_TYPE_JSON);
-                if let Err(e) = ctx
-                    .nats
-                    .publish_with_headers(reply_to, 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");
-                }
-            }
-            Err(e) => {
-                error!(
-                    error = %e,
-                    method = ?parsed.method,
-                    session_id = %parsed.session_id,
-                    "Failed to handle client request"
-                );
-                let bytes = jsonrpc_error_response(
-                    request_id,
-                    ErrorCode::InternalError,
-                    "Internal error while handling client request",
-                );
-                let mut headers = headers_with_trace_context();
-                headers.insert("Content-Type", rpc_reply::CONTENT_TYPE_JSON);
-                if let Err(e) = ctx
-                    .nats
-                    .publish_with_headers(reply_to, 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");
-                }
-            }
         }
     }
 }

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

@@ -2,7 +2,10 @@ use agent_client_protocol::{Client, SessionNotification};
 use tracing::{instrument, warn};
 
 #[instrument(name = "acp.client.session.update", skip(payload, client))]
-pub async fn handle<C: Client>(payload: &[u8], client: &C) {
+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 {
@@ -82,15 +85,15 @@ mod tests {
         );
         let payload = serde_json::to_vec(&notification).unwrap();
 
-        super::handle(&payload, &client).await;
+        super::handle(&payload, &client, false).await;
 
         assert_eq!(client.notification_count(), 1);
     }
 
     #[tokio::test]
     async fn invalid_payload_does_not_panic() {
         let client = MockClient::new();
-        super::handle(b"not json", &client).await;
+        super::handle(b"not json", &client, false).await;
         assert_eq!(client.notification_count(), 0);
     }
 
@@ -103,6 +106,6 @@ mod tests {
         );
         let payload = serde_json::to_vec(&notification).unwrap();
 
-        super::handle(&payload, &client).await;
+        super::handle(&payload, &client, false).await;
     }
 }