feat(a2a-nats): add server push_list handler

rsworkspace/crates/a2a-nats/src/server/handler.rs

@@ -124,6 +124,11 @@ pub trait A2aExecutor: Send + Sync + 'static {
         &self,
         request: a2a::types::GetTaskPushNotificationConfigRequest,
     ) -> Result<a2a::types::TaskPushNotificationConfig, A2aError>;
+
+    async fn push_notification_list(
+        &self,
+        request: a2a::types::ListTaskPushNotificationConfigsRequest,
+    ) -> Result<a2a::types::ListTaskPushNotificationConfigsResponse, A2aError>;
 }
 
 #[cfg(test)]

rsworkspace/crates/a2a-nats/src/server/mod.rs

@@ -7,6 +7,7 @@ pub mod agent_card;
 pub mod handler;
 pub mod message_send;
 pub mod push_get;
+pub mod push_list;
 pub mod push_set;
 pub mod tasks_cancel;
 pub mod tasks_get;

rsworkspace/crates/a2a-nats/src/server/push_list.rs

@@ -0,0 +1,172 @@
+use tracing::{instrument, warn};
+
+use crate::jsonrpc::extract_request_id;
+use crate::server::handler::{A2aError, A2aExecutor};
+use crate::server::wire::{JsonRpcErrorResponse, JsonRpcResponse, is_notification, parse_request};
+
+#[instrument(name = "a2a.server.push_list", skip(handler, payload, reply_subject, nats))]
+pub async fn handle<H, N>(handler: &H, payload: &[u8], reply_subject: Option<String>, nats: &N)
+where
+    H: A2aExecutor,
+    N: trogon_nats::PublishClient,
+{
+    let Some(reply) = reply_subject else {
+        warn!("tasks/pushNotificationConfig/list received without reply subject; dropping");
+        return;
+    };
+
+    let id = extract_request_id(payload);
+    if id.is_none() && is_notification(payload) {
+        return;
+    }
+
+    let result = match parse_request::<serde_json::Value>(payload) {
+        Err(_) => Err(A2aError::new(-32700, "Parse error")),
+        Ok(envelope) => match envelope.params {
+            None => Err(A2aError::new(-32602, "Invalid params: missing params")),
+            Some(raw) => match serde_json::from_value::<a2a::types::ListTaskPushNotificationConfigsRequest>(raw) {
+                Err(e) => Err(A2aError::new(-32602, format!("Invalid params: {e}"))),
+                Ok(params) => handler.push_notification_list(params).await,
+            },
+        },
+    };
+    let bytes = match result {
+        Ok(resp) => JsonRpcResponse::new(id, resp).to_bytes(),
+        Err(e) => JsonRpcErrorResponse::new(id, e.code, e.message).to_bytes(),
+    };
+    match bytes {
+        Ok(b) => {
+            let headers = async_nats::HeaderMap::new();
+            if let Err(e) = nats
+                .publish_with_headers(async_nats::Subject::from(reply.as_str()), headers, b)
+                .await
+            {
+                warn!(error = %e, "failed to publish push_list reply");
+            }
+        }
+        Err(e) => warn!(error = %e, "failed to serialize push_list response"),
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use trogon_nats::AdvancedMockNatsClient;
+
+    use super::*;
+    use crate::server::test_support::{parse_response, stub};
+
+    fn list_request() -> a2a::types::ListTaskPushNotificationConfigsRequest {
+        a2a::types::ListTaskPushNotificationConfigsRequest {
+            task_id: "task-1".to_string(),
+            page_size: None,
+            page_token: None,
+            tenant: None,
+        }
+    }
+
+    fn empty_response() -> a2a::types::ListTaskPushNotificationConfigsResponse {
+        a2a::types::ListTaskPushNotificationConfigsResponse {
+            configs: vec![],
+            next_page_token: None,
+        }
+    }
+
+    fn list_payload(req_id: i64) -> Vec<u8> {
+        serde_json::to_vec(&serde_json::json!({
+            "jsonrpc": "2.0",
+            "id": req_id,
+            "method": "tasks/pushNotificationConfig/list",
+            "params": list_request()
+        }))
+        .unwrap()
+    }
+
+    #[tokio::test]
+    async fn success_publishes_list() {
+        let nats = AdvancedMockNatsClient::new();
+        let handler = stub();
+        handler.lock().unwrap().push_list_result = Some(Ok(empty_response()));
+        handle(&handler, &list_payload(1), Some("r".into()), &nats).await;
+        let body = parse_response(&nats.published_payloads()[0]);
+        // configs is skip_serializing_if=Vec::is_empty, so an empty list omits the field.
+        assert!(body.get("result").is_some());
+        assert!(body["error"].is_null());
+    }
+
+    #[tokio::test]
+    async fn push_not_supported_error_uses_typed_code() {
+        let nats = AdvancedMockNatsClient::new();
+        let handler = stub();
+        handler.lock().unwrap().push_list_result = Some(Err(A2aError::push_notification_not_supported("no push")));
+        handle(&handler, &list_payload(2), Some("r".into()), &nats).await;
+        let body = parse_response(&nats.published_payloads()[0]);
+        assert_eq!(
+            body["error"]["code"].as_i64(),
+            Some(i64::from(crate::error::PUSH_NOTIFICATION_NOT_SUPPORTED))
+        );
+    }
+
+    #[tokio::test]
+    async fn no_reply_drops_request() {
+        let nats = AdvancedMockNatsClient::new();
+        let handler = stub();
+        handle(&handler, &list_payload(3), None, &nats).await;
+        assert!(nats.published_messages().is_empty());
+    }
+
+    #[tokio::test]
+    async fn missing_params_returns_invalid_params_error() {
+        let nats = AdvancedMockNatsClient::new();
+        let handler = stub();
+        let payload = serde_json::to_vec(&serde_json::json!({
+            "jsonrpc": "2.0",
+            "id": 5,
+            "method": "tasks/pushNotificationConfig/list"
+        }))
+        .unwrap();
+        handle(&handler, &payload, Some("r".into()), &nats).await;
+        let body = parse_response(&nats.published_payloads()[0]);
+        assert_eq!(body["error"]["code"], -32602);
+    }
+
+    #[tokio::test]
+    async fn invalid_params_shape_returns_invalid_params_code() {
+        let nats = AdvancedMockNatsClient::new();
+        let handler = stub();
+        let payload = serde_json::to_vec(&serde_json::json!({
+            "jsonrpc": "2.0",
+            "id": 6,
+            "method": "tasks/pushNotificationConfig/list",
+            "params": { "taskId": 42 }
+        }))
+        .unwrap();
+        handle(&handler, &payload, Some("r".into()), &nats).await;
+        let body = parse_response(&nats.published_payloads()[0]);
+        assert_eq!(body["error"]["code"], -32602);
+        assert_eq!(body["id"], 6);
+    }
+
+    #[tokio::test]
+    async fn malformed_json_still_publishes_parse_error_with_null_id() {
+        let nats = AdvancedMockNatsClient::new();
+        let handler = stub();
+        handle(&handler, b"not json", Some("r".into()), &nats).await;
+        let body = parse_response(&nats.published_payloads()[0]);
+        assert_eq!(body["error"]["code"], -32700);
+        assert!(body["id"].is_null());
+    }
+
+    #[tokio::test]
+    async fn notification_without_id_is_dropped() {
+        let nats = AdvancedMockNatsClient::new();
+        let handler = stub();
+        let payload = serde_json::to_vec(&serde_json::json!({
+            "jsonrpc": "2.0",
+            "method": "tasks/pushNotificationConfig/list",
+            "params": list_request()
+        }))
+        .unwrap();
+        handle(&handler, &payload, Some("r".into()), &nats).await;
+        assert!(nats.published_messages().is_empty());
+    }
+}

rsworkspace/crates/a2a-nats/src/server/test_support.rs

@@ -14,6 +14,7 @@ pub struct StubHandler {
     pub tasks_resubscribe_result: Option<Result<a2a::types::Task, A2aError>>,
     pub push_set_result: Option<Result<a2a::types::TaskPushNotificationConfig, A2aError>>,
     pub push_get_result: Option<Result<a2a::types::TaskPushNotificationConfig, A2aError>>,
+    pub push_list_result: Option<Result<a2a::types::ListTaskPushNotificationConfigsResponse, A2aError>>,
 }
 
 fn take_or_unimplemented<T>(slot: &mut Option<Result<T, A2aError>>) -> Result<T, A2aError> {
@@ -66,6 +67,13 @@ impl A2aExecutor for std::sync::Mutex<StubHandler> {
     ) -> Result<a2a::types::TaskPushNotificationConfig, A2aError> {
         take_or_unimplemented(&mut self.lock().unwrap().push_get_result)
     }
+
+    async fn push_notification_list(
+        &self,
+        _req: a2a::types::ListTaskPushNotificationConfigsRequest,
+    ) -> Result<a2a::types::ListTaskPushNotificationConfigsResponse, A2aError> {
+        take_or_unimplemented(&mut self.lock().unwrap().push_list_result)
+    }
 }
 
 pub fn stub() -> std::sync::Mutex<StubHandler> {