From 002f9fe228058b5a920be607df3c8df447def1d3 Mon Sep 17 00:00:00 2001 From: Guinness Chen Date: Tue, 9 Jun 2026 17:16:24 -0700 Subject: [PATCH] Forward realtime assistant output through handoffs --- .../tests/suite/v2/realtime_conversation.rs | 248 ++++++++---------- .../endpoint/realtime_websocket/methods.rs | 12 - .../endpoint/realtime_websocket/protocol.rs | 1 - codex-rs/core/src/realtime_conversation.rs | 176 ++++--------- .../core/src/realtime_conversation_tests.rs | 110 ++++---- codex-rs/core/src/session/mod.rs | 36 ++- .../core/tests/suite/realtime_conversation.rs | 152 ++++++++++- 7 files changed, 381 insertions(+), 354 deletions(-) diff --git a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs index 4fa3d597de..3bff4015b6 100644 --- a/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs +++ b/codex-rs/app-server/tests/suite/v2/realtime_conversation.rs @@ -48,7 +48,6 @@ use codex_protocol::protocol::RealtimeVoice; use codex_protocol::protocol::RealtimeVoicesList; use core_test_support::responses; use core_test_support::responses::WebSocketConnectionConfig; -use core_test_support::responses::WebSocketRequest; use core_test_support::responses::WebSocketTestServer; use core_test_support::responses::start_websocket_server; use core_test_support::responses::start_websocket_server_with_headers; @@ -79,8 +78,6 @@ const DELEGATED_SHELL_TOOL_TIMEOUT_MS: u64 = 30_000; const STARTUP_CONTEXT_HEADER: &str = "Startup context from Codex."; const V2_STEERING_ACKNOWLEDGEMENT: &str = "This was sent to steer the previous background agent task."; -const V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT: &str = - "Background agent finished. Use the preceding [BACKEND] messages as the result."; #[derive(Debug, Clone, Copy)] enum StartupContextConfig<'a> { @@ -419,6 +416,24 @@ fn no_main_loop_responses() -> MainLoopResponsesScript { main_loop_responses(Vec::new()) } +fn assistant_preamble_and_final_sse_response(preamble: &str, final_answer: &str) -> String { + responses::sse(vec![ + responses::ev_response_created("resp-1"), + json!({ + "type": "response.output_item.done", + "item": { + "type": "message", + "role": "assistant", + "id": "msg-preamble", + "phase": "commentary", + "content": [{"type": "output_text", "text": preamble}] + } + }), + responses::ev_assistant_message("msg-final", final_answer), + responses::ev_completed("resp-1"), + ]) +} + fn realtime_sideband(connections: Vec) -> RealtimeSidebandScript { RealtimeSidebandScript { connections } } @@ -1256,9 +1271,10 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()> // Phase 1: script one v1 handoff request on the sideband and one delegated Responses turn. let mut harness = RealtimeE2eHarness::new( RealtimeTestVersion::V1, - main_loop_responses(vec![create_final_assistant_message_sse_response( + main_loop_responses(vec![assistant_preamble_and_final_sse_response( + "working on v1 delegation", "delegated from v1", - )?]), + )]), realtime_sideband(vec![realtime_sideband_connection(vec![ vec![ session_updated("sess_v1_handoff"), @@ -1282,6 +1298,7 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()> }), ], vec![], + vec![], ])]), ) .await?; @@ -1317,9 +1334,18 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()> "delegated Responses request should contain realtime delegation envelope: {}", requests[0] ); - let handoff_append = harness.sideband_outbound_request(/*request_index*/ 1).await; + let preamble_append = harness.sideband_outbound_request(/*request_index*/ 1).await; assert_eq!( - handoff_append, + preamble_append, + json!({ + "type": "conversation.handoff.append", + "handoff_id": "handoff_v1", + "output_text": "working on v1 delegation", + }) + ); + let final_append = harness.sideband_outbound_request(/*request_index*/ 2).await; + assert_eq!( + final_append, json!({ "type": "conversation.handoff.append", "handoff_id": "handoff_v1", @@ -1335,52 +1361,47 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()> async fn webrtc_terminal_output_without_handoff_reaches_realtime() -> Result<()> { skip_if_no_network!(Ok(())); - for (version, output_texts) in [ + for (version, outputs) in [ ( RealtimeTestVersion::V1, [ - "first direct result from v1", - "second direct result from v1", + ( + "first direct preamble from v1", + "first direct result from v1", + ), + ( + "second direct preamble from v1", + "second direct result from v1", + ), ], ), ( RealtimeTestVersion::V2, [ - "first direct result from v2", - "second direct result from v2", + ( + "first direct preamble from v2", + "first direct result from v2", + ), + ( + "second direct preamble from v2", + "second direct result from v2", + ), ], ), ] { - let sideband_responses = match version { - RealtimeTestVersion::V1 => vec![ - vec![session_updated("sess_terminal_output")], - vec![], - vec![], - ], - RealtimeTestVersion::V2 => vec![ - vec![session_updated("sess_terminal_output")], - vec![], - vec![ - json!({ - "type": "response.created", - "response": { "id": "resp_direct_1" }, - }), - json!({ - "type": "response.done", - "response": { "id": "resp_direct_1" }, - }), - ], - vec![], - vec![], - ], - }; let mut harness = RealtimeE2eHarness::new( version, main_loop_responses(vec![ - create_final_assistant_message_sse_response(output_texts[0])?, - create_final_assistant_message_sse_response(output_texts[1])?, + assistant_preamble_and_final_sse_response(outputs[0].0, outputs[0].1), + assistant_preamble_and_final_sse_response(outputs[1].0, outputs[1].1), ]), - realtime_sideband(vec![realtime_sideband_connection(sideband_responses)]), + realtime_sideband(vec![realtime_sideband_connection(vec![ + vec![session_updated("sess_terminal_output")], + vec![], + vec![], + vec![], + vec![], + ])]), ) .await?; @@ -1393,7 +1414,7 @@ async fn webrtc_terminal_output_without_handoff_reaches_realtime() -> Result<()> } ); - for (turn_index, output_text) in output_texts.into_iter().enumerate() { + for (turn_index, (preamble, final_answer)) in outputs.into_iter().enumerate() { let turn_request_id = harness .mcp .send_turn_start_request(TurnStartParams { @@ -1417,32 +1438,20 @@ async fn webrtc_terminal_output_without_handoff_reaches_realtime() -> Result<()> .read_notification::("turn/completed") .await?; - let request_index = match version { - RealtimeTestVersion::V1 => 1 + turn_index, - RealtimeTestVersion::V2 => 1 + turn_index * 2, - }; - assert_eq!( - harness.sideband_outbound_request(request_index).await, - json!({ - "type": "conversation.item.create", - "item": { - "type": "message", - "role": "developer", - "content": [{ - "type": "input_text", - "text": format!("Speak the following text:\n{output_text}"), - }], - }, - }) + assert_handoff_append( + &harness + .sideband_outbound_request(/*request_index*/ 1 + turn_index * 2) + .await, + "codex", + preamble, + ); + assert_handoff_append( + &harness + .sideband_outbound_request(/*request_index*/ 2 + turn_index * 2) + .await, + "codex", + final_answer, ); - match version { - RealtimeTestVersion::V1 => {} - RealtimeTestVersion::V2 => { - assert_v2_response_create( - &harness.sideband_outbound_request(request_index + 1).await, - ); - } - } } harness.shutdown().await; @@ -1664,23 +1673,21 @@ async fn webrtc_v2_text_input_is_append_only_when_response_is_cancelled() -> Res Ok(()) } -/// Regression coverage for the Realtime V2 background-agent final-output path. +/// Regression coverage for Realtime V2 background-agent assistant output. /// -/// Once the background agent finishes, app-server sends the final function-call -/// output to realtime and then requests a new `response.create` so realtime can -/// react to that final output. +/// Preambles and the final answer use the same handoff channel and handoff ID. #[tokio::test] -async fn webrtc_v2_background_agent_tool_call_delegates_and_returns_function_output() -> Result<()> -{ +async fn webrtc_v2_background_agent_tool_call_appends_preamble_and_final() -> Result<()> { skip_if_no_network!(Ok(())); // Phase 1: script a v2 background agent function call and a delegated Responses turn that // returns final assistant text. let mut harness = RealtimeE2eHarness::new( RealtimeTestVersion::V2, - main_loop_responses(vec![create_final_assistant_message_sse_response( + main_loop_responses(vec![assistant_preamble_and_final_sse_response( + "working on v2 delegation", "delegated from v2", - )?]), + )]), realtime_sideband(vec![realtime_sideband_connection(vec![ vec![ session_updated("sess_v2_tool"), @@ -1715,7 +1722,6 @@ async fn webrtc_v2_background_agent_tool_call_delegates_and_returns_function_out ], vec![], vec![], - vec![], ])]), ) .await?; @@ -1733,8 +1739,8 @@ async fn webrtc_v2_background_agent_tool_call_delegates_and_returns_function_out .await?; assert_eq!(turn_completed.thread_id, harness.thread_id); - // Phase 3: assert the delegated prompt went to Responses and the result - // returned as exactly one v2 function-call output event on the sideband. + // Phase 3: assert the delegated prompt went to Responses and both assistant messages + // returned through the handoff channel. let requests = harness.main_loop_responses_requests().await?; assert_eq!(requests.len(), 1); assert!( @@ -1751,19 +1757,16 @@ async fn webrtc_v2_background_agent_tool_call_delegates_and_returns_function_out requests[0] ); - let terminal_output = harness.sideband_outbound_request(/*request_index*/ 1).await; - assert_v2_terminal_output(&terminal_output, "delegated from v2"); - - let tool_output = harness.sideband_outbound_request(/*request_index*/ 2).await; - assert_v2_function_call_output(&tool_output, "call_v2", V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT); - assert_eq!( - function_call_output_sideband_requests(&harness.realtime_server).len(), - 1 + assert_handoff_append( + &harness.sideband_outbound_request(/*request_index*/ 1).await, + "call_v2", + "working on v2 delegation", + ); + assert_handoff_append( + &harness.sideband_outbound_request(/*request_index*/ 2).await, + "call_v2", + "delegated from v2", ); - - // Phase 4: after the final function-call output, realtime needs an explicit - // `response.create` to produce the next user-visible response. - assert_v2_response_create(&harness.sideband_outbound_request(/*request_index*/ 3).await); harness.shutdown().await; Ok(()) @@ -1859,7 +1862,7 @@ async fn webrtc_v2_background_agent_steering_ack_requests_response_create() -> R } #[tokio::test] -async fn webrtc_v2_terminal_output_is_sent_before_function_output() -> Result<()> { +async fn webrtc_v2_terminal_output_uses_handoff_append() -> Result<()> { skip_if_no_network!(Ok(())); let mut harness = RealtimeE2eHarness::new( @@ -1873,7 +1876,6 @@ async fn webrtc_v2_terminal_output_is_sent_before_function_output() -> Result<() v2_background_agent_tool_call("call_progress_order", "stream progress"), ], vec![], - vec![], ])]), ) .await?; @@ -1886,14 +1888,10 @@ async fn webrtc_v2_terminal_output_is_sent_before_function_output() -> Result<() .await?; assert_eq!(turn_completed.thread_id, harness.thread_id); - let terminal_output = harness.sideband_outbound_request(/*request_index*/ 1).await; - assert_v2_terminal_output(&terminal_output, "progress before final"); - - let tool_output = harness.sideband_outbound_request(/*request_index*/ 2).await; - assert_v2_function_call_output( - &tool_output, + assert_handoff_append( + &harness.sideband_outbound_request(/*request_index*/ 1).await, "call_progress_order", - V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT, + "progress before final", ); harness.shutdown().await; @@ -1924,7 +1922,6 @@ async fn webrtc_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<( v2_background_agent_tool_call("call_shell", "run shell through delegated turn"), ], vec![], - vec![], ])]); let mut harness = RealtimeE2eHarness::new_with_sandbox( @@ -1962,7 +1959,7 @@ async fn webrtc_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<( assert_eq!(aggregated_output.as_deref(), Some("realtime-tool-ok")); // Phase 3: verify the shell output reached Responses and the final delegated answer returned - // to realtime as a single function-call-output item. + // through the handoff channel. let turn_completed = harness .read_notification::("turn/completed") .await?; @@ -1976,18 +1973,10 @@ async fn webrtc_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<( requests[1] ); - let terminal_output = harness.sideband_outbound_request(/*request_index*/ 1).await; - assert_v2_terminal_output(&terminal_output, "shell tool finished"); - - let tool_output = harness.sideband_outbound_request(/*request_index*/ 2).await; - assert_v2_function_call_output( - &tool_output, + assert_handoff_append( + &harness.sideband_outbound_request(/*request_index*/ 1).await, "call_shell", - V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT, - ); - assert_eq!( - function_call_output_sideband_requests(&harness.realtime_server).len(), - 1 + "shell tool finished", ); harness.shutdown().await; @@ -2033,7 +2022,6 @@ async fn webrtc_v2_tool_call_does_not_block_sideband_audio() -> Result<()> { }), ], vec![], - vec![], ])]), ) .await?; @@ -2052,22 +2040,18 @@ async fn webrtc_v2_tool_call_does_not_block_sideband_audio() -> Result<()> { .await?; assert_eq!(audio.audio.data, "CQoL"); - // Phase 3: release the delegated turn and assert the sideband function-call output is delivered - // after the nonblocking audio. + // Phase 3: release the delegated turn and assert the handoff output is delivered after the + // nonblocking audio. let _ = gate_completed_tx.send(()); let turn_completed = harness .read_notification::("turn/completed") .await?; assert_eq!(turn_completed.thread_id, harness.thread_id); - let terminal_output = harness.sideband_outbound_request(/*request_index*/ 1).await; - assert_v2_terminal_output(&terminal_output, "late delegated result"); - - let tool_output = harness.sideband_outbound_request(/*request_index*/ 2).await; - assert_v2_function_call_output( - &tool_output, + assert_handoff_append( + &harness.sideband_outbound_request(/*request_index*/ 1).await, "call_audio", - V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT, + "late delegated result", ); harness.shutdown().await; @@ -2293,18 +2277,6 @@ fn realtime_tool_ok_command() -> Vec { } } -fn function_call_output_sideband_requests(server: &WebSocketTestServer) -> Vec { - server - .single_connection() - .iter() - .map(WebSocketRequest::body_json) - .filter(|request| { - request["type"] == "conversation.item.create" - && request["item"]["type"] == "function_call_output" - }) - .collect() -} - fn assert_v2_function_call_output(request: &Value, call_id: &str, expected_output: &str) { assert_eq!( request, @@ -2319,19 +2291,13 @@ fn assert_v2_function_call_output(request: &Value, call_id: &str, expected_outpu ); } -fn assert_v2_terminal_output(request: &Value, expected_text: &str) { +fn assert_handoff_append(request: &Value, handoff_id: &str, output_text: &str) { assert_eq!( request, &json!({ - "type": "conversation.item.create", - "item": { - "type": "message", - "role": "user", - "content": [{ - "type": "input_text", - "text": format!("[BACKEND] {expected_text}") - }] - } + "type": "conversation.handoff.append", + "handoff_id": handoff_id, + "output_text": output_text, }) ); } diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs index 3e8b401d96..f13bb68ca3 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs @@ -306,18 +306,6 @@ impl RealtimeWebsocketWriter { .await } - pub async fn send_conversation_developer_item_create( - &self, - text: String, - ) -> Result<(), ApiError> { - self.send_json(&conversation_item_create_message( - self.event_parser, - ConversationRole::Developer, - text, - )) - .await - } - pub async fn send_conversation_function_call_output( &self, call_id: String, diff --git a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs index 1e5f228050..a5c7bde2cf 100644 --- a/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs +++ b/codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs @@ -172,7 +172,6 @@ pub(super) enum ConversationItemType { #[derive(Debug, Clone, Copy, Serialize)] #[serde(rename_all = "snake_case")] pub(super) enum ConversationRole { - Developer, User, } diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 215a99143b..7b12368df9 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -61,14 +61,12 @@ use tracing::warn; const AUDIO_IN_QUEUE_CAPACITY: usize = 256; const USER_TEXT_IN_QUEUE_CAPACITY: usize = 64; -const TERMINAL_OUTPUT_QUEUE_CAPACITY: usize = 64; +const ASSISTANT_OUTPUT_QUEUE_CAPACITY: usize = 64; const OUTPUT_EVENTS_QUEUE_CAPACITY: usize = 256; const REALTIME_STARTUP_CONTEXT_TOKEN_BUDGET: usize = 5_300; const DEFAULT_REALTIME_MODEL: &str = "gpt-realtime-1.5"; pub(crate) const REALTIME_USER_TEXT_PREFIX: &str = "[USER] "; -pub(crate) const REALTIME_BACKEND_TEXT_PREFIX: &str = "[BACKEND] "; -const REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT: &str = - "Background agent finished. Use the preceding [BACKEND] messages as the result."; +const DEFAULT_HANDOFF_ID: &str = "codex"; const REALTIME_V2_STEER_ACKNOWLEDGEMENT: &str = "This was sent to steer the previous background agent task."; const REALTIME_ACTIVE_RESPONSE_ERROR_PREFIX: &str = @@ -98,20 +96,14 @@ enum RealtimeSessionKind { #[derive(Clone, Debug)] struct RealtimeHandoffState { - terminal_output_tx: Sender, + assistant_output_tx: Sender, active_handoff: Arc>>, - session_kind: RealtimeSessionKind, } #[derive(Debug, PartialEq, Eq)] -enum RealtimeTerminalOutput { - Direct { - output_text: String, - }, - Handoff { - handoff_id: String, - output_text: String, - }, +struct RealtimeAssistantOutput { + handoff_id: Option, + output_text: String, } #[derive(Debug, PartialEq, Eq)] @@ -186,50 +178,37 @@ struct RealtimeInputTask { writer: RealtimeWebsocketWriter, events: RealtimeWebsocketEvents, user_text_rx: Receiver, - terminal_output_rx: Receiver, + assistant_output_rx: Receiver, audio_rx: Receiver, events_tx: Sender, handoff_state: RealtimeHandoffState, session_kind: RealtimeSessionKind, - event_parser: RealtimeEventParser, } struct RealtimeInputChannels { user_text_rx: Receiver, - terminal_output_rx: Receiver, + assistant_output_rx: Receiver, audio_rx: Receiver, } impl RealtimeHandoffState { - fn new( - terminal_output_tx: Sender, - session_kind: RealtimeSessionKind, - ) -> Self { + fn new(assistant_output_tx: Sender) -> Self { Self { - terminal_output_tx, + assistant_output_tx, active_handoff: Arc::new(Mutex::new(None)), - session_kind, } } - async fn take_terminal_output( - &self, - output_text: Option, - ) -> Option { - let handoff_id = self.active_handoff.lock().await.take(); - let output_text = output_text?; - match handoff_id { - Some(handoff_id) => Some(RealtimeTerminalOutput::Handoff { - handoff_id, - output_text: prefix_realtime_text( - output_text, - REALTIME_BACKEND_TEXT_PREFIX, - self.session_kind, - ), - }), - None => Some(RealtimeTerminalOutput::Direct { output_text }), + async fn assistant_output(&self, output_text: String) -> RealtimeAssistantOutput { + RealtimeAssistantOutput { + handoff_id: self.active_handoff.lock().await.clone(), + output_text, } } + + async fn finish_turn(&self) { + self.active_handoff.lock().await.take(); + } } #[allow(dead_code)] @@ -312,16 +291,16 @@ impl RealtimeConversationManager { async_channel::bounded::(AUDIO_IN_QUEUE_CAPACITY); let (user_text_tx, user_text_rx) = async_channel::bounded::(USER_TEXT_IN_QUEUE_CAPACITY); - let (terminal_output_tx, terminal_output_rx) = - async_channel::bounded::(TERMINAL_OUTPUT_QUEUE_CAPACITY); + let (assistant_output_tx, assistant_output_rx) = + async_channel::bounded::(ASSISTANT_OUTPUT_QUEUE_CAPACITY); let (events_tx, events_rx) = async_channel::bounded::(OUTPUT_EVENTS_QUEUE_CAPACITY); let realtime_active = Arc::new(AtomicBool::new(true)); - let handoff = RealtimeHandoffState::new(terminal_output_tx, session_kind); + let handoff = RealtimeHandoffState::new(assistant_output_tx); let input_channels = RealtimeInputChannels { user_text_rx, - terminal_output_rx, + assistant_output_rx, audio_rx, }; @@ -343,7 +322,6 @@ impl RealtimeConversationManager { events_tx, handoff_state: handoff.clone(), session_kind, - event_parser, realtime_active: Arc::clone(&realtime_active), }); (task, Some(call.sdp)) @@ -360,12 +338,11 @@ impl RealtimeConversationManager { writer: connection.writer(), events: connection.events(), user_text_rx: input_channels.user_text_rx, - terminal_output_rx: input_channels.terminal_output_rx, + assistant_output_rx: input_channels.assistant_output_rx, audio_rx: input_channels.audio_rx, events_tx, handoff_state: handoff.clone(), session_kind, - event_parser, }); (task, None) }; @@ -468,7 +445,7 @@ impl RealtimeConversationManager { Ok(()) } - pub(crate) async fn finish_turn(&self, output_text: Option) -> CodexResult<()> { + pub(crate) async fn send_assistant_output(&self, output_text: String) -> CodexResult<()> { let handoff = { let guard = self.state.lock().await; guard.as_ref().map(|state| state.handoff.clone()) @@ -477,17 +454,25 @@ impl RealtimeConversationManager { return Ok(()); }; - let Some(terminal_output) = handoff.take_terminal_output(output_text).await else { - return Ok(()); - }; handoff - .terminal_output_tx - .send(terminal_output) + .assistant_output_tx + .send(handoff.assistant_output(output_text).await) .await .map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?; Ok(()) } + pub(crate) async fn finish_turn(&self) { + let handoff = { + let guard = self.state.lock().await; + guard.as_ref().map(|state| state.handoff.clone()) + }; + let Some(handoff) = handoff else { + return; + }; + handoff.finish_turn().await; + } + pub(crate) async fn shutdown(&self) -> CodexResult<()> { let state = { let mut guard = self.state.lock().await; @@ -987,7 +972,6 @@ struct RealtimeWebrtcSidebandInputTask { events_tx: Sender, handoff_state: RealtimeHandoffState, session_kind: RealtimeSessionKind, - event_parser: RealtimeEventParser, realtime_active: Arc, } @@ -1001,7 +985,6 @@ fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> J events_tx, handoff_state, session_kind, - event_parser, realtime_active, } = input; @@ -1040,12 +1023,11 @@ fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> J writer: connection.writer(), events: connection.events(), user_text_rx: input_channels.user_text_rx, - terminal_output_rx: input_channels.terminal_output_rx, + assistant_output_rx: input_channels.assistant_output_rx, audio_rx: input_channels.audio_rx, events_tx, handoff_state, session_kind, - event_parser, }) .await; }) @@ -1056,12 +1038,11 @@ async fn run_realtime_input_task(input: RealtimeInputTask) { writer, events, user_text_rx, - terminal_output_rx, + assistant_output_rx, audio_rx, events_tx, handoff_state, session_kind, - event_parser, } = input; let mut output_audio_state: Option = None; @@ -1078,14 +1059,12 @@ async fn run_realtime_input_task(input: RealtimeInputTask) { ) .await } - // Terminal Codex output that should be sent back to realtime. - terminal_output = terminal_output_rx.recv() => { - handle_terminal_output( - terminal_output, + // Assistant output that should be sent back to realtime. + assistant_output = assistant_output_rx.recv() => { + handle_assistant_output( + assistant_output, &writer, &events_tx, - event_parser, - &mut response_create_queue, ) .await } @@ -1132,70 +1111,31 @@ async fn handle_user_text_input( Ok(()) } -async fn handle_terminal_output( - terminal_output: Result, +async fn handle_assistant_output( + assistant_output: Result, writer: &RealtimeWebsocketWriter, events_tx: &Sender, - event_parser: RealtimeEventParser, - response_create_queue: &mut RealtimeResponseCreateQueue, ) -> anyhow::Result<()> { - let terminal_output = terminal_output.context("terminal output channel closed")?; - - let (result, request_response) = match terminal_output { - RealtimeTerminalOutput::Direct { output_text } => { - let request_response = match event_parser { - RealtimeEventParser::V1 => false, - RealtimeEventParser::RealtimeV2 => true, - }; - ( - writer - .send_conversation_developer_item_create(format!( - "Speak the following text:\n{output_text}" - )) - .await, - request_response, - ) - } - RealtimeTerminalOutput::Handoff { - handoff_id, - output_text, - } => match event_parser { - RealtimeEventParser::V1 => ( - writer - .send_conversation_handoff_append(Some(handoff_id), output_text) - .await, - false, + let assistant_output = assistant_output.context("assistant output channel closed")?; + if let Err(err) = writer + .send_conversation_handoff_append( + Some( + assistant_output + .handoff_id + .unwrap_or_else(|| DEFAULT_HANDOFF_ID.to_string()), ), - RealtimeEventParser::RealtimeV2 => ( - if let Err(err) = writer.send_conversation_item_create(output_text).await { - Err(err) - } else { - writer - .send_conversation_function_call_output( - handoff_id, - REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT.to_string(), - ) - .await - }, - true, - ), - }, - }; - if let Err(err) = result { + assistant_output.output_text, + ) + .await + { let mapped_error = map_api_error(err); - warn!("failed to send terminal output: {mapped_error}"); + warn!("failed to send assistant output: {mapped_error}"); let _ = events_tx .send(RealtimeEvent::Error(mapped_error.to_string())) .await; return Err(mapped_error.into()); } - if request_response { - response_create_queue - .request_create(writer, events_tx, "terminal output") - .await - } else { - Ok(()) - } + Ok(()) } async fn handle_realtime_server_event( diff --git a/codex-rs/core/src/realtime_conversation_tests.rs b/codex-rs/core/src/realtime_conversation_tests.rs index 7d815c92ec..c70cfc267f 100644 --- a/codex-rs/core/src/realtime_conversation_tests.rs +++ b/codex-rs/core/src/realtime_conversation_tests.rs @@ -1,6 +1,5 @@ +use super::RealtimeAssistantOutput; use super::RealtimeHandoffState; -use super::RealtimeSessionKind; -use super::RealtimeTerminalOutput; use super::realtime_delegation_from_handoff; use super::realtime_request_headers; use super::realtime_text_from_handoff_request; @@ -127,71 +126,66 @@ fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() { } #[tokio::test] -async fn terminal_output_consumes_only_its_handoff() { - for (session_kind, expected_output_text) in [ - (RealtimeSessionKind::V1, "finished"), - (RealtimeSessionKind::V2, "[BACKEND] finished"), - ] { - let (tx, _rx) = bounded(1); - let state = RealtimeHandoffState::new(tx, session_kind); +async fn assistant_outputs_preserve_active_handoff_until_turn_completion() { + let (tx, _rx) = bounded(1); + let state = RealtimeHandoffState::new(tx); + *state.active_handoff.lock().await = Some("handoff_1".to_string()); - *state.active_handoff.lock().await = Some("handoff_1".to_string()); - let first_output = state - .take_terminal_output(Some("finished".to_string())) - .await; - *state.active_handoff.lock().await = Some("handoff_2".to_string()); - - assert_eq!( - first_output, - Some(RealtimeTerminalOutput::Handoff { - handoff_id: "handoff_1".to_string(), - output_text: expected_output_text.to_string(), - }) - ); - assert_eq!( - state.active_handoff.lock().await.clone(), - Some("handoff_2".to_string()) - ); - assert_eq!( - state - .take_terminal_output(Some("finished again".to_string())) - .await, - Some(RealtimeTerminalOutput::Handoff { - handoff_id: "handoff_2".to_string(), - output_text: format!("{expected_output_text} again"), - }) - ); - } + assert_eq!( + state.assistant_output("working".to_string()).await, + RealtimeAssistantOutput { + handoff_id: Some("handoff_1".to_string()), + output_text: "working".to_string(), + } + ); + assert_eq!( + state.active_handoff.lock().await.as_deref(), + Some("handoff_1") + ); + assert_eq!( + state.assistant_output("finished".to_string()).await, + RealtimeAssistantOutput { + handoff_id: Some("handoff_1".to_string()), + output_text: "finished".to_string(), + } + ); + assert_eq!( + state.active_handoff.lock().await.as_deref(), + Some("handoff_1") + ); + state.finish_turn().await; + assert_eq!(state.active_handoff.lock().await.as_deref(), None); } #[tokio::test] -async fn terminal_output_without_handoff_stays_plain() { - for session_kind in [RealtimeSessionKind::V1, RealtimeSessionKind::V2] { - let (tx, _rx) = bounded(1); - let state = RealtimeHandoffState::new(tx, session_kind); +async fn assistant_output_without_handoff_has_no_active_id() { + let (tx, _rx) = bounded(1); + let state = RealtimeHandoffState::new(tx); - assert_eq!( - state - .take_terminal_output(Some("finished".to_string())) - .await, - Some(RealtimeTerminalOutput::Direct { - output_text: "finished".to_string(), - }) - ); - } + assert_eq!( + state.assistant_output("working".to_string()).await, + RealtimeAssistantOutput { + handoff_id: None, + output_text: "working".to_string(), + } + ); + assert_eq!( + state.assistant_output("finished".to_string()).await, + RealtimeAssistantOutput { + handoff_id: None, + output_text: "finished".to_string(), + } + ); } #[tokio::test] -async fn terminal_output_without_message_still_consumes_handoff() { - for session_kind in [RealtimeSessionKind::V1, RealtimeSessionKind::V2] { - let (tx, _rx) = bounded(1); - let state = RealtimeHandoffState::new(tx, session_kind); +async fn finishing_turn_consumes_handoff() { + let (tx, _rx) = bounded(1); + let state = RealtimeHandoffState::new(tx); + *state.active_handoff.lock().await = Some("handoff_1".to_string()); - *state.active_handoff.lock().await = Some("handoff_1".to_string()); - - assert_eq!(state.take_terminal_output(/*output_text*/ None).await, None); - assert_eq!(state.active_handoff.lock().await.clone(), None); - } + state.finish_turn().await; + assert_eq!(state.active_handoff.lock().await.as_deref(), None); } #[test] diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 472011a2f8..4bf74b7739 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -1672,7 +1672,7 @@ impl Session { id: turn_context.sub_id.clone(), msg, }; - self.maybe_send_realtime_terminal_output(&legacy_source) + self.maybe_send_realtime_assistant_output(&legacy_source) .await; self.send_event_raw(event).await; self.maybe_notify_parent_of_terminal_turn(turn_context, &legacy_source) @@ -1782,16 +1782,32 @@ impl Session { } } - async fn maybe_send_realtime_terminal_output(&self, msg: &EventMsg) { - let EventMsg::TurnComplete(event) = msg else { - return; + async fn maybe_send_realtime_assistant_output(&self, msg: &EventMsg) { + let result = match msg { + EventMsg::ItemCompleted(ItemCompletedEvent { + item: TurnItem::AgentMessage(message), + .. + }) => { + let output_text = message + .content + .iter() + .map(|content| match content { + codex_protocol::items::AgentMessageContent::Text { text } => text.as_str(), + }) + .collect::(); + if output_text.trim().is_empty() { + return; + } + self.conversation.send_assistant_output(output_text).await + } + EventMsg::TurnComplete(_) | EventMsg::TurnAborted(_) => { + self.conversation.finish_turn().await; + return; + } + _ => return, }; - if let Err(err) = self - .conversation - .finish_turn(event.last_agent_message.clone()) - .await - { - debug!("failed to send terminal output to realtime conversation: {err}"); + if let Err(err) = result { + debug!("failed to send assistant output to realtime conversation: {err}"); } } diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index 3e95494c53..bdf1a12005 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/codex-rs/core/tests/suite/realtime_conversation.rs @@ -2391,7 +2391,7 @@ async fn conversation_sends_terminal_assistant_message_to_realtime_handoff() -> } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn conversation_sends_only_last_assistant_message_at_turn_complete() -> Result<()> { +async fn conversation_sends_each_assistant_message_once() -> Result<()> { skip_if_no_network!(Ok(())); let (gate_second_message_tx, gate_second_message_rx) = oneshot::channel(); @@ -2439,6 +2439,7 @@ async fn conversation_sends_only_last_assistant_message_at_turn_complete() -> Re }), ], vec![], + vec![], ]]) .await; @@ -2481,14 +2482,16 @@ async fn conversation_sends_only_last_assistant_message_at_turn_complete() -> Re }) .await; - let intermediate_output = timeout( - Duration::from_millis(200), - realtime_server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 1), - ) - .await; - assert!( - intermediate_output.is_err(), - "assistant items should not be sent before turn completion" + let first_output = realtime_server + .wait_for_request(/*connection_index*/ 0, /*request_index*/ 1) + .await; + assert_eq!( + first_output.body_json(), + json!({ + "type": "conversation.handoff.append", + "handoff_id": "handoff_item_done", + "output_text": "assistant message 1" + }) ); let _ = gate_second_message_tx.send(()); @@ -2505,11 +2508,11 @@ async fn conversation_sends_only_last_assistant_message_at_turn_complete() -> Re }) .await; - let terminal_output = realtime_server - .wait_for_request(/*connection_index*/ 0, /*request_index*/ 1) + let second_output = realtime_server + .wait_for_request(/*connection_index*/ 0, /*request_index*/ 2) .await; assert_eq!( - terminal_output.body_json(), + second_output.body_json(), json!({ "type": "conversation.handoff.append", "handoff_id": "handoff_item_done", @@ -2518,12 +2521,133 @@ async fn conversation_sends_only_last_assistant_message_at_turn_complete() -> Re ); let duplicate_output = timeout( Duration::from_millis(200), - realtime_server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 2), + realtime_server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 3), ) .await; assert!( duplicate_output.is_err(), - "turn completion should emit exactly one terminal output" + "each assistant message should be emitted exactly once" + ); + + realtime_server.shutdown().await; + api_server.shutdown().await; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn conversation_clears_handoff_after_turn_abort() -> Result<()> { + skip_if_no_network!(Ok(())); + + let (gate_first_turn_tx, gate_first_turn_rx) = oneshot::channel(); + let first_chunks = vec![ + StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_response_created("resp-1")), + }, + StreamingSseChunk { + gate: Some(gate_first_turn_rx), + body: sse_event(responses::ev_completed("resp-1")), + }, + ]; + let second_chunks = vec![ + StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_response_created("resp-2")), + }, + StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_assistant_message( + "msg-2", + "assistant after abort", + )), + }, + StreamingSseChunk { + gate: None, + body: sse_event(responses::ev_completed("resp-2")), + }, + ]; + let (api_server, _completions) = + start_streaming_sse_server(vec![first_chunks, second_chunks]).await; + + let realtime_server = start_websocket_server(vec![vec![ + vec![ + json!({ + "type": "session.updated", + "session": { "id": "sess_abort", "instructions": "backend prompt" } + }), + json!({ + "type": "conversation.input_transcript.delta", + "delta": "delegate then abort" + }), + json!({ + "type": "conversation.handoff.requested", + "handoff_id": "handoff_abort", + "item_id": "item_abort", + "input_transcript": "delegate then abort" + }), + ], + 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); + config.realtime.version = RealtimeWsVersion::V1; + } + }); + let test = builder.build_with_streaming_server(&api_server).await?; + + test.codex + .submit(Op::RealtimeConversationStart(ConversationStartParams { + output_modality: RealtimeOutputModality::Audio, + prompt: Some(Some("backend prompt".to_string())), + realtime_session_id: None, + transport: None, + voice: None, + })) + .await?; + + let _ = wait_for_event_match(&test.codex, |msg| match msg { + EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent { + payload: RealtimeEvent::HandoffRequested(handoff), + }) if handoff.handoff_id == "handoff_abort" => Some(()), + _ => None, + }) + .await; + api_server.wait_for_request_count(1).await; + + test.codex.submit(Op::Interrupt).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnAborted(_)) + }) + .await; + let _ = gate_first_turn_tx.send(()); + + test.codex + .submit(Op::UserInput { + items: vec![UserInput::Text { + text: "continue after abort".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + + let output = realtime_server + .wait_for_request(/*connection_index*/ 0, /*request_index*/ 1) + .await; + assert_eq!( + output.body_json(), + json!({ + "type": "conversation.handoff.append", + "handoff_id": "codex", + "output_text": "assistant after abort" + }) ); realtime_server.shutdown().await;