diff --git a/codex-rs/core/src/codex.rs b/codex-rs/core/src/codex.rs index 832592d40..e6dd69dfb 100644 --- a/codex-rs/core/src/codex.rs +++ b/codex-rs/core/src/codex.rs @@ -184,6 +184,7 @@ use crate::protocol::ModelRerouteEvent; use crate::protocol::ModelRerouteReason; use crate::protocol::NetworkApprovalContext; use crate::protocol::Op; +use crate::protocol::PatchApplyStatus; use crate::protocol::PlanDeltaEvent; use crate::protocol::RateLimitSnapshot; use crate::protocol::ReasoningContentDeltaEvent; @@ -2206,6 +2207,8 @@ impl Session { msg, }; self.send_event_raw(event).await; + self.maybe_mirror_event_text_to_realtime(&legacy_source) + .await; let show_raw_agent_reasoning = self.show_raw_agent_reasoning(); for legacy in legacy_source.as_legacy_events(show_raw_agent_reasoning) { @@ -2217,6 +2220,18 @@ impl Session { } } + async fn maybe_mirror_event_text_to_realtime(&self, msg: &EventMsg) { + let Some(text) = realtime_text_for_event(msg) else { + return; + }; + if self.conversation.running_state().await.is_none() { + return; + } + if let Err(err) = self.conversation.text_in(text).await { + debug!("failed to mirror event text to realtime conversation: {err}"); + } + } + pub(crate) async fn send_event_raw(&self, event: Event) { // Record the last known agent status. if let Some(status) = agent_status_from_event(&event.msg) { @@ -5451,6 +5466,56 @@ fn agent_message_text(item: &codex_protocol::items::AgentMessageItem) -> String .collect() } +fn realtime_text_for_event(msg: &EventMsg) -> Option { + match msg { + EventMsg::AgentMessage(event) => Some(event.message.clone()), + EventMsg::ItemCompleted(event) => match &event.item { + TurnItem::AgentMessage(item) => Some(agent_message_text(item)), + _ => None, + }, + EventMsg::ExecCommandBegin(event) => { + let command = event.command.join(" "); + Some(format!( + "Exec command started: {command}\nWorking directory: {}", + event.cwd.display() + )) + } + EventMsg::PatchApplyBegin(event) => { + let mut files: Vec = event + .changes + .keys() + .map(|path| path.display().to_string()) + .collect(); + files.sort(); + let file_list = if files.is_empty() { + "none".to_string() + } else { + files.join(", ") + }; + Some(format!( + "apply_patch started ({count} file change(s))\nFiles: {file_list}", + count = files.len() + )) + } + EventMsg::PatchApplyEnd(event) => { + let status = match event.status { + PatchApplyStatus::Completed => "completed", + PatchApplyStatus::Failed => "failed", + PatchApplyStatus::Declined => "declined", + }; + let mut text = format!("apply_patch {status}"); + if !event.stdout.is_empty() { + text.push_str(&format!("\nstdout:\n{}", event.stdout)); + } + if !event.stderr.is_empty() { + text.push_str(&format!("\nstderr:\n{}", event.stderr)); + } + Some(text) + } + _ => None, + } +} + /// Split the stream into normal assistant text vs. proposed plan content. /// Normal text becomes AgentMessage deltas; plan content becomes PlanDelta + /// TurnItem::Plan. diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index fc7b21e0e..c5015206f 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -9,6 +9,8 @@ use codex_protocol::protocol::Op; use codex_protocol::protocol::RealtimeAudioFrame; use codex_protocol::protocol::RealtimeConversationRealtimeEvent; use codex_protocol::protocol::RealtimeEvent; +use core_test_support::responses; +use core_test_support::responses::start_mock_server; use core_test_support::responses::start_websocket_server; use core_test_support::skip_if_no_network; use core_test_support::test_codex::test_codex; @@ -459,3 +461,82 @@ async fn conversation_uses_experimental_realtime_ws_backend_prompt_override() -> server.shutdown().await; Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_mirrors_assistant_message_text_to_realtime_websocket() -> Result<()> { + skip_if_no_network!(Ok(())); + + let api_server = start_mock_server().await; + let _response_mock = responses::mount_sse_once( + &api_server, + responses::sse(vec![ + responses::ev_response_created("resp_1"), + responses::ev_assistant_message("msg_1", "assistant says hi"), + responses::ev_completed("resp_1"), + ]), + ) + .await; + + let realtime_server = start_websocket_server(vec![vec![ + vec![json!({ + "type": "session.created", + "session": { "id": "sess_1" } + })], + vec![], + ]]) + .await; + + let mut builder = test_codex().with_config({ + let realtime_base_url = realtime_server.uri().to_string(); + move |config| { + config.experimental_realtime_ws_base_url = Some(realtime_base_url); + } + }); + let test = builder.build(&api_server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(ConversationStartParams { + prompt: "backend prompt".to_string(), + session_id: None, + })) + .await?; + + let session_created = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::SessionCreated { session_id }, + }) => Some(session_id.clone()), + _ => None, + }) + .await; + assert_eq!(session_created, "sess_1"); + + test.submit_turn("hello").await?; + + let deadline = tokio::time::Instant::now() + Duration::from_secs(2); + while tokio::time::Instant::now() < deadline { + let connections = realtime_server.connections(); + if connections.len() == 1 && connections[0].len() >= 2 { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + + let realtime_connections = realtime_server.connections(); + assert_eq!(realtime_connections.len(), 1); + assert_eq!(realtime_connections[0].len(), 2); + assert_eq!( + realtime_connections[0][0].body_json()["type"].as_str(), + Some("session.create") + ); + assert_eq!( + realtime_connections[0][1].body_json()["type"].as_str(), + Some("conversation.item.create") + ); + assert_eq!( + realtime_connections[0][1].body_json()["item"]["content"][0]["text"].as_str(), + Some("assistant says hi") + ); + + realtime_server.shutdown().await; + Ok(()) +}