cache: add grpc read-only response caching

crates/crank-adapter-grpc/src/client.rs

@@ -229,6 +229,7 @@ mod tests {
             package: "echo".to_owned(),
             service: "EchoService".to_owned(),
             method: "UnaryEcho".to_owned(),
+            read_only: false,
             descriptor_ref: DescriptorId::new("desc_echo"),
             descriptor_set_b64: test_support::descriptor_set_b64(),
         };
@@ -252,6 +253,7 @@ mod tests {
             package: "echo".to_owned(),
             service: "EchoService".to_owned(),
             method: "ServerEcho".to_owned(),
+            read_only: false,
             descriptor_ref: DescriptorId::new("desc_echo"),
             descriptor_set_b64: test_support::descriptor_set_b64(),
         };
@@ -288,6 +290,7 @@ mod tests {
             package: "echo".to_owned(),
             service: "EchoService".to_owned(),
             method: "UnaryEcho".to_owned(),
+            read_only: false,
             descriptor_ref: DescriptorId::new("desc_echo"),
             descriptor_set_b64: test_support::descriptor_set_b64(),
         };

crates/crank-adapter-grpc/src/test_support.rs

@@ -1,5 +1,9 @@
 use base64::{Engine as _, engine::general_purpose::STANDARD};
 use futures_util::stream;
+use std::sync::{
+    Arc,
+    atomic::{AtomicUsize, Ordering},
+};
 use tokio::net::TcpListener;
 use tonic::{Request, Response, Status, transport::Server};
 
@@ -121,6 +125,54 @@ pub async fn spawn_metadata_echo_server() -> String {
     format!("http://{}", address)
 }
 
+pub async fn spawn_counting_unary_echo_server(request_count: Arc<AtomicUsize>) -> String {
+    struct CountingEchoServiceImpl {
+        request_count: Arc<AtomicUsize>,
+    }
+
+    #[tonic::async_trait]
+    impl echo::echo_service_server::EchoService for CountingEchoServiceImpl {
+        async fn unary_echo(
+            &self,
+            request: Request<echo::EchoRequest>,
+        ) -> Result<Response<echo::EchoResponse>, Status> {
+            let message = request.into_inner().message;
+            let count = self.request_count.fetch_add(1, Ordering::SeqCst) + 1;
+
+            Ok(Response::new(echo::EchoResponse {
+                message: format!("{message}-{count}"),
+            }))
+        }
+
+        type ServerEchoStream = std::pin::Pin<
+            Box<dyn futures_util::Stream<Item = Result<echo::EchoResponse, Status>> + Send>,
+        >;
+
+        async fn server_echo(
+            &self,
+            _request: Request<echo::EchoRequest>,
+        ) -> Result<Response<Self::ServerEchoStream>, Status> {
+            Ok(Response::new(Box::pin(stream::empty())))
+        }
+    }
+
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let address = listener.local_addr().unwrap();
+    let incoming = tonic::transport::server::TcpIncoming::from(listener);
+
+    tokio::spawn(async move {
+        Server::builder()
+            .add_service(echo::echo_service_server::EchoServiceServer::new(
+                CountingEchoServiceImpl { request_count },
+            ))
+            .serve_with_incoming(incoming)
+            .await
+            .unwrap();
+    });
+
+    format!("http://{}", address)
+}
+
 pub fn descriptor_set_b64() -> String {
     STANDARD.encode(echo::FILE_DESCRIPTOR_SET)
 }

crates/crank-core/src/operation.rs

@@ -55,6 +55,8 @@ pub struct GrpcTarget {
     pub package: String,
     pub service: String,
     pub method: String,
+    #[serde(default)]
+    pub read_only: bool,
     pub descriptor_ref: DescriptorId,
     pub descriptor_set_b64: String,
 }

crates/crank-runtime/src/executor.rs

