From b9680065b05accae1d2093de1ada06e009abb128 Mon Sep 17 00:00:00 2001 From: pakrym-oai Date: Fri, 17 Jul 2026 19:28:44 +0000 Subject: [PATCH] Track realtime conversation state in world state (#33893) ## What changed - Represent whether realtime is active as a persisted world-state section. - Emit start and end instructions from world-state transitions, including when reconstructing context after resume. - Retain existing realtime start fragments and avoid emitting another start when custom instructions change while realtime remains active. ## Testing - Add snapshot coverage for absent, unknown, active, and inactive realtime state transitions. - Cover realtime start and end updates plus resuming with changed custom start instructions. GitOrigin-RevId: ec670b8132b0f1c9317738bfdeb272069927d91d --- .../src/context/realtime_end_instructions.rs | 4 + codex-rs/core/src/context/world_state/mod.rs | 2 + .../core/src/context/world_state/realtime.rs | 87 +++++++ .../src/context/world_state/realtime_tests.rs | 43 ++++ ...rld_state__realtime__tests__snapshots.snap | 71 ++++++ codex-rs/core/src/context_manager/updates.rs | 41 --- codex-rs/core/src/session/mod.rs | 16 +- codex-rs/core/src/session/tests.rs | 93 +------ codex-rs/core/src/session/world_state.rs | 9 +- codex-rs/core/tests/suite/compact_remote.rs | 234 ++++-------------- ...t_resume_restates_realtime_end_shapes.snap | 21 -- ...mpaction_restates_realtime_end_shapes.snap | 21 -- 12 files changed, 275 insertions(+), 367 deletions(-) create mode 100644 codex-rs/core/src/context/world_state/realtime.rs create mode 100644 codex-rs/core/src/context/world_state/realtime_tests.rs create mode 100644 codex-rs/core/src/context/world_state/snapshots/codex_core__context__world_state__realtime__tests__snapshots.snap delete mode 100644 codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_compact_resume_restates_realtime_end_shapes.snap delete mode 100644 codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_pre_turn_compaction_restates_realtime_end_shapes.snap diff --git a/codex-rs/core/src/context/realtime_end_instructions.rs b/codex-rs/core/src/context/realtime_end_instructions.rs index b4225dfb49..2db5947618 100644 --- a/codex-rs/core/src/context/realtime_end_instructions.rs +++ b/codex-rs/core/src/context/realtime_end_instructions.rs @@ -32,6 +32,10 @@ impl ContextualUserFragment for RealtimeEndInstructions { ) } + fn matches_text(text: &str) -> bool { + text.contains(END_INSTRUCTIONS.trim()) + } + fn body(&self) -> String { format!("\n{}\n\nReason: {}\n", END_INSTRUCTIONS.trim(), self.reason) } diff --git a/codex-rs/core/src/context/world_state/mod.rs b/codex-rs/core/src/context/world_state/mod.rs index 47b420d719..c47fba8446 100644 --- a/codex-rs/core/src/context/world_state/mod.rs +++ b/codex-rs/core/src/context/world_state/mod.rs @@ -4,6 +4,7 @@ mod collaboration_mode; mod environment; mod environments_instructions; mod plugins_instructions; +mod realtime; #[cfg(test)] mod test_support; @@ -27,6 +28,7 @@ pub(crate) use collaboration_mode::CollaborationModeState; pub(crate) use environment::EnvironmentsState; pub(crate) use environments_instructions::EnvironmentsInstructionsState; pub(crate) use plugins_instructions::PluginsInstructionsState; +pub(crate) use realtime::RealtimeState; trait ErasedWorldStateSection: Send + Sync { fn snapshot(&self) -> Option; diff --git a/codex-rs/core/src/context/world_state/realtime.rs b/codex-rs/core/src/context/world_state/realtime.rs new file mode 100644 index 0000000000..383bb4b2cd --- /dev/null +++ b/codex-rs/core/src/context/world_state/realtime.rs @@ -0,0 +1,87 @@ +use super::PreviousSectionState; +use super::WorldStateSection; +use crate::context::ContextualUserFragment; +use crate::context::RealtimeEndInstructions; +use crate::context::RealtimeStartInstructions; +use crate::context::RealtimeStartWithInstructions; +use serde::Deserialize; +use serde::Serialize; + +/// The realtime conversation state currently visible to the model. +#[derive(Clone, Debug)] +pub(crate) struct RealtimeState { + snapshot: RealtimeSnapshot, + start_instructions: Option, +} + +#[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)] +pub(crate) struct RealtimeSnapshot { + active: bool, +} + +impl RealtimeState { + pub(crate) fn new(active: bool, start_instructions: Option<&str>) -> Self { + Self { + snapshot: RealtimeSnapshot { active }, + start_instructions: start_instructions.map(str::to_string), + } + } + + fn render_start(&self) -> Box { + match self.start_instructions.as_deref() { + Some(instructions) => Box::new(RealtimeStartWithInstructions::new(instructions)), + None => Box::new(RealtimeStartInstructions), + } + } + + fn render_transition(&self, previous_active: bool) -> Option> { + match (previous_active, self.snapshot.active) { + (false, true) => Some(self.render_start()), + (true, false) => Some(Box::new(RealtimeEndInstructions::new("inactive"))), + (false, false) | (true, true) => None, + } + } +} + +impl WorldStateSection for RealtimeState { + const ID: &'static str = "realtime"; + type Snapshot = RealtimeSnapshot; + + fn snapshot(&self) -> Self::Snapshot { + self.snapshot.clone() + } + + fn matches_legacy_fragment(role: &str, text: &str) -> bool { + role == "developer" + && RealtimeStartInstructions::matches_text(text) + && !RealtimeEndInstructions::matches_text(text) + } + + fn has_retained_fragment_matcher() -> bool { + true + } + + fn matches_retained_fragment(role: &str, text: &str) -> bool { + Self::matches_legacy_fragment(role, text) + } + + fn render_diff( + &self, + previous: PreviousSectionState<'_, Self::Snapshot>, + ) -> Option> { + match previous { + PreviousSectionState::Known(previous) if previous == &self.snapshot => None, + PreviousSectionState::Known(previous) => self.render_transition(previous.active), + PreviousSectionState::Absent | PreviousSectionState::Unknown + if self.snapshot.active => + { + Some(self.render_start()) + } + PreviousSectionState::Absent | PreviousSectionState::Unknown => None, + } + } +} + +#[cfg(test)] +#[path = "realtime_tests.rs"] +mod tests; diff --git a/codex-rs/core/src/context/world_state/realtime_tests.rs b/codex-rs/core/src/context/world_state/realtime_tests.rs new file mode 100644 index 0000000000..06e3d8e4ce --- /dev/null +++ b/codex-rs/core/src/context/world_state/realtime_tests.rs @@ -0,0 +1,43 @@ +use super::super::test_support::render_section_cases; +use super::*; + +fn state(active: bool, start_instructions: Option<&str>) -> RealtimeState { + RealtimeState::new(active, start_instructions) +} + +#[test] +fn snapshots() { + use PreviousSectionState::Absent; + use PreviousSectionState::Known; + use PreviousSectionState::Unknown; + + let inactive = state(/*active*/ false, /*start_instructions*/ None); + let active = state(/*active*/ true, /*start_instructions*/ None); + let custom_active = state(/*active*/ true, Some("custom realtime instructions")); + let changed_custom_active = state( + /*active*/ true, + Some("changed custom realtime instructions"), + ); + + insta::assert_snapshot!(render_section_cases(&[ + (Absent, Absent), + (Absent, Known(&inactive)), + (Absent, Known(&active)), + (Known(&inactive), Known(&active)), + (Known(&inactive), Known(&custom_active)), + (Known(&active), Known(&active)), + (Known(&custom_active), Known(&changed_custom_active)), + (Known(&active), Known(&inactive)), + (Unknown, Known(&active)), + (Unknown, Known(&inactive)), + ])); +} + +#[test] +fn retained_fragment_matcher_only_matches_starts() { + let start = RealtimeStartWithInstructions::new("custom instructions").render(); + let end = RealtimeEndInstructions::new("inactive").render(); + + assert!(RealtimeState::matches_legacy_fragment("developer", &start)); + assert!(!RealtimeState::matches_legacy_fragment("developer", &end)); +} diff --git a/codex-rs/core/src/context/world_state/snapshots/codex_core__context__world_state__realtime__tests__snapshots.snap b/codex-rs/core/src/context/world_state/snapshots/codex_core__context__world_state__realtime__tests__snapshots.snap new file mode 100644 index 0000000000..f5f4345b9f --- /dev/null +++ b/codex-rs/core/src/context/world_state/snapshots/codex_core__context__world_state__realtime__tests__snapshots.snap @@ -0,0 +1,71 @@ +--- +source: core/src/context/world_state/realtime_tests.rs +expression: "render_section_cases(&[(Absent, Absent), (Absent, Known(&inactive)),\n(Absent, Known(&active)), (Known(&inactive), Known(&active)),\n(Known(&inactive), Known(&custom_active)), (Known(&active), Known(&active)),\n(Known(&custom_active), Known(&changed_custom_active)),\n(Known(&active), Known(&inactive)), (Unknown, Known(&active)),\n(Unknown, Known(&inactive)),])" +--- +Absent -> Absent +None + +Absent -> {"active":false} +None + +Absent -> {"active":true} (role - developer) + +Realtime conversation started. + +You are operating as a backend executor behind an intermediary. The user does not talk to you directly. Any response you produce will be consumed by the intermediary and may be summarized before the user sees it. + +When invoked, you receive the latest conversation transcript and any relevant mode or metadata. The intermediary may invoke you even when backend help is not actually needed. Use the transcript to decide whether you should do work. If backend help is unnecessary, avoid verbose responses that add user-visible latency. + +When user text is routed from realtime, treat it as a transcript. It may be unpunctuated or contain recognition errors. + +- Keep responses concise and action-oriented. Your updates should help the intermediary respond to the user. + + +{"active":false} -> {"active":true} (role - developer) + +Realtime conversation started. + +You are operating as a backend executor behind an intermediary. The user does not talk to you directly. Any response you produce will be consumed by the intermediary and may be summarized before the user sees it. + +When invoked, you receive the latest conversation transcript and any relevant mode or metadata. The intermediary may invoke you even when backend help is not actually needed. Use the transcript to decide whether you should do work. If backend help is unnecessary, avoid verbose responses that add user-visible latency. + +When user text is routed from realtime, treat it as a transcript. It may be unpunctuated or contain recognition errors. + +- Keep responses concise and action-oriented. Your updates should help the intermediary respond to the user. + + +{"active":false} -> {"active":true} (role - developer) + +custom realtime instructions + + +{"active":true} -> {"active":true} +None + +{"active":true} -> {"active":true} +None + +{"active":true} -> {"active":false} (role - developer) + +Realtime conversation ended. + +Subsequent user input will return to typed text rather than transcript-style text. Do not assume recognition errors or missing punctuation once realtime has ended. Resume normal chat behavior. + +Reason: inactive + + +Unknown -> {"active":true} (role - developer) + +Realtime conversation started. + +You are operating as a backend executor behind an intermediary. The user does not talk to you directly. Any response you produce will be consumed by the intermediary and may be summarized before the user sees it. + +When invoked, you receive the latest conversation transcript and any relevant mode or metadata. The intermediary may invoke you even when backend help is not actually needed. Use the transcript to decide whether you should do work. If backend help is unnecessary, avoid verbose responses that add user-visible latency. + +When user text is routed from realtime, treat it as a transcript. It may be unpunctuated or contain recognition errors. + +- Keep responses concise and action-oriented. Your updates should help the intermediary respond to the user. + + +Unknown -> {"active":false} +None diff --git a/codex-rs/core/src/context_manager/updates.rs b/codex-rs/core/src/context_manager/updates.rs index 3218ababed..d7e1c47b68 100644 --- a/codex-rs/core/src/context_manager/updates.rs +++ b/codex-rs/core/src/context_manager/updates.rs @@ -4,9 +4,6 @@ use crate::context::ModelSwitchInstructions; use crate::context::MultiAgentModeInstructions; use crate::context::PermissionsInstructions; use crate::context::PersonalitySpecInstructions; -use crate::context::RealtimeEndInstructions; -use crate::context::RealtimeStartInstructions; -use crate::context::RealtimeStartWithInstructions; use crate::session::PreviousTurnSettings; use crate::session::turn_context::TurnContext; use codex_execpolicy::Policy; @@ -85,43 +82,6 @@ fn build_multi_agent_mode_update_item( } } -pub(crate) fn build_realtime_update_item( - previous: Option<&TurnContextItem>, - previous_turn_settings: Option<&PreviousTurnSettings>, - next: &TurnContext, -) -> Option { - match ( - previous.and_then(|item| item.realtime_active), - next.realtime_active, - ) { - (Some(true), false) => Some(RealtimeEndInstructions::new("inactive").render()), - (Some(false), true) | (None, true) => Some( - if let Some(instructions) = next - .config - .experimental_realtime_start_instructions - .as_deref() - { - RealtimeStartWithInstructions::new(instructions).render() - } else { - RealtimeStartInstructions.render() - }, - ), - (Some(true), true) | (Some(false), false) => None, - (None, false) => previous_turn_settings - .and_then(|settings| settings.realtime_active) - .filter(|realtime_active| *realtime_active) - .map(|_| RealtimeEndInstructions::new("inactive").render()), - } -} - -pub(crate) fn build_initial_realtime_item( - previous: Option<&TurnContextItem>, - previous_turn_settings: Option<&PreviousTurnSettings>, - next: &TurnContext, -) -> Option { - build_realtime_update_item(previous, previous_turn_settings, next) -} - fn build_personality_update_item( previous: Option<&TurnContextItem>, next: &TurnContext, @@ -238,7 +198,6 @@ pub(crate) fn build_settings_update_items( build_model_instructions_update_item(previous_turn_settings, next), build_permissions_update_item(previous, next, exec_policy), build_multi_agent_mode_update_item(previous, next), - build_realtime_update_item(previous, previous_turn_settings, next), build_personality_update_item(previous, next, personality_feature_enabled), ] .into_iter() diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 26b6689946..0d5e5b7144 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -3193,16 +3193,9 @@ impl Session { let mut developer_sections = Vec::::with_capacity(8); let mut contextual_user_sections = Vec::::with_capacity(2); let mut separate_developer_sections = Vec::::new(); - let ( - reference_context_item, - previous_turn_settings, - base_instructions, - session_source, - auto_compact_window_ids, - ) = { + let (previous_turn_settings, base_instructions, session_source, auto_compact_window_ids) = { let state = self.state.lock().await; ( - state.reference_context_item(), state.previous_turn_settings(), state.session_configuration.base_instructions.clone(), state.session_configuration.session_source.clone(), @@ -3260,13 +3253,6 @@ impl Session { { developer_sections.push(developer_instructions.to_string()); } - if let Some(realtime_update) = crate::context_manager::updates::build_initial_realtime_item( - reference_context_item.as_ref(), - previous_turn_settings.as_ref(), - turn_context, - ) { - developer_sections.push(realtime_update); - } if self.features.enabled(Feature::Personality) && let Some(personality) = turn_context.personality { diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 95b5adefa9..fcfe649678 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -8180,7 +8180,7 @@ async fn record_context_update_items( } #[tokio::test] -async fn build_settings_update_items_emits_realtime_start_when_session_becomes_live() { +async fn record_context_updates_emits_realtime_start_when_session_becomes_live() { let (session, previous_context) = make_session_and_context().await; let previous_context = Arc::new(previous_context); let mut current_context = previous_context @@ -8191,12 +8191,8 @@ async fn build_settings_update_items_emits_realtime_start_when_session_becomes_l .await; current_context.realtime_active = true; - let update_items = session - .build_settings_update_items( - Some(&previous_context.to_turn_context_item()), - ¤t_context, - ) - .await; + let update_items = + record_context_update_items(&session, previous_context, current_context).await; let developer_texts = developer_input_texts(&update_items); assert!( @@ -8208,7 +8204,7 @@ async fn build_settings_update_items_emits_realtime_start_when_session_becomes_l } #[tokio::test] -async fn build_settings_update_items_emits_realtime_end_when_session_stops_being_live() { +async fn record_context_updates_emits_realtime_end_when_session_stops_being_live() { let (session, mut previous_context) = make_session_and_context().await; previous_context.realtime_active = true; let mut current_context = previous_context @@ -8219,12 +8215,8 @@ async fn build_settings_update_items_emits_realtime_end_when_session_stops_being .await; current_context.realtime_active = false; - let update_items = session - .build_settings_update_items( - Some(&previous_context.to_turn_context_item()), - ¤t_context, - ) - .await; + let update_items = + record_context_update_items(&session, Arc::new(previous_context), current_context).await; let developer_texts = developer_input_texts(&update_items); assert!( @@ -8236,41 +8228,7 @@ async fn build_settings_update_items_emits_realtime_end_when_session_stops_being } #[tokio::test] -async fn build_settings_update_items_uses_previous_turn_settings_for_realtime_end() { - let (session, previous_context) = make_session_and_context().await; - let mut previous_context_item = previous_context.to_turn_context_item(); - previous_context_item.realtime_active = None; - let previous_turn_settings = PreviousTurnSettings { - model: previous_context.model_info.slug.clone(), - comp_hash: None, - realtime_active: Some(true), - }; - let mut current_context = previous_context - .with_model( - previous_context.model_info.slug.clone(), - &session.services.models_manager, - ) - .await; - current_context.realtime_active = false; - - session - .set_previous_turn_settings(Some(previous_turn_settings)) - .await; - let update_items = session - .build_settings_update_items(Some(&previous_context_item), ¤t_context) - .await; - - let developer_texts = developer_input_texts(&update_items); - assert!( - developer_texts - .iter() - .any(|text| text.contains("Reason: inactive")), - "expected a realtime end update from previous turn settings, got {developer_texts:?}" - ); -} - -#[tokio::test] -async fn build_initial_context_uses_previous_realtime_state() { +async fn build_initial_context_describes_active_realtime_state() { let (session, mut turn_context) = make_session_and_context().await; turn_context.realtime_active = true; let turn_context = Arc::new(turn_context); @@ -8283,20 +8241,6 @@ async fn build_initial_context_uses_previous_realtime_state() { .any(|text| text.contains("")), "expected initial context to describe active realtime state, got {developer_texts:?}" ); - - let previous_context_item = turn_context.to_turn_context_item(); - { - let mut state = session.state.lock().await; - state.set_reference_context_item(Some(previous_context_item)); - } - let resumed_context = build_initial_context(&session, &turn_context).await; - let resumed_developer_texts = developer_input_texts(&resumed_context); - assert!( - !resumed_developer_texts - .iter() - .any(|text| text.contains("")), - "did not expect a duplicate realtime update, got {resumed_developer_texts:?}" - ); } async fn make_multi_agent_v2_usage_hint_test_session( @@ -8800,29 +8744,6 @@ async fn build_initial_context_emits_thread_start_skill_warning_on_repeated_buil )); } -#[tokio::test] -async fn build_initial_context_uses_previous_turn_settings_for_realtime_end() { - let (session, turn_context) = make_session_and_context().await; - let previous_turn_settings = PreviousTurnSettings { - model: turn_context.model_info.slug.clone(), - comp_hash: None, - realtime_active: Some(true), - }; - - session - .set_previous_turn_settings(Some(previous_turn_settings)) - .await; - let turn_context = Arc::new(turn_context); - let initial_context = build_initial_context(&session, &turn_context).await; - let developer_texts = developer_input_texts(&initial_context); - assert!( - developer_texts - .iter() - .any(|text| text.contains("Reason: inactive")), - "expected initial context to describe an ended realtime session, got {developer_texts:?}" - ); -} - #[tokio::test] async fn build_initial_context_restates_realtime_start_when_reference_context_is_missing() { let (session, mut turn_context) = make_session_and_context().await; diff --git a/codex-rs/core/src/session/world_state.rs b/codex-rs/core/src/session/world_state.rs index a00d4dbf35..636150c767 100644 --- a/codex-rs/core/src/session/world_state.rs +++ b/codex-rs/core/src/session/world_state.rs @@ -7,6 +7,7 @@ use crate::context::world_state::CollaborationModeState; use crate::context::world_state::EnvironmentsInstructionsState; use crate::context::world_state::EnvironmentsState; use crate::context::world_state::PluginsInstructionsState; +use crate::context::world_state::RealtimeState; use crate::context::world_state::WorldState; use codex_extension_api::WorldStateContributionInput; use codex_features::Feature; @@ -30,8 +31,14 @@ impl Session { } else { String::new() }; - let mut world_state = WorldState::default(); + world_state.add_section(RealtimeState::new( + turn_context.realtime_active, + turn_context + .config + .experimental_realtime_start_instructions + .as_deref(), + )); world_state.add_section(AgentsMdState::new(step_context.loaded_agents_md.as_deref())); if turn_context.config.include_collaboration_mode_instructions && let Some(collaboration_mode) = diff --git a/codex-rs/core/tests/suite/compact_remote.rs b/codex-rs/core/tests/suite/compact_remote.rs index 43caddf0ee..885f8c61f4 100644 --- a/codex-rs/core/tests/suite/compact_remote.rs +++ b/codex-rs/core/tests/suite/compact_remote.rs @@ -2942,95 +2942,74 @@ async fn remote_request_uses_custom_experimental_realtime_start_instructions() - } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn snapshot_request_shape_remote_pre_turn_compaction_restates_realtime_end() -> Result<()> { +async fn active_realtime_does_not_diff_changed_start_instructions_after_resume() -> Result<()> { skip_if_no_network!(Ok(())); let server = wiremock::MockServer::start().await; - let realtime_server = start_remote_realtime_server().await; - let mut builder = remote_realtime_test_codex_builder(&realtime_server).with_config(|config| { - config.model_auto_compact_token_limit = Some(200); - }); - let test = builder.build(&server).await?; - + let initial_realtime_server = start_remote_realtime_server().await; + let initial_instructions = "initial custom realtime start instructions"; + let mut initial_builder = remote_realtime_test_codex_builder(&initial_realtime_server) + .with_config({ + let initial_instructions = initial_instructions.to_string(); + move |config| { + config.experimental_realtime_start_instructions = Some(initial_instructions); + } + }); + let initial = initial_builder.build(&server).await?; + let home = initial.home.clone(); + let rollout_path = initial + .session_configured + .rollout_path + .clone() + .expect("rollout path"); let responses_mock = responses::mount_sse_sequence( &server, vec![ - responses::sse(vec![ - responses::ev_assistant_message("m1", "REMOTE_FIRST_REPLY"), - responses::ev_completed_with_tokens("r1", /*total_tokens*/ 500), - ]), - responses::sse(vec![ - responses::ev_assistant_message("m2", "REMOTE_SECOND_REPLY"), - responses::ev_completed_with_tokens("r2", /*total_tokens*/ 80), - ]), + responses::sse(vec![responses::ev_completed("r1")]), + responses::sse(vec![responses::ev_completed("r2")]), ], ) .await; - let compact_mock = responses::mount_compact_json_once( - &server, - serde_json::json!({ - "output": compacted_summary_only_output( - "REMOTE_PRETURN_REALTIME_CLOSED_SUMMARY" - ) - }), - ) + + start_realtime_conversation(initial.codex.as_ref()).await?; + initial.submit_turn("USER_ONE").await?; + close_realtime_conversation(initial.codex.as_ref()).await?; + initial.codex.submit(Op::Shutdown).await?; + wait_for_event(&initial.codex, |ev| { + matches!(ev, EventMsg::ShutdownComplete) + }) .await; + initial_realtime_server.shutdown().await; - start_realtime_conversation(test.codex.as_ref()).await?; + let resumed_realtime_server = start_remote_realtime_server().await; + let changed_instructions = "changed custom realtime start instructions"; + let mut resume_builder = remote_realtime_test_codex_builder(&resumed_realtime_server) + .with_config({ + let changed_instructions = changed_instructions.to_string(); + move |config| { + config.experimental_realtime_start_instructions = Some(changed_instructions); + } + }); + let resumed = resume_builder.resume(&server, home, rollout_path).await?; - test.codex - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: "USER_ONE".to_string(), - 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, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; + start_realtime_conversation(resumed.codex.as_ref()).await?; + resumed.submit_turn("USER_TWO").await?; - close_realtime_conversation(test.codex.as_ref()).await?; - - test.codex - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: "USER_TWO".to_string(), - 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, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - - assert_eq!(compact_mock.requests().len(), 1); let requests = responses_mock.requests(); - assert_eq!(requests.len(), 2, "expected two model requests"); - - let compact_request = compact_mock.single_request(); - let post_compact_request = &requests[1]; - assert_request_contains_realtime_end(post_compact_request); - - insta::assert_snapshot!( - "remote_pre_turn_compaction_restates_realtime_end_shapes", - format_labeled_requests_snapshot( - "Remote pre-turn auto-compaction after realtime was closed between turns: the follow-up request emits realtime-end instructions from previous-turn settings even though compaction cleared the reference baseline.", - &[ - ("Remote Compaction Request", &compact_request), - ( - "Remote Post-Compaction History Layout", - post_compact_request - ), - ] - ) + assert_eq!(requests.len(), 2, "expected one request per session"); + assert_request_contains_custom_realtime_start(&requests[0], initial_instructions); + let resumed_body = requests[1].body_json().to_string(); + assert!( + resumed_body.contains(initial_instructions), + "expected resumed history to retain the original realtime instructions" + ); + assert!( + !resumed_body.contains(changed_instructions), + "did not expect an active-to-active instruction change to emit a diff" ); - realtime_server.shutdown().await; + close_realtime_conversation(resumed.codex.as_ref()).await?; + resumed_realtime_server.shutdown().await; Ok(()) } @@ -3234,115 +3213,6 @@ async fn snapshot_request_shape_remote_mid_turn_compaction_does_not_restate_real Ok(()) } -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn snapshot_request_shape_remote_compact_resume_restates_realtime_end() -> Result<()> { - skip_if_no_network!(Ok(())); - - let server = wiremock::MockServer::start().await; - let realtime_server = start_remote_realtime_server().await; - let mut builder = remote_realtime_test_codex_builder(&realtime_server); - let initial = builder.build(&server).await?; - let home = initial.home.clone(); - let rollout_path = initial - .session_configured - .rollout_path - .clone() - .expect("rollout path"); - - let responses_mock = responses::mount_sse_sequence( - &server, - vec![ - responses::sse(vec![ - responses::ev_assistant_message("m1", "REMOTE_FIRST_REPLY"), - responses::ev_completed_with_tokens("r1", /*total_tokens*/ 60), - ]), - responses::sse(vec![ - responses::ev_assistant_message("m2", "REMOTE_AFTER_RESUME_REPLY"), - responses::ev_completed_with_tokens("r2", /*total_tokens*/ 80), - ]), - ], - ) - .await; - let compact_mock = responses::mount_compact_json_once( - &server, - serde_json::json!({ - "output": compacted_summary_only_output( - "REMOTE_RESUME_REALTIME_CLOSED_SUMMARY" - ) - }), - ) - .await; - - start_realtime_conversation(initial.codex.as_ref()).await?; - - initial - .codex - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: "USER_ONE".to_string(), - 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(&initial.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - - close_realtime_conversation(initial.codex.as_ref()).await?; - - initial.codex.submit(Op::Compact).await?; - wait_for_event(&initial.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - - initial.codex.submit(Op::Shutdown).await?; - wait_for_event(&initial.codex, |ev| { - matches!(ev, EventMsg::ShutdownComplete) - }) - .await; - - let mut resume_builder = - test_codex().with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()); - let resumed = resume_builder.resume(&server, home, rollout_path).await?; - - resumed - .codex - .submit(Op::UserInput { - items: vec![UserInput::Text { - text: "USER_TWO".to_string(), - 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(&resumed.codex, |ev| matches!(ev, EventMsg::TurnComplete(_))).await; - - assert_eq!(compact_mock.requests().len(), 1); - let requests = responses_mock.requests(); - assert_eq!(requests.len(), 2, "expected two model requests"); - - let compact_request = compact_mock.single_request(); - let after_resume_request = &requests[1]; - assert_request_contains_realtime_end(after_resume_request); - - insta::assert_snapshot!( - "remote_compact_resume_restates_realtime_end_shapes", - format_labeled_requests_snapshot( - "After remote manual /compact and resume, the first resumed turn rebuilds history from the compaction item and restates realtime-end instructions from reconstructed previous-turn settings.", - &[ - ("Remote Compaction Request", &compact_request), - ("Remote Post-Resume History Layout", after_resume_request), - ] - ) - ); - - realtime_server.shutdown().await; - Ok(()) -} - #[tokio::test(flavor = "multi_thread", worker_threads = 2)] // TODO(ccunningham): Update once remote pre-turn compaction includes incoming user input. async fn snapshot_request_shape_remote_pre_turn_compaction_including_incoming_user_message() diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_compact_resume_restates_realtime_end_shapes.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_compact_resume_restates_realtime_end_shapes.snap deleted file mode 100644 index 0289370633..0000000000 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_compact_resume_restates_realtime_end_shapes.snap +++ /dev/null @@ -1,21 +0,0 @@ ---- -source: core/tests/suite/compact_remote.rs -expression: "format_labeled_requests_snapshot(\"After remote manual /compact and resume, the first resumed turn rebuilds history from the compaction item and restates realtime-end instructions from reconstructed previous-turn settings.\",\n&[(\"Remote Compaction Request\", &compact_request),\n(\"Remote Post-Resume History Layout\", after_resume_request),])" ---- -Scenario: After remote manual /compact and resume, the first resumed turn rebuilds history from the compaction item and restates realtime-end instructions from reconstructed previous-turn settings. - -## Remote Compaction Request -00:message/developer[2]: - [01] - [02] \nRealtime conversation started.\n\nYou a... -01:message/user:> -02:message/user:USER_ONE -03:message/assistant:REMOTE_FIRST_REPLY - -## Remote Post-Resume History Layout -00:compaction:encrypted=true -01:message/developer[2]: - [01] - [02] \nRealtime conversation ended.\n\nSubsequ... -02:message/user:> -03:message/user:USER_TWO diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_pre_turn_compaction_restates_realtime_end_shapes.snap b/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_pre_turn_compaction_restates_realtime_end_shapes.snap deleted file mode 100644 index c00b9dcce8..0000000000 --- a/codex-rs/core/tests/suite/snapshots/all__suite__compact_remote__remote_pre_turn_compaction_restates_realtime_end_shapes.snap +++ /dev/null @@ -1,21 +0,0 @@ ---- -source: core/tests/suite/compact_remote.rs -expression: "format_labeled_requests_snapshot(\"Remote pre-turn auto-compaction after realtime was closed between turns: the follow-up request emits realtime-end instructions from previous-turn settings even though compaction cleared the reference baseline.\",\n&[(\"Remote Compaction Request\", &compact_request),\n(\"Remote Post-Compaction History Layout\", post_compact_request),])" ---- -Scenario: Remote pre-turn auto-compaction after realtime was closed between turns: the follow-up request emits realtime-end instructions from previous-turn settings even though compaction cleared the reference baseline. - -## Remote Compaction Request -00:message/developer[2]: - [01] - [02] \nRealtime conversation started.\n\nYou a... -01:message/user:> -02:message/user:USER_ONE -03:message/assistant:REMOTE_FIRST_REPLY - -## Remote Post-Compaction History Layout -00:compaction:encrypted=true -01:message/developer[2]: - [01] - [02] \nRealtime conversation ended.\n\nSubsequ... -02:message/user:> -03:message/user:USER_TWO