feat(acp-nats): add fs_read_text_file client handler

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

Author: yordis

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

@@ -0,0 +1,174 @@
+use crate::jsonrpc::JsonRpcRequest;
+use agent_client_protocol::{Client, ErrorCode, ReadTextFileRequest};
+use tracing::instrument;
+
+#[derive(Debug)]
+pub enum FsReadTextFileError {
+    InvalidRequest(serde_json::Error),
+    ClientError(agent_client_protocol::Error),
+    SerializationError(serde_json::Error),
+}
+
+impl std::fmt::Display for FsReadTextFileError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::InvalidRequest(e) => write!(f, "invalid request: {}", e),
+            Self::ClientError(e) => write!(f, "client error: {}", e),
+            Self::SerializationError(e) => write!(f, "serialization error: {}", e),
+        }
+    }
+}
+
+impl std::error::Error for FsReadTextFileError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::InvalidRequest(e) => Some(e),
+            Self::ClientError(e) => Some(e),
+            Self::SerializationError(e) => Some(e),
+        }
+    }
+}
+
+pub fn error_code_and_message(e: &FsReadTextFileError) -> (ErrorCode, String) {
+    match e {
+        FsReadTextFileError::InvalidRequest(inner) => (
+            ErrorCode::InvalidParams,
+            format!("Invalid read_text_file request: {}", inner),
+        ),
+        FsReadTextFileError::ClientError(inner) => (inner.code, inner.message.clone()),
+        FsReadTextFileError::SerializationError(inner) => (
+            ErrorCode::InternalError,
+            format!("Failed to serialize read_text_file response: {}", inner),
+        ),
+    }
+}
+
+/// Forwards read_text_file to the client. NATS enforces payload limits when publishing.
+/// Expects a JSON-RPC envelope with a "params" object containing the ReadTextFileRequest fields.
+#[instrument(name = "acp.client.fs.read_text_file", skip(payload, client))]
+pub async fn handle<C: Client>(payload: &[u8], client: &C) -> Result<Vec<u8>, FsReadTextFileError> {
+    let envelope: JsonRpcRequest<ReadTextFileRequest> =
+        serde_json::from_slice(payload).map_err(FsReadTextFileError::InvalidRequest)?;
+    let request = envelope.params.ok_or_else(|| {
+        FsReadTextFileError::InvalidRequest(
+            serde_json::from_value::<ReadTextFileRequest>(serde_json::Value::Null).unwrap_err(),
+        )
+    })?;
+    let response = client
+        .read_text_file(request)
+        .await
+        .map_err(FsReadTextFileError::ClientError)?;
+    serde_json::to_vec(&response).map_err(FsReadTextFileError::SerializationError)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::jsonrpc::JsonRpcRequest;
+    use agent_client_protocol::RequestPermissionRequest;
+    use agent_client_protocol::RequestPermissionResponse;
+    use agent_client_protocol::SessionNotification;
+    use agent_client_protocol::{ReadTextFileRequest, ReadTextFileResponse};
+    use async_trait::async_trait;
+
+    struct MockClient {
+        content: String,
+    }
+
+    impl MockClient {
+        fn new(content: &str) -> Self {
+            Self {
+                content: content.to_string(),
+            }
+        }
+    }
+
+    #[async_trait(?Send)]
+    impl Client for MockClient {
+        async fn session_notification(
+            &self,
+            _: SessionNotification,
+        ) -> agent_client_protocol::Result<()> {
+            Ok(())
+        }
+
+        async fn request_permission(
+            &self,
+            _: RequestPermissionRequest,
+        ) -> agent_client_protocol::Result<RequestPermissionResponse> {
+            Err(agent_client_protocol::Error::new(
+                -32603,
+                "not implemented in test mock",
+            ))
+        }
+
+        async fn read_text_file(
+            &self,
+            _: ReadTextFileRequest,
+        ) -> agent_client_protocol::Result<ReadTextFileResponse> {
+            Ok(ReadTextFileResponse::new(self.content.clone()))
+        }
+    }
+
+    #[tokio::test]
+    async fn fs_read_text_file_forwards_request_and_returns_response() {
+        let client = MockClient::new("hello world");
+        let request = ReadTextFileRequest::new(
+            agent_client_protocol::SessionId::from("sess-1"),
+            "/tmp/foo.txt".to_string(),
+        );
+        let envelope = JsonRpcRequest {
+            jsonrpc: "2.0".to_string(),
+            id: serde_json::json!(1),
+            method: Some("fs/read_text_file".to_string()),
+            params: Some(request),
+        };
+        let payload = serde_json::to_vec(&envelope).unwrap();
+
+        let result = handle(&payload, &client).await;
+        assert!(result.is_ok());
+        let response = serde_json::from_slice::<ReadTextFileResponse>(&result.unwrap()).unwrap();
+        assert_eq!(response.content, "hello world");
+    }
+
+    #[tokio::test]
+    async fn fs_read_text_file_returns_error_when_payload_is_invalid_json() {
+        let client = MockClient::new("hello");
+        let result = handle(b"not json", &client).await;
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn error_code_and_message_invalid_request_returns_invalid_params() {
+        let err = serde_json::from_slice::<ReadTextFileRequest>(b"not json").unwrap_err();
+        let fs_err = FsReadTextFileError::InvalidRequest(err);
+        let (code, message) = error_code_and_message(&fs_err);
+        assert_eq!(code, ErrorCode::InvalidParams);
+        assert!(message.contains("Invalid read_text_file request"));
+    }
+
+    #[test]
+    fn error_code_and_message_client_error_preserves_client_code() {
+        let client_err =
+            agent_client_protocol::Error::new(ErrorCode::InvalidParams.into(), "file not found");
+        let fs_err = FsReadTextFileError::ClientError(client_err);
+        let (code, message) = error_code_and_message(&fs_err);
+        assert_eq!(code, ErrorCode::InvalidParams);
+        assert_eq!(message, "file not found");
+    }
+
+    #[test]
+    fn error_code_and_message_serialization_error_returns_internal_error() {
+        struct Unserializable;
+        impl serde::Serialize for Unserializable {
+            fn serialize<S: serde::Serializer>(&self, _: S) -> Result<S::Ok, S::Error> {
+                Err(serde::ser::Error::custom("test serialization failure"))
+            }
+        }
+        let err = serde_json::to_vec(&Unserializable).unwrap_err();
+        let fs_err = FsReadTextFileError::SerializationError(err);
+        let (code, message) = error_code_and_message(&fs_err);
+        assert_eq!(code, ErrorCode::InternalError);
+        assert!(message.contains("Failed to serialize read_text_file response"));
+    }
+}

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

@@ -1,12 +1,16 @@
+pub(crate) mod fs_read_text_file;
 pub(crate) mod session_update;
 
 use crate::agent::Bridge;
+use crate::error::AGENT_UNAVAILABLE;
 use crate::in_flight_slot_guard::InFlightSlotGuard;
+use crate::jsonrpc::{ErrorResponse, ResultResponse, extract_request_id};
 use crate::nats::{
     ClientMethod, FlushClient, PublishClient, RequestClient, SubscribeClient, client,
-    parse_client_subject,
+    headers_with_trace_context, parse_client_subject,
 };
 use agent_client_protocol::Client;
+use agent_client_protocol::ErrorCode;
 use async_nats::Message;
 use bytes::Bytes;
 use futures::StreamExt;
@@ -84,6 +88,9 @@ async fn process_message<
         }
     };
 
