fix(client): serialize create-push requests for go interop (#35)

* fix(client): serialize create-push requests for go interop

Signed-off-by: Luca Muscariello <muscariello@ieee.org>

* test(client): cover jsonrpc push-config transport paths

Signed-off-by: Luca Muscariello <muscariello@ieee.org>

* build(just): emit html report from coverage-lcov

Signed-off-by: Luca Muscariello <muscariello@ieee.org>

---------

Signed-off-by: Luca Muscariello <muscariello@ieee.org>

Author: muscariello

a2a-client/src/jsonrpc.rs

@@ -9,6 +9,7 @@ use reqwest::Client;
 use crate::push_config_compat::{
     deserialize_list_task_push_notification_configs_response,
     deserialize_task_push_notification_config,
+    serialize_create_task_push_notification_config_request,
 };
 use crate::transport::{ServiceParams, Transport, TransportFactory};
 
@@ -26,19 +27,13 @@ impl JsonRpcTransport {
         JsonRpcTransport { client, endpoint }
     }
 
-    async fn call_value<Req>(
+    async fn call_value_with_payload(
         &self,
         params: &ServiceParams,
         method: &str,
-        request_params: &Req,
-    ) -> Result<serde_json::Value, A2AError>
-    where
-        Req: ProtoJsonPayload,
-    {
+        payload: serde_json::Value,
+    ) -> Result<serde_json::Value, A2AError> {
         let id = JsonRpcId::String(uuid::Uuid::now_v7().to_string());
-        let payload = protojson_conv::to_value(request_params).map_err(|e| {
-            A2AError::internal(format!("failed to serialize request as ProtoJSON: {e}"))
-        })?;
         let rpc_request = JsonRpcRequest::new(id, method, Some(payload));
 
         let mut builder = self.client.post(&self.endpoint);
@@ -63,11 +58,25 @@ impl JsonRpcTransport {
             return Err(A2AError::new(err.code, err.message));
         }
 
-        let result = rpc_response
+        rpc_response
             .result
-            .ok_or_else(|| A2AError::internal("JSON-RPC response missing result"))?;
+            .ok_or_else(|| A2AError::internal("JSON-RPC response missing result"))
+    }
+
+    async fn call_value<Req>(
+        &self,
+        params: &ServiceParams,
+        method: &str,
+        request_params: &Req,
+    ) -> Result<serde_json::Value, A2AError>
+    where
+        Req: ProtoJsonPayload,
+    {
+        let payload = protojson_conv::to_value(request_params).map_err(|e| {
+            A2AError::internal(format!("failed to serialize request as ProtoJSON: {e}"))
+        })?;
 
-        Ok(result)
+        self.call_value_with_payload(params, method, payload).await
     }
 
     async fn call<Req, Resp>(
@@ -341,8 +350,9 @@ impl Transport for JsonRpcTransport {
         params: &ServiceParams,
         req: &CreateTaskPushNotificationConfigRequest,
     ) -> Result<TaskPushNotificationConfig, A2AError> {
+        let payload = serialize_create_task_push_notification_config_request(req)?;
         let result = self
-            .call_value(params, methods::CREATE_PUSH_CONFIG, req)
+            .call_value_with_payload(params, methods::CREATE_PUSH_CONFIG, payload)
             .await?;
         deserialize_task_push_notification_config(result)
     }
@@ -456,6 +466,10 @@ mod tests {
     use super::*;
     use a2a_pb::protojson_conv;
     use futures::StreamExt;
+    use serde_json::{Value, json};
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+    use tokio::net::{TcpListener, TcpStream};
+    use tokio::sync::oneshot;
 
     /// Helper: build an SSE byte stream from raw text chunks.
     fn byte_stream(
@@ -469,6 +483,91 @@ mod tests {
         )
     }
 
+    async fn spawn_jsonrpc_server(response_body: String) -> (String, oneshot::Receiver<String>) {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        let (request_tx, request_rx) = oneshot::channel();
+
+        tokio::spawn(async move {
+            let (mut socket, _) = listener.accept().await.unwrap();
+            let request = read_http_request(&mut socket).await;
+            let response = format!(
+                "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
+                response_body.len(),
+                response_body,
+            );
+
+            let _ = request_tx.send(request);
+            socket.write_all(response.as_bytes()).await.unwrap();
+        });
+
+        (format!("http://{addr}"), request_rx)
+    }
+
+    async fn read_http_request(socket: &mut TcpStream) -> String {
+        let mut buffer = Vec::new();
+        let mut chunk = [0_u8; 1024];
+        let mut expected_len = None;
+
+        loop {
+            let read = socket.read(&mut chunk).await.unwrap();
+            if read == 0 {
+                break;
+            }
+            buffer.extend_from_slice(&chunk[..read]);
+
+            if expected_len.is_none() {
+                if let Some(header_end) = find_header_end(&buffer) {
+                    let headers = String::from_utf8_lossy(&buffer[..header_end]);
+                    expected_len = Some(header_end + parse_content_length(&headers));
+                }
+            }
+
+            if let Some(total_len) = expected_len {
+                if buffer.len() >= total_len {
+                    break;
+                }
+            }
+        }
+
+        String::from_utf8(buffer).unwrap()
+    }
+
+    fn find_header_end(buffer: &[u8]) -> Option<usize> {
+        buffer
+            .windows(4)
+            .position(|window| window == b"\r\n\r\n")
+            .map(|position| position + 4)
+    }
+
+    fn parse_content_length(headers: &str) -> usize {
+        headers
+            .lines()
+            .find_map(|line| {
+                let (name, value) = line.split_once(':')?;
+                name.eq_ignore_ascii_case("content-length")
+                    .then(|| value.trim().parse::<usize>().ok())
+                    .flatten()
+            })
+            .unwrap_or(0)
+    }
+
+    fn sample_create_push_config_request() -> CreateTaskPushNotificationConfigRequest {
+        CreateTaskPushNotificationConfigRequest {
+            task_id: "task-1".into(),
+            config: PushNotificationConfig {
+                url: "https://example.invalid/webhook".into(),
+                id: Some("cfg-1".into()),
+                token: Some("secret-token".into()),
+                authentication: Some(AuthenticationInfo {
+                    scheme: "Bearer".into(),
+                    credentials: Some("credential".into()),
+                }),
+            },
+            tenant: Some("tenant-1".into()),
+        }
+    }
+
     #[tokio::test]
     async fn test_parse_sse_stream_jsonrpc_envelope() {
         // Build a JSON-RPC response wrapping a StreamResponse (StatusUpdate)
@@ -705,4 +804,154 @@ mod tests {
         // Just verify it was created (it's a real transport but we can't call it without a server)
         transport.destroy().await.unwrap();
     }
+
+    #[tokio::test]
+    async fn test_create_push_config_sends_nested_request_shape() {
+        let response = json!({
+            "jsonrpc": "2.0",
+            "id": "1",
+            "result": {
+                "taskId": "task-1",
+                "config": {
+                    "url": "https://example.invalid/webhook",
+                    "id": "cfg-1",
+                    "token": "secret-token",
+                    "authentication": {
+                        "scheme": "Bearer",
+                        "credentials": "credential"
+                    }
+                },
+                "tenant": "tenant-1"
+            }
+        })
+        .to_string();
+        let (endpoint, request_rx) = spawn_jsonrpc_server(response).await;
+        let transport = JsonRpcTransport::new(Client::new(), endpoint);
+        let mut params = ServiceParams::new();
+        params.insert("x-trace".into(), vec!["alpha".into(), "beta".into()]);
+
+        let result = transport
+            .create_push_config(&params, &sample_create_push_config_request())
+            .await
+            .unwrap();
+
+        assert_eq!(result.task_id, "task-1");
+        assert_eq!(result.config.id.as_deref(), Some("cfg-1"));
+
+        let request = request_rx.await.unwrap();
+        let request_lower = request.to_ascii_lowercase();
+        assert!(request_lower.contains("x-trace: alpha"));
+        assert!(request_lower.contains("x-trace: beta"));
+
+        let body = request.split("\r\n\r\n").nth(1).unwrap();
+        let payload: Value = serde_json::from_str(body).unwrap();
+        assert_eq!(payload["method"], methods::CREATE_PUSH_CONFIG);
+        assert_eq!(
+            payload["params"],
+            json!({
+                "taskId": "task-1",
+                "config": {
+                    "url": "https://example.invalid/webhook",
+                    "id": "cfg-1",
+                    "token": "secret-token",
+                    "authentication": {
+                        "scheme": "Bearer",
+                        "credentials": "credential"
+                    }
+                },
+                "tenant": "tenant-1"
+            })
+        );
+    }
+
+    #[tokio::test]
+    async fn test_create_push_config_surfaces_jsonrpc_error() {
+        let response = json!({
+            "jsonrpc": "2.0",
+            "id": "1",
+            "error": {
+                "code": error_code::INVALID_PARAMS,
+                "message": "invalid params",
+                "data": null
+            }
+        })
+        .to_string();
+        let (endpoint, _request_rx) = spawn_jsonrpc_server(response).await;
+        let transport = JsonRpcTransport::new(Client::new(), endpoint);
+
+        let error = transport
+            .create_push_config(&ServiceParams::new(), &sample_create_push_config_request())
+            .await
+            .unwrap_err();
+
+        assert_eq!(error.code, error_code::INVALID_PARAMS);
+        assert_eq!(error.message, "invalid params");
+    }
+
+    #[tokio::test]
+    async fn test_create_push_config_rejects_missing_result() {
+        let response = json!({
+            "jsonrpc": "2.0",
+            "id": "1"
+        })
+        .to_string();
+        let (endpoint, _request_rx) = spawn_jsonrpc_server(response).await;
+        let transport = JsonRpcTransport::new(Client::new(), endpoint);
+
+        let error = transport
+            .create_push_config(&ServiceParams::new(), &sample_create_push_config_request())
+            .await
+            .unwrap_err();
+
+        assert_eq!(error.code, error_code::INTERNAL_ERROR);
+        assert_eq!(error.message, "JSON-RPC response missing result");
+    }
+
+    #[tokio::test]
+    async fn test_get_push_config_uses_protojson_request_path() {
+        let response = json!({
+            "jsonrpc": "2.0",
+            "id": "1",
+            "result": {
+                "taskId": "task-1",
+                "config": {
+                    "url": "https://example.invalid/webhook",
+                    "id": "cfg-1",
+                    "token": "secret-token"
+                },
+                "tenant": "tenant-1"
+            }
+        })
+        .to_string();
+        let (endpoint, request_rx) = spawn_jsonrpc_server(response).await;
+        let transport = JsonRpcTransport::new(Client::new(), endpoint);
+
+        let result = transport
+            .get_push_config(
+                &ServiceParams::new(),
+                &GetTaskPushNotificationConfigRequest {
+                    task_id: "task-1".into(),
+                    id: "cfg-1".into(),
+                    tenant: Some("tenant-1".into()),
+                },
+            )
+            .await
+            .unwrap();
+
+        assert_eq!(result.task_id, "task-1");
+        assert_eq!(result.config.id.as_deref(), Some("cfg-1"));
+
+        let request = request_rx.await.unwrap();
+        let body = request.split("\r\n\r\n").nth(1).unwrap();
+        let payload: Value = serde_json::from_str(body).unwrap();
+        assert_eq!(payload["method"], methods::GET_PUSH_CONFIG);
+        assert_eq!(
+            payload["params"],
+            json!({
+                "taskId": "task-1",
+                "id": "cfg-1",
+                "tenant": "tenant-1"
+            })
+        );
+    }
 }

a2a-client/src/push_config_compat.rs

@@ -4,6 +4,16 @@ use a2a::*;
 use a2a_pb::protojson_conv;
 use serde_json::Value;
 
+pub(crate) fn serialize_create_task_push_notification_config_request(
+    request: &CreateTaskPushNotificationConfigRequest,
+) -> Result<Value, A2AError> {
+    serde_json::to_value(request).map_err(|error| {
+        A2AError::internal(format!(
+            "failed to serialize create-push-config request: {error}"
+        ))
+    })
+}
+
 pub(crate) fn deserialize_task_push_notification_config(
     payload: Value,
 ) -> Result<TaskPushNotificationConfig, A2AError> {
@@ -40,6 +50,7 @@ pub(crate) fn deserialize_list_task_push_notification_configs_response(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use serde_json::json;
 
     fn sample_task_push_config() -> TaskPushNotificationConfig {
         TaskPushNotificationConfig {
@@ -57,6 +68,40 @@ mod tests {
         }
     }
 
+    fn sample_create_task_push_config_request() -> CreateTaskPushNotificationConfigRequest {
+        let config = sample_task_push_config();
+        CreateTaskPushNotificationConfigRequest {
+            task_id: config.task_id.clone(),
+            config: config.config,
+            tenant: config.tenant,
+        }
+    }
+
+    #[test]
+    fn serializes_nested_create_task_push_config_request_shape() {
+        let payload = serialize_create_task_push_notification_config_request(
+            &sample_create_task_push_config_request(),
+        )
+        .unwrap();
+
+        assert_eq!(
+            payload,
+            json!({
+                "taskId": "t1",
+                "config": {
+                    "id": "cfg1",
+                    "url": "https://example.invalid/webhook",
+                    "token": "token-1",
+                    "authentication": {
+                        "scheme": "Bearer",
+                        "credentials": "secret"
+                    }
+                },
+                "tenant": "tenant-1"
+            })
+        );
+    }
+
     #[test]
     fn parses_nested_task_push_config_shape() {
         let payload = serde_json::to_value(sample_task_push_config()).unwrap();