Refresh realtime context for each model request (#47596)

## Why

A realtime call can start or end while a text turn is still running. Using call activity captured at turn creation leaves subsequent model requests with stale realtime context.

## What changed

Capture realtime activity and mode instructions together in a `StepContext` snapshot for each model request. Use that snapshot to build realtime instructions and persist `realtime_active` in the request's turn context. Store the conversation and mode instructions under one mutex so they are installed together when a call starts.

## Testing

Extend the realtime handoff integration test to start and close a call during one text turn, verifying that successive requests include the appropriate start and end instructions.

GitOrigin-RevId: 4c5c2b47b7c0c06ae28167013218dc2e3cd72c42

codex-rs/core/src/realtime_conversation.rs

@@ -165,8 +165,12 @@ impl RealtimeHandoffAdmission {
 }
 
 pub(crate) struct RealtimeConversationManager {
-    state: Mutex<Option<ConversationState>>,
-    mode_instructions: Mutex<Option<RealtimeModeInstructions>>,
+    state: Mutex<RealtimeConversationManagerState>,
+}
+
+struct RealtimeConversationManagerState {
+    conversation: Option<ConversationState>,
+    mode_instructions: Option<RealtimeModeInstructions>,
 }
 
 #[derive(Clone, Debug)]
@@ -175,6 +179,12 @@ pub(crate) struct RealtimeModeInstructions {
     pub(crate) end: Option<String>,
 }
 
+#[derive(Clone, Debug)]
+pub(crate) struct RealtimeConversationSnapshot {
+    pub(crate) active: bool,
+    pub(crate) mode_instructions: Option<RealtimeModeInstructions>,
+}
+
 #[derive(Clone, Copy, Debug, PartialEq, Eq)]
 enum RealtimeSessionKind {
     V1,
@@ -543,13 +553,22 @@ struct RealtimeStartOutput {
 impl RealtimeConversationManager {
     pub(crate) fn new() -> Self {
         Self {
-            state: Mutex::new(None),
-            mode_instructions: Mutex::new(None),
+            state: Mutex::new(RealtimeConversationManagerState {
+                conversation: None,
+                mode_instructions: None,
+            }),
         }
     }
 
-    pub(crate) async fn mode_instructions(&self) -> Option<RealtimeModeInstructions> {
-        self.mode_instructions.lock().await.clone()
+    pub(crate) async fn snapshot(&self) -> RealtimeConversationSnapshot {
+        let state = self.state.lock().await;
+        RealtimeConversationSnapshot {
+            active: state
+                .conversation
+                .as_ref()
+                .is_some_and(|conversation| conversation.realtime_active.load(Ordering::Relaxed)),
+            mode_instructions: state.mode_instructions.clone(),
+        }
     }
 
     #[tracing::instrument(
@@ -560,6 +579,7 @@ impl RealtimeConversationManager {
     pub(crate) async fn running_state(&self) -> Option<()> {
         let state = self.state.lock().await;
         state
+            .conversation
             .as_ref()
             .and_then(|state| state.realtime_active.load(Ordering::Relaxed).then_some(()))
     }
@@ -567,7 +587,7 @@ impl RealtimeConversationManager {
     pub(crate) async fn is_running_v2(&self) -> bool {
         let state = self.state.lock().await;
         matches!(
-            state.as_ref(),
+            state.conversation.as_ref(),
             Some(state)
                 if state.realtime_active.load(Ordering::Relaxed)
                     && state.session_kind == RealtimeSessionKind::V2
@@ -581,6 +601,7 @@ impl RealtimeConversationManager {
             .state
             .lock()
             .await
+            .conversation
             .as_ref()
             .map(|state| Arc::clone(&state.route_handoffs));
         if let Some(route_handoffs) = route_handoffs {
@@ -595,18 +616,20 @@ impl RealtimeConversationManager {
     ) -> CodexResult<RealtimeStartOutput> {
         let previous_state = {
             let mut guard = self.state.lock().await;
-            guard.take()
+            guard.conversation.take()
         };
         if let Some(state) = previous_state {
             stop_conversation_state(state, RealtimeFanoutTaskStop::Await).await;
         }
 
-        let output = self.start_inner(start).await?;
-        *self.mode_instructions.lock().await = Some(mode_instructions);
-        Ok(output)
+        self.start_inner(start, mode_instructions).await
     }
 
-    async fn start_inner(&self, start: RealtimeStart) -> CodexResult<RealtimeStartOutput> {
+    async fn start_inner(
+        &self,
+        start: RealtimeStart,
+        mode_instructions: RealtimeModeInstructions,
+    ) -> CodexResult<RealtimeStartOutput> {
         let RealtimeStart {
             api_provider,
             http_client_factory,
@@ -744,8 +767,8 @@ impl RealtimeConversationManager {
             (task, None)
         };
 
-        let mut guard = self.state.lock().await;
-        *guard = Some(ConversationState {
+        let mut state = self.state.lock().await;
+        state.conversation = Some(ConversationState {
             audio_tx,
             text_tx,
             session_kind,
@@ -756,6 +779,7 @@ impl RealtimeConversationManager {
             route_handoffs: Arc::clone(&route_handoffs),
             stop_token,
         });
+        state.mode_instructions = Some(mode_instructions);
         Ok(RealtimeStartOutput {
             realtime_active,
             route_handoffs,
@@ -773,7 +797,7 @@ impl RealtimeConversationManager {
         let mut fanout_task = Some(fanout_task);
         {
             let mut guard = self.state.lock().await;
-            if let Some(state) = guard.as_mut()
+            if let Some(state) = guard.conversation.as_mut()
                 && Arc::ptr_eq(&state.realtime_active, realtime_active)
             {
                 state.fanout_task = fanout_task.take();
@@ -789,8 +813,10 @@ impl RealtimeConversationManager {
     pub(crate) async fn finish_if_active(&self, realtime_active: &Arc<AtomicBool>) {
         let state = {
             let mut guard = self.state.lock().await;
-            match guard.as_ref() {
-                Some(state) if Arc::ptr_eq(&state.realtime_active, realtime_active) => guard.take(),
+            match guard.conversation.as_ref() {
+                Some(state) if Arc::ptr_eq(&state.realtime_active, realtime_active) => {
+                    guard.conversation.take()
+                }
                 _ => None,
             }
         };
@@ -803,7 +829,10 @@ impl RealtimeConversationManager {
     pub(crate) async fn audio_in(&self, frame: RealtimeAudioFrame) -> CodexResult<()> {
         let sender = {
             let guard = self.state.lock().await;
-            guard.as_ref().map(|state| state.audio_tx.clone())
+            guard
+                .conversation
+                .as_ref()
+                .map(|state| state.audio_tx.clone())
         };
 
         let Some(sender) = sender else {
@@ -828,6 +857,7 @@ impl RealtimeConversationManager {
         let sender = {
             let guard = self.state.lock().await;
             guard
+                .conversation
                 .as_ref()
                 .map(|state| (state.text_tx.clone(), state.session_kind))
         };
@@ -854,6 +884,7 @@ impl RealtimeConversationManager {
             .state
             .lock()
             .await
+            .conversation
             .as_ref()
             .map(|state| state.handoff.clone());
         let Some(handoff) = handoff.filter(|handoff| {
@@ -884,7 +915,7 @@ impl RealtimeConversationManager {
     ) -> CodexResult<()> {
         let handoff = {
             let guard = self.state.lock().await;
-            let Some(state) = guard.as_ref() else {
+            let Some(state) = guard.conversation.as_ref() else {
                 return Err(CodexErr::InvalidRequest(
                     "conversation is not running".to_string(),
                 ));
@@ -979,7 +1010,10 @@ impl RealtimeConversationManager {
     ) {
         let handoff = {
             let guard = self.state.lock().await;
-            guard.as_ref().map(|state| state.handoff.clone())
+            guard
+                .conversation
+                .as_ref()
+                .map(|state| state.handoff.clone())
         };
         let Some(handoff) = handoff else {
             return;
@@ -1037,7 +1071,7 @@ impl RealtimeConversationManager {
         }
         let handoff = {
             let guard = self.state.lock().await;
-            let Some(state) = guard.as_ref() else {
+            let Some(state) = guard.conversation.as_ref() else {
                 return Err(CodexErr::InvalidRequest(
                     "conversation is not running".to_string(),
                 ));
@@ -1069,7 +1103,10 @@ impl RealtimeConversationManager {
     pub(crate) async fn finish_handoff_stream_item(&self, item_id: &str) -> bool {
         let handoff = {
             let guard = self.state.lock().await;
-            guard.as_ref().map(|state| state.handoff.clone())
+            guard
+                .conversation
+                .as_ref()
+                .map(|state| state.handoff.clone())
         };
         let Some(handoff) = handoff else {
             return false;
@@ -1103,7 +1140,7 @@ impl RealtimeConversationManager {
 
         let handoff = {
             let guard = self.state.lock().await;
-            let Some(state) = guard.as_ref() else {
+            let Some(state) = guard.conversation.as_ref() else {
                 return Err(CodexErr::InvalidRequest(
                     "conversation is not running".to_string(),
                 ));
@@ -1124,7 +1161,10 @@ impl RealtimeConversationManager {
     pub(crate) async fn handoff_complete(&self) -> CodexResult<()> {
         let handoff = {
             let guard = self.state.lock().await;
-            guard.as_ref().map(|state| state.handoff.clone())
+            guard
+                .conversation
+                .as_ref()
+                .map(|state| state.handoff.clone())
         };
         let Some(handoff) = handoff else {
             return Ok(());
@@ -1164,7 +1204,10 @@ impl RealtimeConversationManager {
     pub(crate) async fn clear_active_handoff(&self) {
         let handoff = {
             let guard = self.state.lock().await;
-            guard.as_ref().map(|state| state.handoff.clone())
+            guard
+                .conversation
+                .as_ref()
+                .map(|state| state.handoff.clone())
         };
         if let Some(handoff) = handoff {
             {
@@ -1179,7 +1222,7 @@ impl RealtimeConversationManager {
     pub(crate) async fn shutdown(&self) -> CodexResult<()> {
         let state = {
             let mut guard = self.state.lock().await;
-            guard.take()
+            guard.conversation.take()
         };
 
         if let Some(state) = state {

codex-rs/core/src/session/mod.rs

@@ -3947,6 +3947,7 @@ impl Session {
         ) = prepared_tools??;
         turn_context.extension_data.insert(selected_plugins);
         Ok(Arc::new(StepContext {
+            realtime: self.conversation.snapshot().await,
             settings,
             token_budget,
             session_telemetry,

codex-rs/core/src/session/step_context.rs

@@ -5,6 +5,7 @@ use std::sync::Arc;
 use crate::agents_md::LoadedAgentsMd;
 use crate::config::TokenBudgetConfig;
 use crate::environment_selection::TurnEnvironmentSnapshot;
+use crate::realtime_conversation::RealtimeConversationSnapshot;
 use crate::session::step_settings::ResolvedStepSettings;
 use crate::session::turn_context::TurnContext;
 use crate::tools::router::ToolRouter;
@@ -18,6 +19,8 @@ use codex_protocol::protocol::TurnContextItem;
 /// Request-scoped state that may change between model sampling requests.
 pub(crate) struct StepContext {
     pub(crate) turn: Arc<TurnContext>,
+    /// Realtime call activity and instructions captured for this sampling request.
+    pub(crate) realtime: RealtimeConversationSnapshot,
     /// One immutable settings version captured before request preparation.
     pub(crate) settings: Arc<ResolvedStepSettings>,
     /// Frozen turn preferences resolved against this step's captured model.
@@ -38,9 +41,10 @@ pub(crate) struct StepContext {
 }
 
 impl StepContext {
-    /// Persist the summary captured for this request, even after a live settings update.
+    /// Persist the context captured for this request, even after a live update.
     pub(crate) fn to_turn_context_item(&self) -> TurnContextItem {
         let mut item = self.turn.to_turn_context_item();
+        item.realtime_active = Some(self.realtime.active);
         item.summary = self.settings.reasoning_summary;
         item
     }

codex-rs/core/src/session/tests.rs

@@ -26,6 +26,7 @@ use crate::environment_selection::TurnEnvironmentState;
 use crate::function_tool::FunctionCallError;
 use crate::hook_mcp_executor::CoreHookMcpExecutor;
 use crate::plugins::plugins_manager_for_config;
+use crate::realtime_conversation::RealtimeConversationSnapshot;
 use crate::session::step_context::StepContext;
 use crate::shell::default_user_shell;
 use crate::shell_snapshot::ShellSnapshot;
@@ -275,6 +276,10 @@ impl StepContext {
             ),
             settings: Arc::new(settings),
             session_telemetry: turn.session_telemetry.clone(),
+            realtime: RealtimeConversationSnapshot {
+                active: turn.realtime_active,
+                mode_instructions: None,
+            },
             turn: Arc::clone(&turn),
             environments,
             selected_capability_roots: Vec::new(),

codex-rs/core/src/session/turn_context.rs

@@ -304,6 +304,7 @@ pub(crate) struct NewTurnContextOptions {
 pub struct TurnContext {
     pub(crate) sub_id: String,
     pub(crate) trace_id: Option<String>,
+    /// Call state at turn creation; model requests use the `StepContext` snapshot.
     pub(crate) realtime_active: bool,
     pub(crate) code_mode_available: bool,
     /// Turn-scoped configuration. Read step-specific settings such as service tier and

codex-rs/core/src/session/world_state.rs

@@ -143,17 +143,19 @@ impl Session {
             .and_then(|config| config.guidance_message.as_deref())
             .filter(|_| token_budget_enabled);
         world_state.add_section(ContextWindowGuidanceState::new(guidance));
-        let realtime_mode_instructions = self.conversation.mode_instructions().await;
+        let realtime = &step_context.realtime;
         world_state.add_section(RealtimeState::new(
-            turn_context.realtime_active,
-            realtime_mode_instructions
+            realtime.active,
+            realtime
+                .mode_instructions
                 .as_ref()
                 .and_then(|instructions| instructions.start.as_deref())
                 .or(turn_context
                     .config
                     .experimental_realtime_start_instructions
                     .as_deref()),
-            realtime_mode_instructions
+            realtime
+                .mode_instructions
                 .as_ref()
                 .and_then(|instructions| instructions.end.as_deref()),
         ));

codex-rs/core/tests/suite/realtime_conversation.rs

@@ -9,6 +9,7 @@ use codex_config::config_toml::RealtimeWsMode;
 use codex_config::config_toml::RealtimeWsVersion;
 use codex_core::StartThreadOptions;
 use codex_core::TurnInputRequest;
+use codex_core::TurnInputSubmission;
 use codex_core::test_support::auth_manager_from_auth;
 use codex_history::InitialHistory;
 use codex_history::RolloutItem;
@@ -5537,7 +5538,7 @@ async fn inbound_handoff_request_does_not_block_realtime_event_forwarding() -> R
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
+async fn inbound_handoff_request_updates_realtime_state_during_active_turn() -> Result<()> {
     skip_if_no_network!(Ok(()));
 
     let (gate_completed_tx, gate_completed_rx) = oneshot::channel();
@@ -5567,18 +5568,23 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
             body: sse_event(responses::ev_completed("resp-1")),
         },
     ];
+    let (second_completed_tx, second_completed_rx) = oneshot::channel();
     let second_chunks = vec![
         StreamingSseChunk {
             gate: None,
             body: sse_event(responses::ev_response_created("resp-2")),
         },
         StreamingSseChunk {
-            gate: None,
+            gate: Some(second_completed_rx),
             body: sse_event(responses::ev_completed("resp-2")),
         },
     ];
+    let third_chunks = vec![StreamingSseChunk {
+        gate: None,
+        body: sse_event(responses::ev_completed("resp-3")),
+    }];
     let (api_server, completions) =
-        start_streaming_sse_server(vec![first_chunks, second_chunks]).await;
+        start_streaming_sse_server(vec![first_chunks, second_chunks, third_chunks]).await;
 
     let realtime_server = start_websocket_server_with_headers(vec![WebSocketConnectionConfig {
         requests: vec![
@@ -5614,6 +5620,22 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
     });
     let test = builder.build_with_streaming_server(&api_server).await?;
 
+    let started = test
+        .codex
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "first prompt".to_string(),
+            text_elements: Vec::new(),
+        }]))
+        .await?;
+    let TurnInputSubmission::Started { turn_id } = started else {
+        panic!("expected the text input to start a turn");
+    };
+
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::AgentMessageContentDelta(_))
+    })
+    .await;
+
     test.codex
         .submit(Op::RealtimeConversationStart(ConversationStartParams {
             client_managed_handoffs: false,
@@ -5629,8 +5651,8 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
             output_modality: RealtimeOutputModality::Audio,
             include_startup_context: true,
             initial_items: Vec::new(),
-            realtime_start_instructions: None,
-            realtime_end_instructions: None,
+            realtime_start_instructions: Some("Give frequent spoken progress updates.".to_string()),
+            realtime_end_instructions: Some("Return to normal text updates.".to_string()),
             prompt: Some(Some("backend prompt".to_string())),
             realtime_session_id: None,
             transport: None,
@@ -5650,18 +5672,6 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
     })
     .await;
 
-    test.codex
-        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
-            text: "first prompt".to_string(),
-            text_elements: Vec::new(),
-        }]))
-        .await?;
-
-    wait_for_event(&test.codex, |event| {
-        matches!(event, EventMsg::AgentMessageContentDelta(_))
-    })
-    .await;
-
     test.codex
         .submit(Op::RealtimeConversationAudio(ConversationAudioParams {
             frame: RealtimeAudioFrame {
@@ -5685,21 +5695,47 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
     let mut completion_iter = completions.into_iter();
     let first_completion = completion_iter.next().expect("missing first completion");
     let second_completion = completion_iter.next().expect("missing second completion");
+    let third_completion = completion_iter.next().expect("missing third completion");
 
     let _ = gate_completed_tx.send(());
     first_completion
         .await
         .expect("first request did not complete");
+    timeout(
+        Duration::from_secs(10),
+        api_server.wait_for_request_count(/*count*/ 2),
+    )
+    .await
+    .context("steered request did not start")?;
+
+    // End the call while the same text turn is still in flight.
+    test.codex.submit(Op::RealtimeConversationClose).await?;
+    wait_for_event(&test.codex, |event| {
+        matches!(event, EventMsg::RealtimeConversationClosed(_))
+    })
+    .await;
+    let steered = test
+        .codex
+        .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text {
+            text: "continue in text".to_string(),
+            text_elements: Vec::new(),
+        }]))
+        .await?;
+    assert_eq!(steered, TurnInputSubmission::Steered { turn_id });
+    let _ = second_completed_tx.send(());
     second_completion
         .await
         .expect("second request did not complete");
+    third_completion
+        .await
+        .expect("third request did not complete");
     wait_for_event(&test.codex, |event| {
         matches!(event, EventMsg::TurnComplete(_))
     })
     .await;
 
     let requests = api_server.requests().await;
-    assert_eq!(requests.len(), 2);
+    assert_eq!(requests.len(), 3);
 
     let first_body: Value = serde_json::from_slice(&requests[0]).expect("parse first request");
     let second_body: Value = serde_json::from_slice(&requests[1]).expect("parse second request");
@@ -5717,6 +5753,29 @@ async fn inbound_handoff_request_steers_active_turn() -> Result<()> {
     assert!(second_texts.iter().any(|text| text
         == "<realtime_delegation>\n  <input>steer via realtime</input>\n  <transcript_delta>user: steer via realtime</transcript_delta>\n</realtime_delegation>"));
 
+    let realtime_instructions = requests
+        .iter()
+        .map(|request| {
+            let body: Value = serde_json::from_slice(request).expect("parse request");
+            message_input_texts(&body, "developer")
+                .into_iter()
+                .filter(|text| text.contains("<realtime_conversation>"))
+                .collect::<Vec<_>>()
+        })
+        .collect::<Vec<_>>();
+    let start_instructions =
+        "<realtime_conversation>\nGive frequent spoken progress updates.\n</realtime_conversation>";
+    let end_instructions =
+        "<realtime_conversation>\nReturn to normal text updates.\n</realtime_conversation>";
+    assert_eq!(
+        realtime_instructions,
+        vec![
+            vec![],
+            vec![start_instructions.to_string()],
+            vec![start_instructions.to_string(), end_instructions.to_string()],
+        ]
+    );
+
     realtime_server.shutdown().await;
     api_server.shutdown().await;
     Ok(())