openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-09-06 15:29:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

Forward realtime assistant output through handoffs

Browse Source
This commit is contained in:
Guinness Chen
2026-06-09 17:16:24 -07:00
parent f7b6cda8c8
commit 002f9fe228
7 changed files with 381 additions and 354 deletions
Download Patch File Download Diff File Expand all files Collapse all files

248
codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
View File

@@ -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<WebSocketConnectionConfig>) -> 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::<TurnCompletedNotification>("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::<TurnCompletedNotification>("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::<TurnCompletedNotification>("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<String> {
}
}
fn function_call_output_sideband_requests(server: &WebSocketTestServer) -> Vec<Value> {
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,
})
);
}

12
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
View File

@@ -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,

1
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs
View File

@@ -172,7 +172,6 @@ pub(super) enum ConversationItemType {
#[derive(Debug, Clone, Copy, Serialize)]
#[serde(rename_all = "snake_case")]
pub(super) enum ConversationRole {
Developer,
User,
}

176
codex-rs/core/src/realtime_conversation.rs
View File

@@ -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<RealtimeTerminalOutput>,
assistant_output_tx: Sender<RealtimeAssistantOutput>,
active_handoff: Arc<Mutex<Option<String>>>,
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<String>,
output_text: String,
}
#[derive(Debug, PartialEq, Eq)]
@@ -186,50 +178,37 @@ struct RealtimeInputTask {
writer: RealtimeWebsocketWriter,
events: RealtimeWebsocketEvents,
user_text_rx: Receiver<String>,
terminal_output_rx: Receiver<RealtimeTerminalOutput>,
assistant_output_rx: Receiver<RealtimeAssistantOutput>,
audio_rx: Receiver<RealtimeAudioFrame>,
events_tx: Sender<RealtimeEvent>,
handoff_state: RealtimeHandoffState,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
}
struct RealtimeInputChannels {
user_text_rx: Receiver<String>,
terminal_output_rx: Receiver<RealtimeTerminalOutput>,
assistant_output_rx: Receiver<RealtimeAssistantOutput>,
audio_rx: Receiver<RealtimeAudioFrame>,
}
impl RealtimeHandoffState {
fn new(
terminal_output_tx: Sender<RealtimeTerminalOutput>,
session_kind: RealtimeSessionKind,
) -> Self {
fn new(assistant_output_tx: Sender<RealtimeAssistantOutput>) -> 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<String>,
) -> Option<RealtimeTerminalOutput> {
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::<RealtimeAudioFrame>(AUDIO_IN_QUEUE_CAPACITY);
let (user_text_tx, user_text_rx) =
async_channel::bounded::<String>(USER_TEXT_IN_QUEUE_CAPACITY);
let (terminal_output_tx, terminal_output_rx) =
async_channel::bounded::<RealtimeTerminalOutput>(TERMINAL_OUTPUT_QUEUE_CAPACITY);
let (assistant_output_tx, assistant_output_rx) =
async_channel::bounded::<RealtimeAssistantOutput>(ASSISTANT_OUTPUT_QUEUE_CAPACITY);
let (events_tx, events_rx) =
async_channel::bounded::<RealtimeEvent>(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<String>) -> 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<RealtimeEvent>,
handoff_state: RealtimeHandoffState,
session_kind: RealtimeSessionKind,
event_parser: RealtimeEventParser,
realtime_active: Arc<AtomicBool>,
}
@@ -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<OutputAudioState> = 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<RealtimeTerminalOutput, RecvError>,
async fn handle_assistant_output(
assistant_output: Result<RealtimeAssistantOutput, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
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(

110
codex-rs/core/src/realtime_conversation_tests.rs
View File

@@ -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]

36
codex-rs/core/src/session/mod.rs
View File

@@ -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::<String>();
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}");
}
}

152
codex-rs/core/tests/suite/realtime_conversation.rs
View File

@@ -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;