+    let payload = msg.payload.clone();
+    let reply = msg.reply.as_ref().map(|r| r.to_string());
+
     let current_in_flight = in_flight.get();
     if current_in_flight >= max_concurrent {
         warn!(
@@ -96,10 +103,39 @@ async fn process_message<
             .metrics
             .record_error("client", "client_backpressure_rejected");
 
+        if let Some(reply_to) = &reply {
+            let request_id = extract_request_id(&payload);
+            let bytes = ErrorResponse::new(
+                request_id,
+                ErrorCode::Other(AGENT_UNAVAILABLE),
+                "Client proxy overloaded; retry with backoff",
+            )
+            .to_bytes()
+            .unwrap_or_else(|e| {
+                ErrorResponse::new(
+                    serde_json::Value::Null,
+                    ErrorCode::Other(AGENT_UNAVAILABLE),
+                    format!(
+                        "Client proxy overloaded; retry with backoff (serialization failed: {})",
+                        e
+                    ),
+                )
+                .to_bytes()
+                .unwrap()
+            });
+            let headers = headers_with_trace_context();
+            if let Err(e) = nats
+                .publish_with_headers(reply_to.clone(), headers, bytes)
+                .await
+            {
+                error!(error = %e, "Failed to publish backpressure error reply");
+            }
+            if let Err(e) = nats.flush().await {
+                warn!(error = %e, "Failed to flush backpressure error reply");
+            }
+        }
         return;
     }
-
-    let payload = msg.payload.clone();
     let nats = nats.clone();
 
     let bridge_clone = bridge.clone();
@@ -110,6 +146,7 @@ async fn process_message<
             &subject,
             parsed,
             payload,
+            reply,
             &nats,
             client.as_ref(),
             bridge_clone.as_ref(),
@@ -118,7 +155,7 @@ async fn process_message<
     });
 }
 
