openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-22 17:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Stop mirroring Codex user input to realtime

Browse Source
This commit is contained in:
Guinness Chen
2026-06-09 01:46:41 -07:00
parent 08cb633c06
commit 3c85d78cc9
6 changed files with 13 additions and 252 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/realtime_conversation.rs
-4
View File
@@ -739,10 +739,6 @@ fn prefix_realtime_text(text: String, prefix: &str, session_kind: RealtimeSessio
format!("{prefix}{text}")
}
pub(crate) fn prefix_realtime_v2_text(text: String, prefix: &str) -> String {
prefix_realtime_text(text, prefix, RealtimeSessionKind::V2)
}
fn validate_realtime_voice(version: RealtimeWsVersion, voice: RealtimeVoice) -> CodexResult<()> {
let voices = RealtimeVoicesList::builtin();
let allowed = match version {
codex-rs/core/src/session/handlers.rs
+2 -45
View File
@@ -15,10 +15,6 @@ use crate::session::session::Session;
use crate::session::session::SessionSettingsUpdate;
use crate::config::Config;
use crate::realtime_context::REALTIME_TURN_TOKEN_BUDGET;
use crate::realtime_context::truncate_realtime_text_to_token_budget;
use crate::realtime_conversation::REALTIME_USER_TEXT_PREFIX;
use crate::realtime_conversation::prefix_realtime_v2_text;
use crate::review_prompts::resolve_review_request;
use crate::session::spawn_review_thread;
use crate::tasks::CompactTask;
@@ -54,9 +50,7 @@ use codex_protocol::request_user_input::RequestUserInputResponse;
use crate::context_manager::is_user_turn_boundary;
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;
use codex_rmcp_client::ElicitationResponse;
use serde_json::Value;
@@ -91,14 +85,7 @@ pub async fn user_input_or_turn(
op: Op,
client_user_message_id: Option<String>,
) {
user_input_or_turn_inner(
sess,
sub_id,
op,
/*mirror_user_text_to_realtime*/ Some(()),
client_user_message_id,
)
.await;
user_input_or_turn_inner(sess, sub_id, op, client_user_message_id).await;
}
pub async fn update_thread_settings(
@@ -196,7 +183,6 @@ pub(super) async fn user_input_or_turn_inner(
sess: &Arc<Session>,
sub_id: String,
op: Op,
mirror_user_text_to_realtime: Option<()>,
client_user_message_id: Option<String>,
) {
let Op::UserInput {
@@ -230,7 +216,7 @@ pub(super) async fn user_input_or_turn_inner(
}
sess.maybe_emit_unknown_model_warning_for_turn(current_context.as_ref())
.await;
let accepted_items = match sess
match sess
.steer_input(
items.clone(),
additional_context.clone(),
@@ -242,7 +228,6 @@ pub(super) async fn user_input_or_turn_inner(
{
Ok(_) => {
current_context.session_telemetry.user_prompt(&items);
Some(items)
}
Err(SteerInputError::NoActiveTurn(items)) => {
if let Some(responsesapi_client_metadata) = responsesapi_client_metadata {
@@ -256,7 +241,6 @@ pub(super) async fn user_input_or_turn_inner(
Some(sess.mcp_elicitation_reviewer()),
)
.await;
let accepted_items = items.clone();
let additional_context_input = {
let mut state = sess.state.lock().await;
state.additional_context.merge(additional_context)
@@ -278,7 +262,6 @@ pub(super) async fn user_input_or_turn_inner(
crate::tasks::RegularTask::new(),
)
.await;
Some(accepted_items)
}
Err(err) => {
sess.send_event_raw(Event {
@@ -286,33 +269,7 @@ pub(super) async fn user_input_or_turn_inner(
msg: EventMsg::Error(err.to_error_event()),
})
.await;
None
}
};
if let (Some(items), Some(())) = (accepted_items, mirror_user_text_to_realtime) {
self::mirror_user_text_to_realtime(sess, &items).await;
}
}
async fn mirror_user_text_to_realtime(sess: &Arc<Session>, items: &[UserInput]) {
let text = UserMessageItem::new(items).message();
if text.is_empty() {
return;
}
let text = if sess.conversation.is_running_v2().await {
prefix_realtime_v2_text(text, REALTIME_USER_TEXT_PREFIX)
} else {
text
};
let text = truncate_realtime_text_to_token_budget(&text, REALTIME_TURN_TOKEN_BUDGET);
if text.is_empty() {
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}");
}
}
codex-rs/core/src/session/mod.rs
-1
View File
@@ -1128,7 +1128,6 @@ impl Session {
additional_context: Default::default(),
thread_settings: Default::default(),
},
/*mirror_user_text_to_realtime*/ None,
/*client_user_message_id*/ None,
)
.await;
codex-rs/core/tests/suite/realtime_conversation.rs
+11 -180
View File
@@ -27,7 +27,6 @@ use codex_protocol::protocol::RealtimeVoice;
use codex_protocol::protocol::RolloutItem;
use codex_protocol::protocol::SessionSource;
use codex_protocol::user_input::UserInput;
use codex_utils_output_truncation::approx_token_count;
use core_test_support::responses;
use core_test_support::responses::WebSocketConnectionConfig;
use core_test_support::responses::start_mock_server;
@@ -2099,7 +2098,7 @@ async fn conversation_startup_context_is_truncated_and_sent_once_per_start() ->
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_user_text_turn_is_sent_to_realtime_when_active() -> Result<()> {
async fn conversation_user_text_turn_is_not_sent_to_realtime() -> Result<()> {
skip_if_no_network!(Ok(()));
let api_server = start_mock_server().await;
@@ -2107,7 +2106,6 @@ async fn conversation_user_text_turn_is_sent_to_realtime_when_active() -> Result
&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"),
]),
)
@@ -2155,7 +2153,6 @@ async fn conversation_user_text_turn_is_sent_to_realtime_when_active() -> Result
assert_eq!(session_updated, "sess_user_text");
let user_text = "typed follow-up for realtime";
let prefixed_user_text = format!("[USER] {user_text}");
test.codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
@@ -2169,181 +2166,22 @@ async fn conversation_user_text_turn_is_sent_to_realtime_when_active() -> Result
})
.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(prefixed_user_text.as_str()),
)
.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(prefixed_user_text)),
);
let realtime_response_create = timeout(Duration::from_millis(200), async {
wait_for_matching_websocket_request(
&realtime_server,
"unexpected realtime response request for mirrored user text",
|request| request.body_json()["type"].as_str() == Some("response.create"),
)
.await
})
.await;
assert!(
realtime_response_create.is_err(),
"mirrored user text should not request a realtime response"
);
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: {}\nresponse.create: {}",
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(),
realtime_response_create.is_ok(),
);
insta::assert_snapshot!(
"conversation_user_text_turn_is_sent_to_realtime_when_active",
snapshot
);
realtime_server.shutdown().await;
Ok(())
}
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn conversation_user_text_turn_is_capped_when_mirrored_to_realtime() -> 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_long_user_text"),
responses::ev_assistant_message("msg_long_user_text", "ack"),
responses::ev_completed("resp_long_user_text"),
]),
)
.await;
let realtime_server = start_websocket_server(vec![vec![
vec![json!({
"type": "session.updated",
"session": { "id": "sess_long_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.experimental_realtime_ws_startup_context = Some(String::new());
}
});
let test = builder.build(&api_server).await?;
// Phase 1: start realtime so the next normal user turn mirrors over the
// active WebSocket session.
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 session_updated = wait_for_event_match(&test.codex, |msg| match msg {
EventMsg::RealtimeConversationRealtime(RealtimeConversationRealtimeEvent {
payload:
RealtimeEvent::SessionUpdated {
realtime_session_id: session_id,
..
},
}) => Some(session_id.clone()),
let turn_complete = wait_for_event_match(&test.codex, |event| match event {
EventMsg::TurnComplete(turn_complete) => Some(turn_complete.clone()),
_ => None,
})
.await;
assert_eq!(session_updated, "sess_long_user_text");
assert_eq!(turn_complete.last_agent_message, None);
// Phase 2: submit one oversized text turn. The model request should keep
// the exact user text, while the realtime mirror should get the capped copy.
let user_text = format!(
"mirror-head {} mirror-middle {} mirror-tail",
"alpha ".repeat(900),
"omega ".repeat(900),
);
test.codex
.submit(Op::UserInput {
items: vec![UserInput::Text {
text: user_text.clone(),
text_elements: Vec::new(),
}],
final_output_json_schema: None,
responsesapi_client_metadata: None,
additional_context: Default::default(),
thread_settings: Default::default(),
})
.await?;
wait_for_event(&test.codex, |event| {
matches!(event, EventMsg::TurnComplete(_))
})
.await;
// Phase 3: capture the mirrored WebSocket item; the snapshot below records
// the capped payload shape.
let realtime_text_request = wait_for_matching_websocket_request(
&realtime_server,
"capped normal user turn text mirrored to realtime",
|request| websocket_request_text(request).is_some_and(|text| text.contains("mirror-head")),
)
.await;
let realtime_text =
websocket_request_text(&realtime_text_request).expect("realtime request text");
let model_user_texts = response_mock.single_request().message_input_texts("user");
assert!(model_user_texts.iter().any(|text| text == user_text));
let realtime_request_body = realtime_text_request.body_json();
let content = &realtime_request_body["item"]["content"][0];
// Snapshot the request envelope and capped text together so reviewers can
// see the preserved head/tail and truncation marker in one place.
let snapshot = format!(
"type: {}\nitem.type: {}\nitem.role: {}\ncontent[0].type: {}\nmodel_has_full_user_text: {}\nrealtime_text_equal_full_user_text: {}\nrealtime_text_approx_tokens: {}\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(),
model_user_texts.iter().any(|text| text == &user_text),
realtime_text == user_text,
approx_token_count(&realtime_text),
realtime_text,
);
insta::assert_snapshot!(
"conversation_user_text_turn_is_capped_when_mirrored_to_realtime",
snapshot
let realtime_connections = realtime_server.connections();
assert_eq!(realtime_connections.len(), 1);
assert_eq!(realtime_connections[0].len(), 1);
assert_eq!(
realtime_connections[0][0].body_json()["type"].as_str(),
Some("session.update")
);
realtime_server.shutdown().await;
@@ -3441,7 +3279,6 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
"type": "session.updated",
"session": { "id": "sess_steer", "instructions": "backend prompt" }
})],
vec![],
vec![
json!({
"type": "conversation.input_transcript.delta",
@@ -3504,12 +3341,6 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
matches!(event, EventMsg::AgentMessageContentDelta(_))
})
.await;
let _ = wait_for_matching_websocket_request(
&realtime_server,
"first prompt mirrored to realtime",
|request| websocket_request_text(request).as_deref() == Some("first prompt"),
)
.await;
test.codex
.submit(Op::RealtimeConversationAudio(ConversationAudioParams {
codex-rs/core/tests/suite/snapshots/all__suite__realtime_conversation__conversation_user_text_turn_is_capped_when_mirrored_to_realtime.snap
-12
View File
@@ -1,12 +0,0 @@
---
source: core/tests/suite/realtime_conversation.rs
expression: snapshot
---
type: conversation.item.create
item.type: message
item.role: user
content[0].type: input_text
model_has_full_user_text: true
realtime_text_equal_full_user_text: false
realtime_text_approx_tokens: 300
content[0].text: [USER] mirror-head alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alpha alp…2419 tokens truncated…ega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega omega mirror-tail
codex-rs/core/tests/suite/snapshots/all__suite__realtime_conversation__conversation_user_text_turn_is_sent_to_realtime_when_active.snap
-10
View File
@@ -1,10 +0,0 @@
---
source: core/tests/suite/realtime_conversation.rs
expression: snapshot
---
type: conversation.item.create
item.type: message
item.role: user
content[0].type: input_text
content[0].text: [USER] typed follow-up for realtime
response.create: false