openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-01 19:33:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Route terminal Codex output to realtime

Browse Source
This commit is contained in:
Guinness Chen
2026-06-09 09:03:20 -07:00
parent 3c85d78cc9
commit a678a86cb4
14 changed files with 435 additions and 390 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server/tests/suite/v2/realtime_conversation.rs
+121 -11
View File
@@ -36,7 +36,10 @@ use codex_app_server_protocol::ThreadRealtimeTranscriptDoneNotification;
use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse;
use codex_app_server_protocol::TurnCompletedNotification;
use codex_app_server_protocol::TurnStartParams;
use codex_app_server_protocol::TurnStartResponse;
use codex_app_server_protocol::TurnStartedNotification;
use codex_app_server_protocol::UserInput as V2UserInput;
use codex_features::FEATURES;
use codex_features::Feature;
use codex_protocol::protocol::RealtimeConversationVersion;
@@ -1320,7 +1323,7 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()>
json!({
"type": "conversation.handoff.append",
"handoff_id": "handoff_v1",
"output_text": "\"Agent Final Message\":\n\ndelegated from v1",
"output_text": "delegated from v1",
})
);
@@ -1328,6 +1331,113 @@ async fn webrtc_v1_handoff_request_delegates_and_appends_result() -> Result<()>
Ok(())
}
#[tokio::test]
async fn webrtc_terminal_output_without_handoff_reaches_realtime() -> Result<()> {
skip_if_no_network!(Ok(()));
for (version, output_texts) in [
(
RealtimeTestVersion::V1,
[
"first direct result from v1",
"second direct result from v1",
],
),
(
RealtimeTestVersion::V2,
[
"first direct result from v2",
"second direct result from v2",
],
),
] {
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])?,
]),
realtime_sideband(vec![realtime_sideband_connection(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![],
])]),
)
.await?;
let started = harness.start_webrtc_realtime("v=offer\r\n").await?;
assert_eq!(
started.started.version,
match version {
RealtimeTestVersion::V1 => RealtimeConversationVersion::V1,
RealtimeTestVersion::V2 => RealtimeConversationVersion::V2,
}
);
for (turn_index, output_text) in output_texts.into_iter().enumerate() {
let turn_request_id = harness
.mcp
.send_turn_start_request(TurnStartParams {
thread_id: harness.thread_id.clone(),
input: vec![V2UserInput::Text {
text: format!("direct text turn {turn_index}"),
text_elements: Vec::new(),
}],
..Default::default()
})
.await?;
let turn_response: JSONRPCResponse = timeout(
DEFAULT_TIMEOUT,
harness
.mcp
.read_stream_until_response_message(RequestId::Integer(turn_request_id)),
)
.await??;
let _: TurnStartResponse = to_response(turn_response)?;
let _ = harness
.read_notification::<TurnCompletedNotification>("turn/completed")
.await?;
assert_eq!(
harness
.sideband_outbound_request(/*request_index*/ 1 + turn_index * 2)
.await,
json!({
"type": "conversation.item.create",
"item": {
"type": "message",
"role": "developer",
"content": [{
"type": "input_text",
"text": output_text,
}],
},
})
);
assert_v2_response_create(
&harness
.sideband_outbound_request(/*request_index*/ 2 + turn_index * 2)
.await,
);
}
harness.shutdown().await;
}
Ok(())
}
#[tokio::test]
async fn webrtc_v2_forwards_audio_and_text_between_client_and_sideband() -> Result<()> {
skip_if_no_network!(Ok(()));
@@ -1628,8 +1738,8 @@ async fn webrtc_v2_background_agent_tool_call_delegates_and_returns_function_out
requests[0]
);
let progress = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_v2_progress_update(&progress, "delegated from v2");
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);
@@ -1736,7 +1846,7 @@ async fn webrtc_v2_background_agent_steering_ack_requests_response_create() -> R
}
#[tokio::test]
async fn webrtc_v2_background_agent_progress_is_sent_before_function_output() -> Result<()> {
async fn webrtc_v2_terminal_output_is_sent_before_function_output() -> Result<()> {
skip_if_no_network!(Ok(()));
let mut harness = RealtimeE2eHarness::new(
@@ -1763,8 +1873,8 @@ async fn webrtc_v2_background_agent_progress_is_sent_before_function_output() ->
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
let progress = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_v2_progress_update(&progress, "progress before final");
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(
@@ -1853,8 +1963,8 @@ async fn webrtc_v2_tool_call_delegated_turn_can_execute_shell_tool() -> Result<(
requests[1]
);
let progress = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_v2_progress_update(&progress, "shell tool finished");
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(
@@ -1937,8 +2047,8 @@ async fn webrtc_v2_tool_call_does_not_block_sideband_audio() -> Result<()> {
.await?;
assert_eq!(turn_completed.thread_id, harness.thread_id);
let progress = harness.sideband_outbound_request(/*request_index*/ 1).await;
assert_v2_progress_update(&progress, "late delegated result");
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(
@@ -2196,7 +2306,7 @@ fn assert_v2_function_call_output(request: &Value, call_id: &str, expected_outpu
);
}
fn assert_v2_progress_update(request: &Value, expected_text: &str) {
fn assert_v2_terminal_output(request: &Value, expected_text: &str) {
assert_eq!(
request,
&json!({
codex-rs/codex-api/src/endpoint/realtime_websocket/methods.rs
+50 -8
View File
@@ -3,6 +3,7 @@ use crate::endpoint::realtime_websocket::methods_common::conversation_item_creat
use crate::endpoint::realtime_websocket::methods_common::normalized_session_mode;
use crate::endpoint::realtime_websocket::methods_common::session_update_session;
use crate::endpoint::realtime_websocket::methods_common::websocket_intent;
use crate::endpoint::realtime_websocket::protocol::ConversationRole;
use crate::endpoint::realtime_websocket::protocol::RealtimeAudioFrame;
use crate::endpoint::realtime_websocket::protocol::RealtimeEvent;
use crate::endpoint::realtime_websocket::protocol::RealtimeEventParser;
@@ -241,6 +242,16 @@ impl RealtimeWebsocketConnection {
.await
}
pub async fn send_conversation_handoff_append(
&self,
handoff_id: Option<String>,
output_text: String,
) -> Result<(), ApiError> {
self.writer
.send_conversation_handoff_append(handoff_id, output_text)
.await
}
pub async fn close(&self) -> Result<(), ApiError> {
self.writer.close().await
}
@@ -287,8 +298,24 @@ impl RealtimeWebsocketWriter {
}
pub async fn send_conversation_item_create(&self, text: String) -> Result<(), ApiError> {
self.send_json(&conversation_item_create_message(self.event_parser, text))
.await
self.send_json(&conversation_item_create_message(
self.event_parser,
ConversationRole::User,
text,
))
.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(
@@ -304,6 +331,20 @@ impl RealtimeWebsocketWriter {
.await
}
pub async fn send_conversation_handoff_append(
&self,
handoff_id: Option<String>,
output_text: String,
) -> Result<(), ApiError> {
self.send_json(
&crate::endpoint::realtime_websocket::methods_v1::conversation_handoff_append_message(
handoff_id,
output_text,
),
)
.await
}
pub async fn send_response_create(&self) -> Result<(), ApiError> {
self.send_json(&RealtimeOutboundMessage::ResponseCreate)
.await
@@ -1611,11 +1652,12 @@ mod tests {
.into_text()
.expect("text");
let fourth_json: Value = serde_json::from_str(&fourth).expect("json");
assert_eq!(fourth_json["type"], "conversation.handoff.append");
assert_eq!(fourth_json["handoff_id"], "handoff_1");
assert_eq!(
fourth_json["output_text"],
"\"Agent Final Message\":\n\nhello from background agent"
fourth_json,
json!({
"type": "conversation.handoff.append",
"output_text": "hello from background agent"
})
);
ws.send(Message::Text(
@@ -1738,8 +1780,8 @@ mod tests {
.await
.expect("send item");
connection
.send_conversation_function_call_output(
"handoff_1".to_string(),
.send_conversation_handoff_append(
/*handoff_id*/ None,
"hello from background agent".to_string(),
)
.await
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_common.rs
+7 -7
View File
@@ -6,6 +6,7 @@ use crate::endpoint::realtime_websocket::methods_v2::conversation_function_call_
use crate::endpoint::realtime_websocket::methods_v2::conversation_item_create_message as v2_conversation_item_create_message;
use crate::endpoint::realtime_websocket::methods_v2::session_update_session as v2_session_update_session;
use crate::endpoint::realtime_websocket::methods_v2::websocket_intent as v2_websocket_intent;
use crate::endpoint::realtime_websocket::protocol::ConversationRole;
use crate::endpoint::realtime_websocket::protocol::RealtimeEventParser;
use crate::endpoint::realtime_websocket::protocol::RealtimeOutboundMessage;
use crate::endpoint::realtime_websocket::protocol::RealtimeOutputModality;
@@ -18,7 +19,6 @@ use serde_json::Value;
use serde_json::to_value;
pub(super) const REALTIME_AUDIO_SAMPLE_RATE: u32 = 24_000;
const AGENT_FINAL_MESSAGE_PREFIX: &str = "\"Agent Final Message\":\n\n";
pub(super) fn normalized_session_mode(
event_parser: RealtimeEventParser,
@@ -32,11 +32,12 @@ pub(super) fn normalized_session_mode(
pub(super) fn conversation_item_create_message(
event_parser: RealtimeEventParser,
role: ConversationRole,
text: String,
) -> RealtimeOutboundMessage {
match event_parser {
RealtimeEventParser::V1 => v1_conversation_item_create_message(text),
RealtimeEventParser::RealtimeV2 => v2_conversation_item_create_message(text),
RealtimeEventParser::V1 => v1_conversation_item_create_message(role, text),
RealtimeEventParser::RealtimeV2 => v2_conversation_item_create_message(role, text),
}
}
@@ -46,10 +47,9 @@ pub(super) fn conversation_function_call_output_message(
output_text: String,
) -> RealtimeOutboundMessage {
match event_parser {
RealtimeEventParser::V1 => v1_conversation_handoff_append_message(
call_id,
format!("{AGENT_FINAL_MESSAGE_PREFIX}{output_text}"),
),
RealtimeEventParser::V1 => {
v1_conversation_handoff_append_message(Some(call_id), output_text)
}
RealtimeEventParser::RealtimeV2 => {
v2_conversation_function_call_output_message(call_id, output_text)
}
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v1.rs
+6 -3
View File
@@ -15,11 +15,14 @@ use crate::endpoint::realtime_websocket::protocol::SessionAudioOutput;
use crate::endpoint::realtime_websocket::protocol::SessionType;
use crate::endpoint::realtime_websocket::protocol::SessionUpdateSession;
pub(super) fn conversation_item_create_message(text: String) -> RealtimeOutboundMessage {
pub(super) fn conversation_item_create_message(
role: ConversationRole,
text: String,
) -> RealtimeOutboundMessage {
RealtimeOutboundMessage::ConversationItemCreate {
item: ConversationItemPayload::Message(ConversationMessageItem {
r#type: ConversationItemType::Message,
role: ConversationRole::User,
role,
content: vec![ConversationItemContent {
r#type: ConversationContentType::InputText,
text,
@@ -29,7 +32,7 @@ pub(super) fn conversation_item_create_message(text: String) -> RealtimeOutbound
}
pub(super) fn conversation_handoff_append_message(
handoff_id: String,
handoff_id: Option<String>,
output_text: String,
) -> RealtimeOutboundMessage {
RealtimeOutboundMessage::ConversationHandoffAppend {
codex-rs/codex-api/src/endpoint/realtime_websocket/methods_v2.rs
+5 -2
View File
@@ -36,11 +36,14 @@ const REALTIME_V2_SILENCE_TOOL_NAME: &str = "remain_silent";
const REALTIME_V2_SILENCE_TOOL_DESCRIPTION: &str = "Call this when the best response is to say nothing. Use it instead of speaking after hidden system/control messages, after background agent updates in silent modes, or whenever acknowledging aloud would be distracting. This tool has no user-visible effect.";
const REALTIME_V2_INPUT_TRANSCRIPTION_MODEL: &str = "gpt-4o-mini-transcribe";
pub(super) fn conversation_item_create_message(text: String) -> RealtimeOutboundMessage {
pub(super) fn conversation_item_create_message(
role: ConversationRole,
text: String,
) -> RealtimeOutboundMessage {
RealtimeOutboundMessage::ConversationItemCreate {
item: ConversationItemPayload::Message(ConversationMessageItem {
r#type: ConversationItemType::Message,
role: ConversationRole::User,
role,
content: vec![ConversationItemContent {
r#type: ConversationContentType::InputText,
text,
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol.rs
+3 -1
View File
@@ -38,7 +38,8 @@ pub(super) enum RealtimeOutboundMessage {
InputAudioBufferAppend { audio: String },
#[serde(rename = "conversation.handoff.append")]
ConversationHandoffAppend {
handoff_id: String,
#[serde(skip_serializing_if = "Option::is_none")]
handoff_id: Option<String>,
output_text: String,
},
#[serde(rename = "response.create")]
@@ -171,6 +172,7 @@ pub(super) enum ConversationItemType {
#[derive(Debug, Clone, Copy, Serialize)]
#[serde(rename_all = "snake_case")]
pub(super) enum ConversationRole {
Developer,
User,
}
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_common.rs
+15
View File
@@ -65,6 +65,21 @@ pub(super) fn parse_transcript_done_event(
.map(|text| RealtimeTranscriptDone { text })
}
pub(super) fn parse_response_event_response_id(parsed: &Value) -> Option<String> {
parsed
.get("response")
.and_then(Value::as_object)
.and_then(|response| response.get("id"))
.and_then(Value::as_str)
.map(str::to_string)
.or_else(|| {
parsed
.get("response_id")
.and_then(Value::as_str)
.map(str::to_string)
})
}
pub(super) fn parse_error_event(parsed: &Value) -> Option<RealtimeEvent> {
parsed
.get("message")
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v1.rs
+15
View File
@@ -1,11 +1,15 @@
use crate::endpoint::realtime_websocket::protocol_common::parse_error_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_realtime_payload;
use crate::endpoint::realtime_websocket::protocol_common::parse_response_event_response_id;
use crate::endpoint::realtime_websocket::protocol_common::parse_session_updated_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_delta_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_done_event;
use codex_protocol::protocol::RealtimeAudioFrame;
use codex_protocol::protocol::RealtimeEvent;
use codex_protocol::protocol::RealtimeHandoffRequested;
use codex_protocol::protocol::RealtimeResponseCancelled;
use codex_protocol::protocol::RealtimeResponseCreated;
use codex_protocol::protocol::RealtimeResponseDone;
use serde_json::Value;
use tracing::debug;
@@ -62,6 +66,17 @@ pub(super) fn parse_realtime_event_v1(payload: &str) -> Option<RealtimeEvent> {
.cloned()
.map(RealtimeEvent::ConversationItemAdded),
"conversation.item.done" => parse_conversation_item_done_event(&parsed),
"response.created" => Some(RealtimeEvent::ResponseCreated(RealtimeResponseCreated {
response_id: parse_response_event_response_id(&parsed),
})),
"response.cancelled" => Some(RealtimeEvent::ResponseCancelled(
RealtimeResponseCancelled {
response_id: parse_response_event_response_id(&parsed),
},
)),
"response.done" => Some(RealtimeEvent::ResponseDone(RealtimeResponseDone {
response_id: parse_response_event_response_id(&parsed),
})),
"conversation.handoff.requested" => {
let handoff_id = parsed
.get("handoff_id")
codex-rs/codex-api/src/endpoint/realtime_websocket/protocol_v2.rs
+1 -15
View File
@@ -1,5 +1,6 @@
use crate::endpoint::realtime_websocket::protocol_common::parse_error_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_realtime_payload;
use crate::endpoint::realtime_websocket::protocol_common::parse_response_event_response_id;
use crate::endpoint::realtime_websocket::protocol_common::parse_session_updated_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_delta_event;
use crate::endpoint::realtime_websocket::protocol_common::parse_transcript_done_event;
@@ -78,21 +79,6 @@ pub(super) fn parse_realtime_event_v2(payload: &str) -> Option<RealtimeEvent> {
}
}
fn parse_response_event_response_id(parsed: &Value) -> Option<String> {
parsed
.get("response")
.and_then(Value::as_object)
.and_then(|response| response.get("id"))
.and_then(Value::as_str)
.map(str::to_string)
.or_else(|| {
parsed
.get("response_id")
.and_then(Value::as_str)
.map(str::to_string)
})
}
fn parse_output_audio_delta_event(parsed: &Value) -> Option<RealtimeEvent> {
let data = parsed
.get("delta")
codex-rs/core/src/realtime_conversation.rs
+91 -169
View File
@@ -61,7 +61,7 @@ use tracing::warn;
const AUDIO_IN_QUEUE_CAPACITY: usize = 256;
const USER_TEXT_IN_QUEUE_CAPACITY: usize = 64;
const HANDOFF_OUT_QUEUE_CAPACITY: usize = 64;
const TERMINAL_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";
@@ -98,19 +98,17 @@ enum RealtimeSessionKind {
#[derive(Clone, Debug)]
struct RealtimeHandoffState {
output_tx: Sender<HandoffOutput>,
terminal_output_tx: Sender<RealtimeTerminalOutput>,
active_handoff: Arc<Mutex<Option<String>>>,
last_output_text: Arc<Mutex<Option<String>>>,
session_kind: RealtimeSessionKind,
}
#[derive(Debug, PartialEq, Eq)]
enum HandoffOutput {
ProgressUpdate {
handoff_id: String,
enum RealtimeTerminalOutput {
Direct {
output_text: String,
},
FinalUpdate {
Handoff {
handoff_id: String,
output_text: String,
},
@@ -188,7 +186,7 @@ struct RealtimeInputTask {
writer: RealtimeWebsocketWriter,
events: RealtimeWebsocketEvents,
user_text_rx: Receiver<String>,
handoff_output_rx: Receiver<HandoffOutput>,
terminal_output_rx: Receiver<RealtimeTerminalOutput>,
audio_rx: Receiver<RealtimeAudioFrame>,
events_tx: Sender<RealtimeEvent>,
handoff_state: RealtimeHandoffState,
@@ -198,19 +196,40 @@ struct RealtimeInputTask {
struct RealtimeInputChannels {
user_text_rx: Receiver<String>,
handoff_output_rx: Receiver<HandoffOutput>,
terminal_output_rx: Receiver<RealtimeTerminalOutput>,
audio_rx: Receiver<RealtimeAudioFrame>,
}
impl RealtimeHandoffState {
fn new(output_tx: Sender<HandoffOutput>, session_kind: RealtimeSessionKind) -> Self {
fn new(
terminal_output_tx: Sender<RealtimeTerminalOutput>,
session_kind: RealtimeSessionKind,
) -> Self {
Self {
output_tx,
terminal_output_tx,
active_handoff: Arc::new(Mutex::new(None)),
last_output_text: 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 }),
}
}
}
#[allow(dead_code)]
@@ -293,16 +312,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 (handoff_output_tx, handoff_output_rx) =
async_channel::bounded::<HandoffOutput>(HANDOFF_OUT_QUEUE_CAPACITY);
let (terminal_output_tx, terminal_output_rx) =
async_channel::bounded::<RealtimeTerminalOutput>(TERMINAL_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(handoff_output_tx, session_kind);
let handoff = RealtimeHandoffState::new(terminal_output_tx, session_kind);
let input_channels = RealtimeInputChannels {
user_text_rx,
handoff_output_rx,
terminal_output_rx,
audio_rx,
};
@@ -341,7 +360,7 @@ impl RealtimeConversationManager {
writer: connection.writer(),
events: connection.events(),
user_text_rx: input_channels.user_text_rx,
handoff_output_rx: input_channels.handoff_output_rx,
terminal_output_rx: input_channels.terminal_output_rx,
audio_rx: input_channels.audio_rx,
events_tx,
handoff_state: handoff.clone(),
@@ -449,39 +468,7 @@ impl RealtimeConversationManager {
Ok(())
}
pub(crate) async fn handoff_out(&self, output_text: String) -> CodexResult<()> {
let handoff = {
let guard = self.state.lock().await;
let Some(state) = guard.as_ref() else {
return Err(CodexErr::InvalidRequest(
"conversation is not running".to_string(),
));
};
state.handoff.clone()
};
let Some(handoff_id) = handoff.active_handoff.lock().await.clone() else {
return Ok(());
};
let output_text = prefix_realtime_text(
output_text,
REALTIME_BACKEND_TEXT_PREFIX,
handoff.session_kind,
);
*handoff.last_output_text.lock().await = Some(output_text.clone());
handoff
.output_tx
.send(HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
})
.await
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
Ok(())
}
pub(crate) async fn handoff_complete(&self) -> CodexResult<()> {
pub(crate) async fn finish_turn(&self, output_text: Option<String>) -> CodexResult<()> {
let handoff = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.handoff.clone())
@@ -489,45 +476,16 @@ impl RealtimeConversationManager {
let Some(handoff) = handoff else {
return Ok(());
};
match handoff.session_kind {
RealtimeSessionKind::V1 => return Ok(()),
RealtimeSessionKind::V2 => {}
}
let Some(handoff_id) = handoff.active_handoff.lock().await.clone() else {
let Some(terminal_output) = handoff.take_terminal_output(output_text).await else {
return Ok(());
};
let Some(output_text) = handoff.last_output_text.lock().await.clone() else {
return Ok(());
};
handoff
.output_tx
.send(HandoffOutput::FinalUpdate {
handoff_id,
output_text,
})
.terminal_output_tx
.send(terminal_output)
.await
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))
}
pub(crate) async fn active_handoff_id(&self) -> Option<String> {
let handoff = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.handoff.clone())
}?;
handoff.active_handoff.lock().await.clone()
}
pub(crate) async fn clear_active_handoff(&self) {
let handoff = {
let guard = self.state.lock().await;
guard.as_ref().map(|state| state.handoff.clone())
};
if let Some(handoff) = handoff {
*handoff.active_handoff.lock().await = None;
*handoff.last_output_text.lock().await = None;
}
.map_err(|_| CodexErr::InvalidRequest("conversation is not running".to_string()))?;
Ok(())
}
pub(crate) async fn shutdown(&self) -> CodexResult<()> {
@@ -1082,7 +1040,7 @@ fn spawn_webrtc_sideband_input_task(input: RealtimeWebrtcSidebandInputTask) -> J
writer: connection.writer(),
events: connection.events(),
user_text_rx: input_channels.user_text_rx,
handoff_output_rx: input_channels.handoff_output_rx,
terminal_output_rx: input_channels.terminal_output_rx,
audio_rx: input_channels.audio_rx,
events_tx,
handoff_state,
@@ -1098,7 +1056,7 @@ async fn run_realtime_input_task(input: RealtimeInputTask) {
writer,
events,
user_text_rx,
handoff_output_rx,
terminal_output_rx,
audio_rx,
events_tx,
handoff_state,
@@ -1120,13 +1078,12 @@ async fn run_realtime_input_task(input: RealtimeInputTask) {
)
.await
}
// Background agent progress or final output that should be sent back to realtime.
background_agent_output = handoff_output_rx.recv() => {
handle_handoff_output(
background_agent_output,
// Terminal Codex output that should be sent back to realtime.
terminal_output = terminal_output_rx.recv() => {
handle_terminal_output(
terminal_output,
&writer,
&events_tx,
&handoff_state,
event_parser,
&mut response_create_queue,
)
@@ -1175,75 +1132,62 @@ async fn handle_user_text_input(
Ok(())
}
async fn handle_handoff_output(
handoff_output: Result<HandoffOutput, RecvError>,
async fn handle_terminal_output(
terminal_output: Result<RealtimeTerminalOutput, RecvError>,
writer: &RealtimeWebsocketWriter,
events_tx: &Sender<RealtimeEvent>,
handoff_state: &RealtimeHandoffState,
event_parser: RealtimeEventParser,
response_create_queue: &mut RealtimeResponseCreateQueue,
) -> anyhow::Result<()> {
let handoff_output = handoff_output.context("handoff output channel closed")?;
let terminal_output = terminal_output.context("terminal output channel closed")?;
let result = match event_parser {
RealtimeEventParser::V1 => match handoff_output {
HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
}
| HandoffOutput::FinalUpdate {
handoff_id,
output_text,
} => {
let (result, request_response) = match terminal_output {
RealtimeTerminalOutput::Direct { output_text } => (
writer
.send_conversation_developer_item_create(output_text)
.await,
true,
),
RealtimeTerminalOutput::Handoff {
handoff_id,
output_text,
} => match event_parser {
RealtimeEventParser::V1 => (
writer
.send_conversation_function_call_output(handoff_id, output_text)
.await
}
},
RealtimeEventParser::RealtimeV2 => match handoff_output {
HandoffOutput::ProgressUpdate {
handoff_id,
output_text,
} => {
let active_handoff = handoff_state.active_handoff.lock().await.clone();
match active_handoff {
Some(active_handoff) if active_handoff == handoff_id => {}
Some(_) | None => {
debug!("dropping stale realtime handoff progress update");
return Ok(());
}
}
writer.send_conversation_item_create(output_text).await
}
HandoffOutput::FinalUpdate {
handoff_id,
output_text: _,
} => {
if let Err(err) = writer
.send_conversation_function_call_output(
handoff_id,
REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT.to_string(),
)
.await
{
.send_conversation_handoff_append(Some(handoff_id), output_text)
.await,
false,
),
RealtimeEventParser::RealtimeV2 => (
if let Err(err) = writer.send_conversation_item_create(output_text).await {
Err(err)
} else {
return response_create_queue
.request_create(writer, events_tx, "handoff")
.await;
}
}
writer
.send_conversation_function_call_output(
handoff_id,
REALTIME_V2_HANDOFF_COMPLETE_ACKNOWLEDGEMENT.to_string(),
)
.await
},
true,
),
},
};
if let Err(err) = result {
let mapped_error = map_api_error(err);
warn!("failed to send handoff output: {mapped_error}");
warn!("failed to send terminal output: {mapped_error}");
let _ = events_tx
.send(RealtimeEvent::Error(mapped_error.to_string()))
.await;
return Err(mapped_error.into());
}
Ok(())
if request_response {
response_create_queue
.request_create(writer, events_tx, "terminal output")
.await
} else {
Ok(())
}
}
async fn handle_realtime_server_event(
@@ -1311,34 +1255,14 @@ async fn handle_realtime_server_event(
false
}
RealtimeEvent::ResponseCreated(_) => {
match session_kind {
RealtimeSessionKind::V1 => {}
RealtimeSessionKind::V2 => response_create_queue.mark_started(),
}
response_create_queue.mark_started();
false
}
RealtimeEvent::ResponseCancelled(_) => {
RealtimeEvent::ResponseCancelled(_) | RealtimeEvent::ResponseDone(_) => {
*output_audio_state = None;
match session_kind {
RealtimeSessionKind::V1 => {}
RealtimeSessionKind::V2 => {
response_create_queue
.mark_finished(writer, events_tx, "deferred")
.await?;
}
}
false
}
RealtimeEvent::ResponseDone(_) => {
*output_audio_state = None;
match session_kind {
RealtimeSessionKind::V1 => {}
RealtimeSessionKind::V2 => {
response_create_queue
.mark_finished(writer, events_tx, "deferred")
.await?;
}
}
response_create_queue
.mark_finished(writer, events_tx, "deferred")
.await?;
false
}
RealtimeEvent::HandoffRequested(handoff) => {
@@ -1346,7 +1270,6 @@ async fn handle_realtime_server_event(
match session_kind {
RealtimeSessionKind::V1 => {
*handoff_state.last_output_text.lock().await = None;
*handoff_state.active_handoff.lock().await = Some(handoff.handoff_id.clone());
}
RealtimeSessionKind::V2 => {
@@ -1374,7 +1297,6 @@ async fn handle_realtime_server_event(
.await?;
}
None => {
*handoff_state.last_output_text.lock().await = None;
*handoff_state.active_handoff.lock().await =
Some(handoff.handoff_id.clone());
}
codex-rs/core/src/realtime_conversation_tests.rs
+64 -10
View File
@@ -1,5 +1,6 @@
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;
@@ -126,18 +127,71 @@ fn wraps_realtime_delegation_input_with_xml_escaping_without_transcript() {
}
#[tokio::test]
async fn clears_active_handoff_explicitly() {
let (tx, _rx) = bounded(1);
let state = RealtimeHandoffState::new(tx, RealtimeSessionKind::V1);
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);
*state.active_handoff.lock().await = Some("handoff_1".to_string());
assert_eq!(
state.active_handoff.lock().await.clone(),
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());
*state.active_handoff.lock().await = None;
assert_eq!(state.active_handoff.lock().await.clone(), None);
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"),
})
);
}
}
#[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);
assert_eq!(
state
.take_terminal_output(Some("finished".to_string()))
.await,
Some(RealtimeTerminalOutput::Direct {
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);
*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);
}
}
#[test]
codex-rs/core/src/session/mod.rs
+9 -23
View File
@@ -222,7 +222,6 @@ pub(crate) use self::session::SessionSettingsUpdate;
use self::turn::AssistantMessageStreamParsers;
#[cfg(test)]
use self::turn::collect_explicit_app_ids_from_skill_items;
use self::turn::realtime_text_for_event;
use self::turn_context::TurnContext;
use self::turn_context::TurnSkillsContext;
#[cfg(test)]
@@ -1673,13 +1672,11 @@ impl Session {
id: turn_context.sub_id.clone(),
msg,
};
self.maybe_send_realtime_terminal_output(&legacy_source)
.await;
self.send_event_raw(event).await;
self.maybe_notify_parent_of_terminal_turn(turn_context, &legacy_source)
.await;
self.maybe_mirror_event_text_to_realtime(&legacy_source)
.await;
self.maybe_clear_realtime_handoff_for_event(&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) {
@@ -1785,28 +1782,17 @@ impl Session {
}
}
async fn maybe_mirror_event_text_to_realtime(&self, msg: &EventMsg) {
let Some(text) = realtime_text_for_event(msg) else {
async fn maybe_send_realtime_terminal_output(&self, msg: &EventMsg) {
let EventMsg::TurnComplete(event) = msg else {
return;
};
if self.conversation.running_state().await.is_none()
|| self.conversation.active_handoff_id().await.is_none()
if let Err(err) = self
.conversation
.finish_turn(event.last_agent_message.clone())
.await
{
return;
debug!("failed to send terminal output to realtime conversation: {err}");
}
if let Err(err) = self.conversation.handoff_out(text).await {
debug!("failed to mirror event text to realtime conversation: {err}");
}
}
async fn maybe_clear_realtime_handoff_for_event(&self, msg: &EventMsg) {
if !matches!(msg, EventMsg::TurnComplete(_)) {
return;
}
if let Err(err) = self.conversation.handoff_complete().await {
debug!("failed to finalize realtime handoff output: {err}");
}
self.conversation.clear_active_handoff().await;
}
pub(crate) async fn send_event_raw(&self, event: Event) {
codex-rs/core/src/session/turn.rs
-83
View File
@@ -1352,89 +1352,6 @@ fn agent_message_text(item: &codex_protocol::items::AgentMessageItem) -> String
.collect()
}
pub(super) fn realtime_text_for_event(msg: &EventMsg) -> Option<String> {
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::Error(_)
| EventMsg::Warning(_)
| EventMsg::GuardianWarning(_)
| EventMsg::RealtimeConversationStarted(_)
| EventMsg::RealtimeConversationSdp(_)
| EventMsg::RealtimeConversationRealtime(_)
| EventMsg::RealtimeConversationClosed(_)
| EventMsg::ModelReroute(_)
| EventMsg::ModelVerification(_)
| EventMsg::TurnModerationMetadata(_)
| EventMsg::ContextCompacted(_)
| EventMsg::ThreadRolledBack(_)
| EventMsg::TurnStarted(_)
| EventMsg::ThreadSettingsApplied(_)
| EventMsg::TurnComplete(_)
| EventMsg::TokenCount(_)
| EventMsg::UserMessage(_)
| EventMsg::AgentReasoning(_)
| EventMsg::AgentReasoningRawContent(_)
| EventMsg::AgentReasoningSectionBreak(_)
| EventMsg::SessionConfigured(_)
| EventMsg::ThreadGoalUpdated(_)
| EventMsg::McpStartupUpdate(_)
| EventMsg::McpStartupComplete(_)
| EventMsg::McpToolCallBegin(_)
| EventMsg::McpToolCallEnd(_)
| EventMsg::WebSearchBegin(_)
| EventMsg::WebSearchEnd(_)
| EventMsg::ExecCommandBegin(_)
| EventMsg::ExecCommandOutputDelta(_)
| EventMsg::TerminalInteraction(_)
| EventMsg::ExecCommandEnd(_)
| EventMsg::PatchApplyBegin(_)
| EventMsg::PatchApplyUpdated(_)
| EventMsg::PatchApplyEnd(_)
| EventMsg::ImageGenerationBegin(_)
| EventMsg::ImageGenerationEnd(_)
| EventMsg::ViewImageToolCall(_)
| EventMsg::ExecApprovalRequest(_)
| EventMsg::RequestPermissions(_)
| EventMsg::RequestUserInput(_)
| EventMsg::DynamicToolCallRequest(_)
| EventMsg::DynamicToolCallResponse(_)
| EventMsg::GuardianAssessment(_)
| EventMsg::ElicitationRequest(_)
| EventMsg::ApplyPatchApprovalRequest(_)
| EventMsg::DeprecationNotice(_)
| EventMsg::StreamError(_)
| EventMsg::TurnDiff(_)
| EventMsg::RealtimeConversationListVoicesResponse(_)
| EventMsg::PlanUpdate(_)
| EventMsg::TurnAborted(_)
| EventMsg::ShutdownComplete
| EventMsg::EnteredReviewMode(_)
| EventMsg::ExitedReviewMode(_)
| EventMsg::RawResponseItem(_)
| EventMsg::ItemStarted(_)
| EventMsg::HookStarted(_)
| EventMsg::HookCompleted(_)
| EventMsg::AgentMessageContentDelta(_)
| EventMsg::PlanDelta(_)
| EventMsg::ReasoningContentDelta(_)
| EventMsg::ReasoningRawContentDelta(_)
| EventMsg::CollabAgentSpawnBegin(_)
| EventMsg::CollabAgentSpawnEnd(_)
| EventMsg::CollabAgentInteractionBegin(_)
| EventMsg::CollabAgentInteractionEnd(_)
| EventMsg::CollabWaitingBegin(_)
| EventMsg::CollabWaitingEnd(_)
| EventMsg::CollabCloseBegin(_)
| EventMsg::CollabCloseEnd(_)
| EventMsg::CollabResumeBegin(_)
| EventMsg::CollabResumeEnd(_) => None,
}
}
/// Split the stream into normal assistant text vs. proposed plan content.
/// Normal text becomes AgentMessage deltas; plan content becomes PlanDelta +
/// TurnItem::Plan.
codex-rs/core/tests/suite/realtime_conversation.rs
+48 -58
View File
@@ -2117,6 +2117,8 @@ async fn conversation_user_text_turn_is_not_sent_to_realtime() -> Result<()> {
"session": { "id": "sess_user_text", "instructions": "backend prompt" }
})],
vec![],
vec![],
vec![],
]])
.await;
@@ -2275,7 +2277,7 @@ async fn realtime_v2_noop_tool_call_returns_empty_function_output_without_respon
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Result<()> {
async fn conversation_sends_terminal_assistant_message_to_realtime_handoff() -> Result<()> {
skip_if_no_network!(Ok(()));
let api_server = start_mock_server().await;
@@ -2381,7 +2383,7 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re
);
assert_eq!(
realtime_connections[0][1].body_json()["output_text"].as_str(),
Some("\"Agent Final Message\":\n\nassistant says hi")
Some("assistant says hi")
);
realtime_server.shutdown().await;
@@ -2389,7 +2391,7 @@ async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Re
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_handoff_persists_across_item_done_until_turn_complete() -> Result<()> {
async fn conversation_sends_only_last_assistant_message_at_turn_complete() -> Result<()> {
skip_if_no_network!(Ok(()));
let (gate_second_message_tx, gate_second_message_rx) = oneshot::channel();
@@ -2436,10 +2438,6 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() ->
"input_transcript": "delegate now"
}),
],
vec![json!({
"type": "conversation.item.done",
"item": { "id": "item_item_done" }
})],
vec![],
]])
.await;
@@ -2483,48 +2481,18 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() ->
})
.await;
let first_append = realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 1)
.await;
assert_eq!(
first_append.body_json()["type"].as_str(),
Some("conversation.handoff.append")
);
assert_eq!(
first_append.body_json()["handoff_id"].as_str(),
Some("handoff_item_done")
);
assert_eq!(
first_append.body_json()["output_text"].as_str(),
Some("\"Agent Final Message\":\n\nassistant message 1")
);
let _ = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::ConversationItemDone { item_id },
}) if item_id == "item_item_done" => Some(()),
_ => None,
})
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 _ = gate_second_message_tx.send(());
let second_append = realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 2)
.await;
assert_eq!(
second_append.body_json()["type"].as_str(),
Some("conversation.handoff.append")
);
assert_eq!(
second_append.body_json()["handoff_id"].as_str(),
Some("handoff_item_done")
);
assert_eq!(
second_append.body_json()["output_text"].as_str(),
Some("\"Agent Final Message\":\n\nassistant message 2")
);
let completion = completions
.into_iter()
.next()
@@ -2537,6 +2505,27 @@ async fn conversation_handoff_persists_across_item_done_until_turn_complete() ->
})
.await;
let terminal_output = realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 1)
.await;
assert_eq!(
terminal_output.body_json(),
json!({
"type": "conversation.handoff.append",
"handoff_id": "handoff_item_done",
"output_text": "assistant message 2"
})
);
let duplicate_output = timeout(
Duration::from_millis(200),
realtime_server.wait_for_request(/*connection_index*/ 0, /*request_index*/ 2),
)
.await;
assert!(
duplicate_output.is_err(),
"turn completion should emit exactly one terminal output"
);
realtime_server.shutdown().await;
api_server.shutdown().await;
Ok(())
@@ -3054,6 +3043,19 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au
"delegate now"
);
let completion = completions
.into_iter()
.next()
.expect("missing delegated turn completion");
let _ = gate_completed_tx.send(());
completion
.await
.expect("delegated turn request did not complete");
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let mirrored_request = realtime_server
.wait_for_request(/*connection_index*/ 0, /*request_index*/ 1)
.await;
@@ -3075,7 +3077,7 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au
);
assert_eq!(
mirrored_request_body["output_text"].as_str(),
Some("\"Agent Final Message\":\n\nassistant says hi")
Some("assistant says hi")
);
let audio_out = wait_for_event_match(&test.codex, |msg| match msg {
@@ -3094,22 +3096,10 @@ async fn delegated_turn_user_role_echo_does_not_redelegate_and_still_forwards_au
);
assert_eq!(audio_out.data, "AQID");
let completion = completions
.into_iter()
.next()
.expect("missing delegated turn completion");
let _ = gate_completed_tx.send(());
completion
.await
.expect("delegated turn request did not complete");
eprintln!(
"[realtime test +{}ms] delegated completion resolved",
start.elapsed().as_millis()
);
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let requests = api_server.requests().await;
assert_eq!(requests.len(), 1);