From 5d36ff2fa5a356ec264a93ba59f44b5a4f0d4258 Mon Sep 17 00:00:00 2001 From: Raymond Li Date: Mon, 29 Jun 2026 18:24:15 -0700 Subject: [PATCH] fix(rmcp-client): accept empty notification responses --- .../rmcp-client/src/http_client_adapter.rs | 3 ++ .../tests/streamable_http_recovery.rs | 35 ++++++++++++ .../tests/streamable_http_test_support.rs | 54 +++++++++++++------ 3 files changed, 76 insertions(+), 16 deletions(-) diff --git a/codex-rs/rmcp-client/src/http_client_adapter.rs b/codex-rs/rmcp-client/src/http_client_adapter.rs index 19befb6235..46d8e043ee 100644 --- a/codex-rs/rmcp-client/src/http_client_adapter.rs +++ b/codex-rs/rmcp-client/src/http_client_adapter.rs @@ -215,6 +215,9 @@ impl StreamableHttpClient for StreamableHttpClientAdapter { } Some(content_type) if content_type.starts_with(JSON_MIME_TYPE) => { let body = collect_body(&mut body_stream).await?; + if body.is_empty() && matches!(message, JsonRpcMessage::Notification(_)) { + return Ok(StreamableHttpPostResponse::Accepted); + } let message: ServerJsonRpcMessage = serde_json::from_slice(&body).map_err(StreamableHttpError::Deserialize)?; Ok(StreamableHttpPostResponse::Json(message, session_id)) diff --git a/codex-rs/rmcp-client/tests/streamable_http_recovery.rs b/codex-rs/rmcp-client/tests/streamable_http_recovery.rs index 29d2404ede..a9187d65cc 100644 --- a/codex-rs/rmcp-client/tests/streamable_http_recovery.rs +++ b/codex-rs/rmcp-client/tests/streamable_http_recovery.rs @@ -18,7 +18,9 @@ use serde_json::Value; use streamable_http_test_support::arm_initialize_post_failure; use streamable_http_test_support::arm_initialize_post_json_rpc_failure; +use streamable_http_test_support::arm_initialize_post_response; use streamable_http_test_support::arm_initialized_notification_post_json_rpc_failure; +use streamable_http_test_support::arm_initialized_notification_post_response; use streamable_http_test_support::arm_session_post_failure; use streamable_http_test_support::arm_session_post_json_rpc_failure; use streamable_http_test_support::call_echo_tool; @@ -148,6 +150,39 @@ async fn streamable_http_initialize_retries_json_rpc_transient_status() -> anyho Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn streamable_http_rejects_empty_ok_for_initialize_request() -> anyhow::Result<()> { + let (_server, base_url) = spawn_streamable_http_server().await?; + + arm_initialize_post_response(&base_url, /*status*/ 200, /*remaining*/ 10, "").await?; + + let Err(_) = create_client(&base_url).await else { + anyhow::bail!("empty response unexpectedly accepted for initialize request"); + }; + + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +async fn streamable_http_accepts_empty_ok_for_initialized_notification() -> anyhow::Result<()> { + let (_server, base_url) = spawn_streamable_http_server().await?; + + arm_initialized_notification_post_response( + &base_url, /*status*/ 200, /*remaining*/ 10, "", + ) + .await?; + + let client = create_client(&base_url).await?; + let result = call_echo_tool(&client, "after-empty-notification-response").await?; + + assert_eq!( + result, + expected_echo_result("after-empty-notification-response") + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 1)] async fn streamable_http_retries_initialized_notification_status() -> anyhow::Result<()> { let (_server, base_url) = spawn_streamable_http_server().await?; diff --git a/codex-rs/rmcp-client/tests/streamable_http_test_support.rs b/codex-rs/rmcp-client/tests/streamable_http_test_support.rs index 87a52f55df..f11dd1d807 100644 --- a/codex-rs/rmcp-client/tests/streamable_http_test_support.rs +++ b/codex-rs/rmcp-client/tests/streamable_http_test_support.rs @@ -229,6 +229,24 @@ pub(crate) async fn arm_initialized_notification_post_json_rpc_failure( base_url: &str, status: u16, remaining: usize, +) -> anyhow::Result<()> { + let body = json!({ + "jsonrpc": "2.0", + "id": 1, + "error": { + "code": -32000, + "message": "transient session failure", + }, + }) + .to_string(); + arm_initialized_notification_post_response(base_url, status, remaining, &body).await +} + +pub(crate) async fn arm_initialized_notification_post_response( + base_url: &str, + status: u16, + remaining: usize, + body: &str, ) -> anyhow::Result<()> { let response = reqwest::Client::new() .post(format!( @@ -238,14 +256,7 @@ pub(crate) async fn arm_initialized_notification_post_json_rpc_failure( "status": status, "remaining": remaining, "content_type": "application/json", - "body": json!({ - "jsonrpc": "2.0", - "id": 1, - "error": { - "code": -32000, - "message": "transient session failure", - }, - }).to_string(), + "body": body, })) .send() .await?; @@ -276,6 +287,24 @@ pub(crate) async fn arm_initialize_post_json_rpc_failure( base_url: &str, status: u16, remaining: usize, +) -> anyhow::Result<()> { + let body = json!({ + "jsonrpc": "2.0", + "id": 1, + "error": { + "code": -32000, + "message": "transient initialize failure", + }, + }) + .to_string(); + arm_initialize_post_response(base_url, status, remaining, &body).await +} + +pub(crate) async fn arm_initialize_post_response( + base_url: &str, + status: u16, + remaining: usize, + body: &str, ) -> anyhow::Result<()> { let response = reqwest::Client::new() .post(format!("{base_url}{INITIALIZE_POST_FAILURE_CONTROL_PATH}")) @@ -283,14 +312,7 @@ pub(crate) async fn arm_initialize_post_json_rpc_failure( "status": status, "remaining": remaining, "content_type": "application/json", - "body": json!({ - "jsonrpc": "2.0", - "id": 1, - "error": { - "code": -32000, - "message": "transient initialize failure", - }, - }).to_string(), + "body": body, })) .send() .await?;