-#[instrument(skip(payload, _nats, client, _bridge), fields(subject = %subject, session_id = tracing::field::Empty))]
+#[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,
@@ -127,13 +164,68 @@ async fn dispatch_client_method<
     subject: &str,
     parsed: crate::nats::ParsedClientSubject,
     payload: Bytes,
-    _nats: &N,
+    reply: Option<String>,
+    nats: &N,
     client: &Cl,
     _bridge: &Bridge<N, C>,
 ) {
     Span::current().record("session_id", parsed.session_id.as_str());
 
     match parsed.method {
+        ClientMethod::FsReadTextFile => {
+            let request_id = extract_request_id(&payload);
+            match fs_read_text_file::handle(&payload, client).await {
+                Ok(bytes) => {
+                    if let Some(reply_to) = &reply {
+                        let result =
+                            serde_json::from_slice(&bytes).unwrap_or(serde_json::Value::Null);
+                        let response_bytes =
+                            ResultResponse::new(request_id, result).to_bytes().unwrap();
+                        let headers = headers_with_trace_context();
+                        if let Err(e) = nats
+                            .publish_with_headers(reply_to.clone(), headers, response_bytes)
+                            .await
+                        {
+                            error!(error = %e, "Failed to publish fs_read_text_file reply");
+                        }
+                        if let Err(e) = nats.flush().await {
+                            warn!(error = %e, "Failed to flush fs_read_text_file reply");
+                        }
+                    }
+                }
+                Err(e) => {
+                    let (code, message) = fs_read_text_file::error_code_and_message(&e);
+                    warn!(
+                        error = %e,
+                        session_id = %parsed.session_id,
+                        "Failed to handle fs_read_text_file"
+                    );
+                    if let Some(reply_to) = &reply {
+                        let bytes = ErrorResponse::new(request_id, code, message.as_str())
+                            .to_bytes()
+                            .unwrap_or_else(|e| {
+                                ErrorResponse::new(
+                                    serde_json::Value::Null,
+                                    code,
+                                    format!("{} (serialization failed: {})", message, e),
+                                )
+                                .to_bytes()
+                                .unwrap()
+                            });
+                        let headers = headers_with_trace_context();
+                        if let Err(e) = nats
+                            .publish_with_headers(reply_to.clone(), headers, bytes)
+                            .await
+                        {
+                            error!(error = %e, "Failed to publish fs_read_text_file error reply");
+                        }
+                        if let Err(e) = nats.flush().await {
+                            warn!(error = %e, "Failed to flush fs_read_text_file error reply");
+                        }
+                    }
+                }
+            }
+        }
         ClientMethod::SessionUpdate => {
             session_update::handle(&payload, client, &parsed.session_id).await;
         }
@@ -143,10 +235,11 @@ async fn dispatch_client_method<
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::jsonrpc::JsonRpcRequest;
     use crate::session_id::AcpSessionId;
     use agent_client_protocol::{
-        ContentBlock, ContentChunk, RequestPermissionRequest, RequestPermissionResponse,
-        SessionNotification, SessionUpdate,
+        ContentBlock, ContentChunk, ReadTextFileRequest, RequestPermissionRequest,
+        RequestPermissionResponse, SessionNotification, SessionUpdate,
     };
     use async_trait::async_trait;
     use std::cell::RefCell;
@@ -286,6 +379,7 @@ mod tests {
             "acp.sess-1.client.session.update",
             parsed,
             payload,
+            None,
             &nats,
             &client,
             &bridge,
@@ -334,6 +428,33 @@ 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 envelope = JsonRpcRequest {
+            jsonrpc: "2.0".to_string(),
+            id: serde_json::json!(1),
+            method: Some("fs/read_text_file".to_string()),
+            params: Some(ReadTextFileRequest::new(
+                agent_client_protocol::SessionId::from("sess1"),
+                "/tmp/foo.txt".to_string(),
+            )),
+        };
+        let payload = serde_json::to_vec(&envelope).unwrap();
+        let msg = make_msg(
+            "acp.sess1.client.fs.read_text_file",
+            &payload,
+            Some("_INBOX.reply"),
+        );
+        process_message(msg, &nats, client, bridge, &in_flight, 1).await;
+
+        assert_eq!(nats.published_messages(), vec!["_INBOX.reply"]);
+    }
+
     #[tokio::test]
     async fn process_message_valid_dispatch_spawns_task() {
         let local = tokio::task::LocalSet::new();