diff --git a/codex-rs/core/src/realtime_conversation.rs b/codex-rs/core/src/realtime_conversation.rs index 46d2d797a9..db2cde5f0d 100644 --- a/codex-rs/core/src/realtime_conversation.rs +++ b/codex-rs/core/src/realtime_conversation.rs @@ -165,8 +165,12 @@ impl RealtimeHandoffAdmission { } pub(crate) struct RealtimeConversationManager { - state: Mutex>, - mode_instructions: Mutex>, + state: Mutex, +} + +struct RealtimeConversationManagerState { + conversation: Option, + mode_instructions: Option, } #[derive(Clone, Debug)] @@ -175,6 +179,12 @@ pub(crate) struct RealtimeModeInstructions { pub(crate) end: Option, } +#[derive(Clone, Debug)] +pub(crate) struct RealtimeConversationSnapshot { + pub(crate) active: bool, + pub(crate) mode_instructions: Option, +} + #[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 { - 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 { 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 { + async fn start_inner( + &self, + start: RealtimeStart, + mode_instructions: RealtimeModeInstructions, + ) -> CodexResult { 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) { 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 { diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index b31ab6e054..961f7b5d47 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/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, diff --git a/codex-rs/core/src/session/step_context.rs b/codex-rs/core/src/session/step_context.rs index 855d854fd0..62c387509c 100644 --- a/codex-rs/core/src/session/step_context.rs +++ b/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, + /// 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, /// 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 } diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index e241d03d04..9c81bb39fe 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/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(), diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index 76befa42c6..1183ef2db6 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/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, + /// 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 diff --git a/codex-rs/core/src/session/world_state.rs b/codex-rs/core/src/session/world_state.rs index 323ddfbf28..249aca8bd4 100644 --- a/codex-rs/core/src/session/world_state.rs +++ b/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()), )); diff --git a/codex-rs/core/tests/suite/realtime_conversation.rs b/codex-rs/core/tests/suite/realtime_conversation.rs index e4c0ba1295..4a9d557c6a 100644 --- a/codex-rs/core/tests/suite/realtime_conversation.rs +++ b/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 == "\n steer via realtime\n user: steer via realtime\n")); + 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("")) + .collect::>() + }) + .collect::>(); + let start_instructions = + "\nGive frequent spoken progress updates.\n"; + let end_instructions = + "\nReturn to normal text updates.\n"; + 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(())