diff --git a/codex-rs/app-server/tests/common/mcp_process.rs b/codex-rs/app-server/tests/common/mcp_process.rs index cbf54cecf0..3d76c7a115 100644 --- a/codex-rs/app-server/tests/common/mcp_process.rs +++ b/codex-rs/app-server/tests/common/mcp_process.rs @@ -1,3 +1,4 @@ +use std::collections::VecDeque; use std::path::Path; use std::process::Stdio; use std::sync::atomic::AtomicI64; @@ -46,6 +47,7 @@ pub struct McpProcess { process: Child, stdin: ChildStdin, stdout: BufReader, + pending_user_messages: VecDeque, } impl McpProcess { @@ -116,6 +118,7 @@ impl McpProcess { process, stdin, stdout, + pending_user_messages: VecDeque::new(), }) } @@ -380,8 +383,9 @@ impl McpProcess { let message = self.read_jsonrpc_message().await?; match message { - JSONRPCMessage::Notification(_) => { - eprintln!("notification: {message:?}"); + JSONRPCMessage::Notification(notification) => { + eprintln!("notification: {notification:?}"); + self.enqueue_user_message(notification); } JSONRPCMessage::Request(jsonrpc_request) => { return Ok(jsonrpc_request); @@ -405,8 +409,9 @@ impl McpProcess { loop { let message = self.read_jsonrpc_message().await?; match message { - JSONRPCMessage::Notification(_) => { - eprintln!("notification: {message:?}"); + JSONRPCMessage::Notification(notification) => { + eprintln!("notification: {notification:?}"); + self.enqueue_user_message(notification); } JSONRPCMessage::Request(_) => { anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}"); @@ -430,8 +435,9 @@ impl McpProcess { loop { let message = self.read_jsonrpc_message().await?; match message { - JSONRPCMessage::Notification(_) => { - eprintln!("notification: {message:?}"); + JSONRPCMessage::Notification(notification) => { + eprintln!("notification: {notification:?}"); + self.enqueue_user_message(notification); } JSONRPCMessage::Request(_) => { anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}"); @@ -454,6 +460,10 @@ impl McpProcess { ) -> anyhow::Result { eprintln!("in read_stream_until_notification_message({method})"); + if let Some(notification) = self.take_pending_notification_by_method(method) { + return Ok(notification); + } + loop { let message = self.read_jsonrpc_message().await?; match message { @@ -461,6 +471,7 @@ impl McpProcess { if notification.method == method { return Ok(notification); } + self.enqueue_user_message(notification); } JSONRPCMessage::Request(_) => { anyhow::bail!("unexpected JSONRPCMessage::Request: {message:?}"); @@ -474,4 +485,21 @@ impl McpProcess { } } } + + fn take_pending_notification_by_method(&mut self, method: &str) -> Option { + if let Some(pos) = self + .pending_user_messages + .iter() + .position(|notification| notification.method == method) + { + return self.pending_user_messages.remove(pos); + } + None + } + + fn enqueue_user_message(&mut self, notification: JSONRPCNotification) { + if notification.method == "codex/event/user_message" { + self.pending_user_messages.push_back(notification); + } + } } diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 3fc0da7829..3d7a0be639 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -1286,8 +1286,19 @@ async fn submission_loop( let previous_env_context = EnvironmentContext::from(turn_context.as_ref()); let new_env_context = EnvironmentContext::from(&fresh_turn_context); if !new_env_context.equals_except_shell(&previous_env_context) { - sess.record_conversation_items(&[ResponseItem::from(new_env_context)]) + let env_response_item = ResponseItem::from(new_env_context); + sess.record_conversation_items(std::slice::from_ref(&env_response_item)) .await; + for msg in map_response_item_to_event_messages( + &env_response_item, + sess.show_raw_agent_reasoning(), + ) { + let event = Event { + id: sub.id.clone(), + msg, + }; + sess.send_event(event).await; + } } // Install the new persistent context for subsequent tasks/turns.