@@ -757,9 +757,14 @@ fn log_runtime_event(
 }
 
 fn response_cache_ttl(operation: &RuntimeOperation) -> Option<Duration> {
+    if operation.execution_config.streaming.is_some() {
+        return None;
+    }
+
     let is_cacheable_protocol = match &operation.target {
         Target::Rest(target) => target.method == HttpMethod::Get,
         Target::Graphql(target) => target.operation_type == crank_core::GraphqlOperationType::Query,
+        Target::Grpc(target) => target.read_only,
         _ => false,
     };
     if !is_cacheable_protocol {
@@ -1130,6 +1135,79 @@ mod tests {
         assert_eq!(request_count.load(Ordering::SeqCst), 1);
     }
 
+    #[tokio::test]
+    async fn caches_read_only_grpc_unary_responses_within_agent_scope() {
+        let request_count = Arc::new(AtomicUsize::new(0));
+        let server_addr =
+            grpc_test_support::spawn_counting_unary_echo_server(Arc::clone(&request_count)).await;
+        let executor = RuntimeExecutor::new()
+            .with_response_cache_store(Arc::new(InMemoryResponseCacheStore::default()));
+        let operation = test_cached_grpc_operation(&server_addr);
+        let first_context = RuntimeRequestContext::from_request_id("req_cache_grpc_1")
+            .with_response_cache_scope("ws_cache", "agent_sales");
+        let second_context = RuntimeRequestContext::from_request_id("req_cache_grpc_2")
+            .with_response_cache_scope("ws_cache", "agent_sales");
+
+        let first_output = executor
+            .execute_with_context(
+                &operation,
+                &json!({ "message": "hello" }),
+                Some(&first_context),
+            )
+            .await
+            .unwrap();
+        let second_output = executor
+            .execute_with_context(
+                &operation,
+                &json!({ "message": "hello" }),
+                Some(&second_context),
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(first_output, json!({ "message": "hello-1" }));
+        assert_eq!(second_output, json!({ "message": "hello-1" }));
+        assert_eq!(request_count.load(Ordering::SeqCst), 1);
+    }
+
+    #[tokio::test]
+    async fn skips_grpc_response_cache_without_read_only_flag() {
+        let request_count = Arc::new(AtomicUsize::new(0));
+        let server_addr =
+            grpc_test_support::spawn_counting_unary_echo_server(Arc::clone(&request_count)).await;
+        let executor = RuntimeExecutor::new()
+            .with_response_cache_store(Arc::new(InMemoryResponseCacheStore::default()));
+        let mut operation = test_cached_grpc_operation(&server_addr);
+        if let Target::Grpc(target) = &mut operation.target {
+            target.read_only = false;
+        }
+        let first_context = RuntimeRequestContext::from_request_id("req_cache_grpc_skip_1")
+            .with_response_cache_scope("ws_cache", "agent_sales");
+        let second_context = RuntimeRequestContext::from_request_id("req_cache_grpc_skip_2")
+            .with_response_cache_scope("ws_cache", "agent_sales");
+
+        let first_output = executor
+            .execute_with_context(
+                &operation,
+                &json!({ "message": "hello" }),
+                Some(&first_context),
+            )
+            .await
+            .unwrap();
+        let second_output = executor
+            .execute_with_context(
+                &operation,
+                &json!({ "message": "hello" }),
+                Some(&second_context),
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(first_output, json!({ "message": "hello-1" }));
+        assert_eq!(second_output, json!({ "message": "hello-2" }));
+        assert_eq!(request_count.load(Ordering::SeqCst), 2);
+    }
+
     #[tokio::test]
     async fn emits_runtime_tracing_with_request_context() {
         let base_url = spawn_runtime_server().await;
@@ -2244,6 +2322,26 @@ mod tests {
         })
     }
 
+    fn test_cached_grpc_operation(server_addr: &str) -> RuntimeOperation {
+        let mut operation = test_grpc_operation(server_addr);
+        operation.operation_id = OperationId::new("op_grpc_cached_runtime");
+        operation.tool_name = "echo_unary_grpc_cached".to_owned();
+        if let Target::Grpc(target) = &mut operation.target {
+            target.read_only = true;
+        }
+        operation.execution_config.response_cache =
+            Some(crank_core::ResponseCachePolicy { ttl_ms: 5_000 });
+        operation.tool_description = ToolDescription {
+            title: "Unary Echo gRPC Cached".to_owned(),
+            description: "Reads a cacheable unary gRPC payload".to_owned(),
+            tags: vec!["grpc".to_owned(), "cache".to_owned()],
+            examples: vec![ToolExample {
+                input: json!({ "message": "hello" }),
+            }],
+        };
+        operation
+    }
+
     fn test_grpc_operation(server_addr: &str) -> RuntimeOperation {
         RuntimeOperation::from(Operation {
             id: OperationId::new("op_grpc_runtime"),
@@ -2259,6 +2357,7 @@ mod tests {
                 package: "echo".to_owned(),
                 service: "EchoService".to_owned(),
                 method: "UnaryEcho".to_owned(),
+                read_only: false,
                 descriptor_ref: DescriptorId::new("desc_echo"),
                 descriptor_set_b64: grpc_test_support::descriptor_set_b64(),
             }),
@@ -2340,6 +2439,7 @@ mod tests {
                 package: "echo".to_owned(),
                 service: "EchoService".to_owned(),
                 method: "ServerEcho".to_owned(),
+                read_only: false,
                 descriptor_ref: DescriptorId::new("desc_echo"),
                 descriptor_set_b64: grpc_test_support::descriptor_set_b64(),
             }),