Merge pull request #27 from agntcy/feat/push-config-and-grpc-list-interop

feat(server): support push configs and gRPC list interop

Author: muscariello

a2a-grpc/tests/e2e.rs

@@ -8,7 +8,7 @@ use a2a::*;
 use a2a_client::{Transport, TransportFactory};
 use a2a_grpc::{GrpcHandler, GrpcTransport, GrpcTransportFactory};
 use a2a_pb::proto::a2a_service_server::A2aServiceServer;
-use a2a_server::{RequestHandler, ServiceParams};
+use a2a_server::{DefaultRequestHandler, InMemoryTaskStore, RequestHandler, ServiceParams};
 use async_trait::async_trait;
 use futures::StreamExt;
 use futures::stream::{self, BoxStream};
@@ -81,6 +81,8 @@ fn sample_agent_card() -> AgentCard {
 
 struct TestHandler;
 
+struct StoredTaskExecutor;
+
 #[async_trait]
 impl RequestHandler for TestHandler {
     async fn send_message(
@@ -245,6 +247,57 @@ impl RequestHandler for TestHandler {
     }
 }
 
+impl a2a_server::AgentExecutor for StoredTaskExecutor {
+    fn execute(
+        &self,
+        ctx: a2a_server::ExecutorContext,
+    ) -> BoxStream<'static, Result<StreamResponse, A2AError>> {
+        let response = StreamResponse::Task(Task {
+            id: ctx.task_id.clone(),
+            context_id: ctx.context_id.clone(),
+            status: TaskStatus {
+                state: TaskState::Completed,
+                message: Some(Message {
+                    message_id: "stored-task-response".to_string(),
+                    context_id: Some(ctx.context_id.clone()),
+                    task_id: Some(ctx.task_id.clone()),
+                    role: Role::Agent,
+                    parts: vec![Part::text("stored-task-done")],
+                    metadata: None,
+                    extensions: None,
+                    reference_task_ids: None,
+                }),
+                timestamp: None,
+            },
+            artifacts: None,
+            history: ctx.message.clone().map(|message| vec![message]),
+            metadata: None,
+        });
+
+        Box::pin(stream::once(async move { Ok(response) }))
+    }
+
+    fn cancel(
+        &self,
+        ctx: a2a_server::ExecutorContext,
+    ) -> BoxStream<'static, Result<StreamResponse, A2AError>> {
+        let response = StreamResponse::Task(Task {
+            id: ctx.task_id.clone(),
+            context_id: ctx.context_id.clone(),
+            status: TaskStatus {
+                state: TaskState::Canceled,
+                message: None,
+                timestamp: None,
+            },
+            artifacts: None,
+            history: None,
+            metadata: None,
+        });
+
+        Box::pin(stream::once(async move { Ok(response) }))
+    }
+}
+
 async fn spawn_grpc_server() -> (String, tokio::task::JoinHandle<()>) {
     let handler = Arc::new(TestHandler);
     let service = A2aServiceServer::new(GrpcHandler::new(handler));
@@ -262,6 +315,26 @@ async fn spawn_grpc_server() -> (String, tokio::task::JoinHandle<()>) {
     (format!("http://{addr}"), handle)
 }
 
+async fn spawn_default_handler_grpc_server() -> (String, tokio::task::JoinHandle<()>) {
+    let handler = Arc::new(DefaultRequestHandler::new(
+        StoredTaskExecutor,
+        InMemoryTaskStore::new(),
+    ));
+    let service = A2aServiceServer::new(GrpcHandler::new(handler));
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let addr = listener.local_addr().unwrap();
+    let incoming = TcpListenerStream::new(listener);
+    let handle = tokio::spawn(async move {
+        Server::builder()
+            .add_service(service)
+            .serve_with_incoming(incoming)
+            .await
+            .unwrap();
+    });
+    tokio::time::sleep(Duration::from_millis(20)).await;
+    (format!("http://{addr}"), handle)
+}
+
 fn endpoint_without_scheme(endpoint: &str) -> String {
     endpoint
         .split_once("://")
@@ -502,3 +575,51 @@ async fn grpc_transport_accepts_bare_host_port_endpoints() {
     transport.destroy().await.unwrap();
     handle.abort();
 }
+
+#[tokio::test]
+async fn grpc_transport_treats_zero_page_size_as_unset() {
+    let (endpoint, handle) = spawn_default_handler_grpc_server().await;
+    let transport = GrpcTransport::connect(endpoint).await.unwrap();
+
+    let sent = transport
+        .send_message(
+            &ServiceParams::new(),
+            &SendMessageRequest {
+                message: Message::new(Role::User, vec![Part::text("hello")]),
+                configuration: None,
+                metadata: None,
+                tenant: None,
+            },
+        )
+        .await
+        .unwrap();
+
+    let task = match sent {
+        SendMessageResponse::Task(task) => task,
+        SendMessageResponse::Message(_) => panic!("expected task response"),
+    };
+
+    let listed = transport
+        .list_tasks(
+            &ServiceParams::new(),
+            &ListTasksRequest {
+                context_id: Some(task.context_id.clone()),
+                status: None,
+                page_size: Some(0),
+                page_token: None,
+                history_length: None,
+                status_timestamp_after: None,
+                include_artifacts: Some(false),
+                tenant: None,
+            },
+        )
+        .await
+        .unwrap();
+
+    assert_eq!(listed.tasks.len(), 1);
+    assert_eq!(listed.tasks[0].id, task.id);
+    assert_eq!(listed.page_size, 50);
+
+    transport.destroy().await.unwrap();
+    handle.abort();
+}

a2a-pb/src/pbconv.rs

@@ -111,6 +111,10 @@ fn empty_to_none(s: &str) -> Option<String> {
     }
 }
 
+fn non_positive_to_none(value: Option<i32>) -> Option<i32> {
+    value.filter(|value| *value > 0)
+}
+
 // ---------------------------------------------------------------------------
 // Role
 // ---------------------------------------------------------------------------
@@ -519,7 +523,7 @@ pub fn from_proto_list_tasks_request(r: &proto::ListTasksRequest) -> ListTasksRe
         } else {
             Some(from_proto_task_state(r.status))
         },
-        page_size: r.page_size,
+        page_size: non_positive_to_none(r.page_size),
         page_token: empty_to_none(&r.page_token),
         history_length: r.history_length,
         status_timestamp_after: r
@@ -1626,6 +1630,24 @@ mod tests {
         assert_eq!(req.tenant, back.tenant);
     }
 
+    #[test]
+    fn test_from_proto_list_tasks_request_treats_zero_page_size_as_unset() {
+        let proto = proto::ListTasksRequest {
+            tenant: String::new(),
+            context_id: "ctx-1".to_string(),
+            status: 0,
+            page_size: Some(0),
+            page_token: String::new(),
+            history_length: None,
+            status_timestamp_after: None,
+            include_artifacts: Some(false),
+        };
+
+        let back = from_proto_list_tasks_request(&proto);
+        assert_eq!(back.context_id.as_deref(), Some("ctx-1"));
+        assert_eq!(back.page_size, None);
+    }
+
     #[test]
     fn test_list_tasks_response_roundtrip() {
         let resp = ListTasksResponse {