Stop mirroring Codex user input into realtime (#27116)

## Why

The realtime frontend model and the backing Codex thread should present
one coherent assistant. Raw typed messages, steers, and worker reports
belong to the orchestrator; the frontend model should receive the
orchestrator's user-facing result rather than a second copy of those
inputs.

Today normal `turn/start` input is automatically inserted into the
realtime conversation, while `turn/steer` is not. Besides creating
inconsistent context, this can make the frontend model react
independently before Codex has produced the response it should speak.

## What changed

- Remove automatic accepted-user-input mirroring into realtime
- Remove the mirror-only echo-suppression flag and dead V2 prefix helper
- Preserve explicit app-to-realtime text injection and FEM-to-Codex
delegation
- Replace the positive mirror tests and obsolete snapshots with a
negative routing regression test

## Test plan

- `cargo test -p codex-core
conversation_user_text_turn_is_not_sent_to_realtime`
- `cargo test -p codex-core
conversation_startup_context_is_truncated_and_sent_once_per_start`
- `cargo test -p codex-core inbound_handoff_request_starts_turn`

codex-rs/core/src/realtime_conversation.rs

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

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

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

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

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

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