openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-24 18:18:41 +00:00
Code Issues Packages Projects Releases Wiki Activity

Mirror user text into realtime

Browse Source
This commit is contained in:
Ahmed Ibrahim
2026-04-11 23:28:13 -07:00
parent 1325bcd3f6
commit c9438f6891
5 changed files with 159 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/codex.rs
+43 -2
View File
@@ -2265,7 +2265,7 @@ impl Session {
}
pub(crate) async fn route_realtime_text_input(self: &Arc<Self>, text: String) {
handlers::user_input_or_turn(
handlers::user_input_or_turn_with_realtime_text_mirror(
self,
self.next_internal_sub_id(),
Op::UserInput {
@@ -2276,6 +2276,7 @@ impl Session {
final_output_json_schema: None,
responsesapi_client_metadata: None,
},
handlers::RealtimeTextMirror::Disabled,
)
.await;
}
@@ -4910,6 +4911,7 @@ mod handlers {
use codex_protocol::config_types::ModeKind;
use codex_protocol::config_types::Settings;
use codex_protocol::dynamic_tools::DynamicToolResponse;
use codex_protocol::items::UserMessageItem;
use codex_protocol::mcp::RequestId as ProtocolRequestId;
use codex_protocol::user_input::UserInput;
use codex_rmcp_client::ElicitationAction;
@@ -4917,6 +4919,7 @@ mod handlers {
use serde_json::Value;
use std::path::PathBuf;
use std::sync::Arc;
use tracing::debug;
use tracing::info;
use tracing::warn;
@@ -4957,7 +4960,22 @@ mod handlers {
}
}
pub(super) enum RealtimeTextMirror {
Enabled,
Disabled,
}
pub async fn user_input_or_turn(sess: &Arc<Session>, sub_id: String, op: Op) {
user_input_or_turn_with_realtime_text_mirror(sess, sub_id, op, RealtimeTextMirror::Enabled)
.await;
}
pub(super) async fn user_input_or_turn_with_realtime_text_mirror(
sess: &Arc<Session>,
sub_id: String,
op: Op,
realtime_text_mirror: RealtimeTextMirror,
) {
let (items, updates, responsesapi_client_metadata) = match op {
Op::UserTurn {
cwd,
@@ -5016,6 +5034,13 @@ mod handlers {
),
_ => unreachable!(),
};
let realtime_text = match realtime_text_mirror {
RealtimeTextMirror::Enabled => {
let text = UserMessageItem::new(&items).message();
(!text.is_empty()).then_some(text)
}
RealtimeTextMirror::Disabled => None,
};
let Ok(current_context) = sess.new_turn_with_sub_id(sub_id.clone(), updates).await else {
// new_turn_with_sub_id already emits the error event.
@@ -5031,7 +5056,10 @@ mod handlers {
)
.await
{
Ok(_) => current_context.session_telemetry.user_prompt(&items),
Ok(_) => {
current_context.session_telemetry.user_prompt(&items);
mirror_user_text_to_realtime(sess, realtime_text).await;
}
Err(SteerInputError::NoActiveTurn(items)) => {
if let Some(responsesapi_client_metadata) = responsesapi_client_metadata {
current_context
@@ -5047,6 +5075,7 @@ mod handlers {
crate::tasks::RegularTask::new(),
)
.await;
mirror_user_text_to_realtime(sess, realtime_text).await;
}
Err(err) => {
sess.send_event_raw(Event {
@@ -5058,6 +5087,18 @@ mod handlers {
}
}
async fn mirror_user_text_to_realtime(sess: &Arc<Session>, realtime_text: Option<String>) {
let Some(text) = realtime_text else {
return;
};
if sess.conversation.running_state().await.is_none() {
return;
}
if let Err(err) = sess.conversation.text_in(text).await {
debug!("failed to mirror user text to realtime conversation: {err}");
}
}
/// Records an inter-agent assistant envelope, then lets the shared pending-work scheduler
/// decide whether an idle session should start a regular turn.
pub async fn inter_agent_communication(
codex-rs/core/tests/suite/realtime_conversation.rs
+107
View File
@@ -1626,6 +1626,113 @@ async fn conversation_startup_context_is_truncated_and_sent_once_per_start() ->
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_user_text_turn_is_sent_to_realtime_when_active() -> Result<()> {
skip_if_no_network!(Ok(()));
let api_server = start_mock_server().await;
let response_mock = responses::mount_sse_once(
&api_server,
responses::sse(vec![
responses::ev_response_created("resp_user_text"),
responses::ev_assistant_message("msg_user_text", "ack"),
responses::ev_completed("resp_user_text"),
]),
)
.await;
let realtime_server = start_websocket_server(vec![vec![
vec![json!({
"type": "session.updated",
"session": { "id": "sess_user_text", "instructions": "backend prompt" }
})],
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(&api_server).await?;
test.codex
.submit(Op::RealtimeConversationStart(ConversationStartParams {
prompt: Some(Some("backend prompt".to_string())),
session_id: None,
transport: None,
voice: None,
}))
.await?;
let session_updated = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload: RealtimeEvent::SessionUpdated { session_id, .. },
}) => Some(session_id.clone()),
_ => None,
})
.await;
assert_eq!(session_updated, "sess_user_text");
let user_text = "typed follow-up for realtime";
test.codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: user_text.to_string(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
})
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
let realtime_text_request = wait_for_matching_websocket_request(
&realtime_server,
"normal user turn text mirrored to realtime",
|request| websocket_request_text(request).as_deref() == Some(user_text),
)
.await;
let model_user_texts = response_mock.single_request().message_input_texts("user");
assert_eq!(
(
model_user_texts.iter().any(|text| text == user_text),
websocket_request_text(&realtime_text_request),
),
(true, Some(user_text.to_string())),
);
let realtime_request_body = realtime_text_request.body_json();
let content = &realtime_request_body["item"]["content"][0];
let snapshot = format!(
"type: {}\nitem.type: {}\nitem.role: {}\ncontent[0].type: {}\ncontent[0].text: {}",
realtime_request_body["type"].as_str().unwrap_or_default(),
realtime_request_body["item"]["type"]
.as_str()
.unwrap_or_default(),
realtime_request_body["item"]["role"]
.as_str()
.unwrap_or_default(),
content["type"].as_str().unwrap_or_default(),
content["text"].as_str().unwrap_or_default(),
);
insta::assert_snapshot!(
"conversation_user_text_turn_is_sent_to_realtime_when_active",
snapshot
);
realtime_server.shutdown().await;
api_server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_mirrors_assistant_message_text_to_realtime_handoff() -> Result<()> {
skip_if_no_network!(Ok(()));
codex-rs/core/tests/suite/snapshots/all__suite__realtime_conversation__conversation_user_text_turn_is_sent_to_realtime_when_active.snap
+9
View File
@@ -0,0 +1,9 @@
---
source: core/tests/suite/realtime_conversation.rs
expression: snapshot
---
type: conversation.item.create
item.type: message
item.role: user
content[0].type: text
content[0].text: typed follow-up for realtime
codex-rs/tui/src/chatwidget.rs
-10
View File
@@ -5050,11 +5050,6 @@ impl ChatWidget {
{
return;
}
let Some(user_message) =
self.maybe_defer_user_message_for_realtime(user_message)
else {
return;
};
let should_submit_now =
self.is_session_configured() && !self.is_plan_streaming_in_tui();
if should_submit_now {
@@ -5085,11 +5080,6 @@ impl ChatWidget {
.bottom_pane
.take_recent_submission_mention_bindings(),
};
let Some(user_message) =
self.maybe_defer_user_message_for_realtime(user_message)
else {
return;
};
self.queue_user_message(user_message);
}
InputResult::Command(cmd) => {
codex-rs/tui/src/chatwidget/realtime.rs
-26
View File
@@ -29,7 +29,6 @@ pub(super) struct RealtimeConversationUiState {
pub(super) phase: RealtimeConversationPhase,
requested_close: bool,
session_id: Option<String>,
warned_audio_only_submission: bool,
transport: RealtimeConversationUiTransport,
#[cfg(not(target_os = "linux"))]
pub(super) meter_placeholder_id: Option<String>,
@@ -191,28 +190,6 @@ impl ChatWidget {
self.last_rendered_user_message_event.as_ref() != Some(&key)
}
pub(super) fn maybe_defer_user_message_for_realtime(
&mut self,
user_message: UserMessage,
) -> Option<UserMessage> {
if !self.realtime_conversation.is_live() {
return Some(user_message);
}
self.restore_user_message_to_composer(user_message);
if !self.realtime_conversation.warned_audio_only_submission {
self.realtime_conversation.warned_audio_only_submission = true;
self.add_info_message(
"Realtime voice mode is audio-only. Use /realtime to stop.".to_string(),
/*hint*/ None,
);
} else {
self.request_redraw();
}
None
}
fn realtime_footer_hint_items() -> Vec<(String, String)> {
vec![("/realtime".to_string(), "stop live voice".to_string())]
}
@@ -238,7 +215,6 @@ impl ChatWidget {
self.realtime_conversation.phase = RealtimeConversationPhase::Starting;
self.realtime_conversation.requested_close = false;
self.realtime_conversation.session_id = None;
self.realtime_conversation.warned_audio_only_submission = false;
self.set_footer_hint_override(Some(Self::realtime_footer_hint_items()));
match self.config.realtime.transport {
RealtimeTransport::Websocket => {
@@ -297,7 +273,6 @@ impl ChatWidget {
self.realtime_conversation.phase = RealtimeConversationPhase::Inactive;
self.realtime_conversation.requested_close = false;
self.realtime_conversation.session_id = None;
self.realtime_conversation.warned_audio_only_submission = false;
self.realtime_conversation.transport = RealtimeConversationUiTransport::Websocket;
}
@@ -320,7 +295,6 @@ impl ChatWidget {
return;
}
self.realtime_conversation.session_id = ev.session_id;
self.realtime_conversation.warned_audio_only_submission = false;
self.set_footer_hint_override(Some(Self::realtime_footer_hint_items()));
if self.realtime_conversation_uses_webrtc() {
self.realtime_conversation.phase = RealtimeConversationPhase::Starting;