From e2eea071405a4d312ca9eabeed91b7e7cb9685c3 Mon Sep 17 00:00:00 2001 From: rhan-oai Date: Tue, 18 Aug 2026 07:20:58 +0000 Subject: [PATCH] Refresh collaboration instructions when their content changes (#39163) ## Why Collaboration guidance can change without changing the active mode or model, such as after a model catalog refresh. Tracking only the mode and model leaves the earlier guidance in retained history. ## What changed - Include a hash of the rendered collaboration instructions in the world-state snapshot so changed guidance is appended to retained history once. - Emit one empty collaboration fragment when guidance is removed, preventing stale instructions from remaining active without repeatedly clearing them. - Refresh legacy snapshots that lack an instruction hash once on resume. ## Testing Cover same-mode catalog updates, clearing missing or explicitly empty guidance, retained-history behavior, and cold resume from legacy snapshots. GitOrigin-RevId: 69f690adeccd1fb9794a6b15647e6efae3fec4f6 --- .../context/world_state/collaboration_mode.rs | 75 +++-- .../world_state/collaboration_mode_tests.rs | 109 +++--- ..._collaboration_mode__tests__snapshots.snap | 14 +- .../tests/suite/collaboration_instructions.rs | 318 +++++++++++++++--- 4 files changed, 387 insertions(+), 129 deletions(-) diff --git a/codex-rs/core/src/context/world_state/collaboration_mode.rs b/codex-rs/core/src/context/world_state/collaboration_mode.rs index f06a2d5e6a..b0c1611505 100644 --- a/codex-rs/core/src/context/world_state/collaboration_mode.rs +++ b/codex-rs/core/src/context/world_state/collaboration_mode.rs @@ -1,4 +1,5 @@ use super::PreviousSectionState; +use super::WorldStateHash; use super::WorldStateSection; use crate::context::ContextualUserFragment; use codex_protocol::config_types::CollaborationMode; @@ -12,8 +13,7 @@ use serde::Serialize; /// Collaboration-mode instructions currently visible to the model. #[derive(Clone, Debug)] pub(crate) struct CollaborationModeState { - mode: ModeKind, - model: String, + snapshot: CollaborationModeSnapshot, instructions: Option, } @@ -28,16 +28,25 @@ impl CollaborationModeState { ModeKind::Plan => messages.plan.as_ref(), }); - Self { + let instructions = catalog_instructions.cloned().or_else(|| { + collaboration_mode + .settings + .developer_instructions + .clone() + .filter(|instructions| !instructions.is_empty()) + }); + // Keep an empty-state snapshot so removing instructions clears retained history only once. + let fragment = CollaborationModeInstructions { + instructions: instructions.clone().unwrap_or_default(), + }; + let snapshot = CollaborationModeSnapshot::Current { mode: collaboration_mode.mode, model: collaboration_mode.settings.model.clone(), - instructions: catalog_instructions.cloned().or_else(|| { - collaboration_mode - .settings - .developer_instructions - .clone() - .filter(|instructions| !instructions.is_empty()) - }), + instructions: Some(WorldStateHash::from_fragment(&fragment)), + }; + Self { + snapshot, + instructions, } } } @@ -45,7 +54,13 @@ impl CollaborationModeState { #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[serde(untagged)] pub(crate) enum CollaborationModeSnapshot { - Current { mode: ModeKind, model: String }, + Current { + mode: ModeKind, + model: String, + // Older object snapshots do not have an instruction hash. + #[serde(default, skip_serializing_if = "Option::is_none")] + instructions: Option, + }, Legacy(ModeKind), } @@ -54,14 +69,7 @@ impl WorldStateSection for CollaborationModeState { type Snapshot = CollaborationModeSnapshot; fn snapshot(&self) -> Self::Snapshot { - CollaborationModeSnapshot::Current { - mode: self.mode, - model: self.model.clone(), - } - } - - fn should_persist(&self) -> bool { - self.instructions.is_some() + self.snapshot.clone() } fn matches_legacy_fragment(role: &str, text: &str) -> bool { @@ -80,13 +88,28 @@ impl WorldStateSection for CollaborationModeState { &self, previous: PreviousSectionState<'_, Self::Snapshot>, ) -> Option> { - if matches!( - previous, - PreviousSectionState::Known(CollaborationModeSnapshot::Current { mode, model }) - if *mode == self.mode && model == &self.model - ) || matches!(previous, PreviousSectionState::Unknown) - || (self.instructions.is_none() && matches!(previous, PreviousSectionState::Absent)) - { + let unchanged = match previous { + PreviousSectionState::Absent => self.instructions.is_none(), + PreviousSectionState::Unknown => false, + PreviousSectionState::Known(previous) => { + previous == &self.snapshot + || (self.instructions.as_deref().is_none_or(str::is_empty) + && matches!( + (previous, &self.snapshot), + ( + CollaborationModeSnapshot::Current { + instructions: Some(previous_hash), + .. + }, + CollaborationModeSnapshot::Current { + instructions: Some(current_hash), + .. + }, + ) if previous_hash == current_hash + )) + } + }; + if unchanged { return None; } diff --git a/codex-rs/core/src/context/world_state/collaboration_mode_tests.rs b/codex-rs/core/src/context/world_state/collaboration_mode_tests.rs index 02e2771be7..ece7d8234b 100644 --- a/codex-rs/core/src/context/world_state/collaboration_mode_tests.rs +++ b/codex-rs/core/src/context/world_state/collaboration_mode_tests.rs @@ -30,29 +30,44 @@ fn snapshots() { } #[test] -fn persisted_instructions_are_restored_only_when_missing_from_history() { - let state = collaboration_mode_state(ModeKind::Default, "pair with the user"); - let retained: ResponseItem = ContextualUserFragment::into(CollaborationModeInstructions { - instructions: state.instructions.clone().expect("test instructions"), - }); - let mut world_state = WorldState::default(); - world_state.add_section(state); - let snapshot = world_state.snapshot(); +fn instruction_updates_are_applied_once_in_retained_history() { + let mut history: Vec = vec![ContextualUserFragment::into( + CollaborationModeInstructions { + instructions: "old instructions".to_string(), + }, + )]; + let mut previous = None; - assert!( - world_state - .render_history_diff(/*previous*/ None, std::slice::from_ref(&retained)) - .is_empty() - ); - assert_eq!( - world_state.render_history_diff(Some(&snapshot), &[]).len(), - 1, - ); - assert!( - world_state - .render_history_diff(Some(&snapshot), &[retained]) - .is_empty() - ); + for instructions in [Some("old instructions"), Some("new instructions"), None] { + let mut world_state = WorldState::default(); + world_state.add_section(CollaborationModeState::from_collaboration_mode( + &collaboration_mode(ModeKind::Default, instructions), + /*catalog_messages*/ None, + )); + let expected: ResponseItem = ContextualUserFragment::into(CollaborationModeInstructions { + instructions: instructions.unwrap_or_default().to_string(), + }); + let updates = world_state + .render_history_diff(previous.as_ref(), &history) + .into_iter() + .map(ContextualUserFragment::into_boxed_response_item) + .collect::>(); + assert_eq!(updates, vec![expected]); + history.extend(updates); + previous = Some(world_state.snapshot()); + + assert!( + world_state + .render_history_diff(previous.as_ref(), &history) + .is_empty() + ); + assert_eq!( + world_state + .render_history_diff(previous.as_ref(), &[]) + .len(), + usize::from(instructions.is_some()), + ); + } } #[test] @@ -114,31 +129,35 @@ fn missing_catalog_collaboration_message_uses_legacy_instructions() { #[test] fn legacy_collaboration_mode_snapshots_refresh_catalog_messages_once() { - let previous = serde_json::from_str::("\"default\"") - .expect("legacy collaboration mode snapshot"); + for serialized in ["\"default\"", r#"{"mode":"default","model":"test-model"}"#] { + let previous = serde_json::from_str::(serialized) + .expect("legacy collaboration mode snapshot"); - for instructions in ["catalog instructions", ""] { - let messages = CollaborationModeMessages { - default: Some(instructions.to_string()), - plan: None, - }; - let state = CollaborationModeState::from_collaboration_mode( - &collaboration_mode(ModeKind::Default, Some("stale legacy instructions")), - Some(&messages), - ); + for instructions in ["catalog instructions", ""] { + let messages = CollaborationModeMessages { + default: Some(instructions.to_string()), + plan: None, + }; + let state = CollaborationModeState::from_collaboration_mode( + &collaboration_mode(ModeKind::Default, Some("stale legacy instructions")), + Some(&messages), + ); - assert_eq!( - state - .render_diff(PreviousSectionState::Known(&previous)) - .expect("legacy snapshot should refresh collaboration instructions") - .render(), - format!("{COLLABORATION_MODE_OPEN_TAG}{instructions}{COLLABORATION_MODE_CLOSE_TAG}") - ); - assert!( - state - .render_diff(PreviousSectionState::Known(&state.snapshot())) - .is_none() - ); + assert_eq!( + state + .render_diff(PreviousSectionState::Known(&previous)) + .expect("legacy snapshot should refresh collaboration instructions") + .render(), + format!( + "{COLLABORATION_MODE_OPEN_TAG}{instructions}{COLLABORATION_MODE_CLOSE_TAG}" + ) + ); + assert!( + state + .render_diff(PreviousSectionState::Known(&state.snapshot())) + .is_none() + ); + } } } diff --git a/codex-rs/core/src/context/world_state/snapshots/codex_core__context__world_state__collaboration_mode__tests__snapshots.snap b/codex-rs/core/src/context/world_state/snapshots/codex_core__context__world_state__collaboration_mode__tests__snapshots.snap index 406a3aad66..2e8eac75cf 100644 --- a/codex-rs/core/src/context/world_state/snapshots/codex_core__context__world_state__collaboration_mode__tests__snapshots.snap +++ b/codex-rs/core/src/context/world_state/snapshots/codex_core__context__world_state__collaboration_mode__tests__snapshots.snap @@ -5,17 +5,17 @@ expression: "render_section_cases(&[(Absent, Absent), (Absent, Known(&default)), Absent -> Absent None -Absent -> {"mode":"default","model":"test-model"} (role - developer) +Absent -> {"instructions":"90f8f6675171813762488f76c60b062fc8adeaec","mode":"default","model":"test-model"} (role - developer) pair with the user -{"mode":"default","model":"test-model"} -> {"mode":"default","model":"test-model"} +{"instructions":"90f8f6675171813762488f76c60b062fc8adeaec","mode":"default","model":"test-model"} -> {"instructions":"90f8f6675171813762488f76c60b062fc8adeaec","mode":"default","model":"test-model"} None -{"mode":"default","model":"test-model"} -> {"mode":"default","model":"test-model"} -None +{"instructions":"1bfc0435f0536e09f08dbfe37c09920d3344e925","mode":"default","model":"test-model"} -> {"instructions":"8c3a0740268ef8f622810f31715b1924540de415","mode":"default","model":"test-model"} (role - developer) +new instructions -{"mode":"default","model":"test-model"} -> {"mode":"plan","model":"test-model"} (role - developer) +{"instructions":"90f8f6675171813762488f76c60b062fc8adeaec","mode":"default","model":"test-model"} -> {"instructions":"bc50ed412c1138502df45ef1b6634a6ef618da4c","mode":"plan","model":"test-model"} (role - developer) make a plan -Unknown -> {"mode":"default","model":"test-model"} -None +Unknown -> {"instructions":"90f8f6675171813762488f76c60b062fc8adeaec","mode":"default","model":"test-model"} (role - developer) +pair with the user diff --git a/codex-rs/core/tests/suite/collaboration_instructions.rs b/codex-rs/core/tests/suite/collaboration_instructions.rs index 3264bb8611..c7a2081f66 100644 --- a/codex-rs/core/tests/suite/collaboration_instructions.rs +++ b/codex-rs/core/tests/suite/collaboration_instructions.rs @@ -1,5 +1,9 @@ use anyhow::Result; use codex_core::TurnInputRequest; +use codex_features::Feature; +use codex_history::RolloutItem; +use codex_history::RolloutLine; +use codex_login::CodexAuth; use codex_models_manager::model_info::model_info_from_slug; use codex_protocol::config_types::CollaborationMode; use codex_protocol::config_types::ModeKind; @@ -12,10 +16,14 @@ use codex_protocol::protocol::COLLABORATION_MODE_OPEN_TAG; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::user_input::UserInput; +use core_test_support::responses::ev_assistant_message; use core_test_support::responses::ev_completed; use core_test_support::responses::ev_response_created; +use core_test_support::responses::mount_models_once_with_etag; +use core_test_support::responses::mount_response_once; use core_test_support::responses::mount_sse_once; use core_test_support::responses::sse; +use core_test_support::responses::sse_response; use core_test_support::responses::start_mock_server; use core_test_support::skip_if_no_network; use core_test_support::test_codex::local_selections; @@ -23,6 +31,9 @@ use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; use pretty_assertions::assert_eq; use serde_json::Value; +use serde_json::json; +use test_case::test_case; +use wiremock::MockServer; fn collab_mode_with_mode_and_instructions( mode: ModeKind, @@ -223,70 +234,177 @@ async fn catalog_collaboration_messages_track_mode_changes() -> Result<()> { Ok(()) } +#[test_case(ModeKind::Default; "default")] +#[test_case(ModeKind::Plan; "plan")] #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn missing_catalog_and_legacy_collaboration_message_clears_prior_instructions() -> Result<()> -{ +async fn catalog_collaboration_messages_refresh_without_mode_or_model_change( + mode: ModeKind, +) -> Result<()> { + skip_if_no_network!(Ok(())); + + const ETAG_1: &str = "\"collaboration-models-1\""; + const ETAG_2: &str = "\"collaboration-models-2\""; + const ETAG_3: &str = "\"collaboration-models-3\""; + const MODEL: &str = "catalog-collaboration-refresh-model"; + const ORIGINAL: &str = "original catalog collaboration instructions"; + const UPDATED: &str = "updated catalog collaboration instructions"; + const INACTIVE: &str = "inactive mode instructions"; + + let catalog = |instructions: &str| ModelsResponse { + models: vec![match mode { + ModeKind::Default => { + model_with_collaboration_messages(MODEL, Some(instructions), Some(INACTIVE)) + } + ModeKind::Plan => { + model_with_collaboration_messages(MODEL, Some(INACTIVE), Some(instructions)) + } + }], + }; + let server = MockServer::start().await; + let mut models_mocks = + vec![mount_models_once_with_etag(&server, catalog(ORIGINAL), ETAG_1).await]; + let test = test_codex() + .with_auth(CodexAuth::create_dummy_chatgpt_auth_for_testing()) + .with_model(MODEL) + .with_config(|config| { + config.model_provider.request_max_retries = Some(0); + config.model_provider.stream_max_retries = Some(1); + config + .features + .disable(Feature::Apps) + .expect("test config should allow feature update"); + }) + .build_with_auto_env(&server) + .await?; + + core_test_support::submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + collaboration_mode: Some(collab_mode_for_model( + mode, + MODEL, + Some("legacy fallback instructions"), + )), + ..Default::default() + }, + ) + .await?; + + let history = [ORIGINAL, UPDATED, ""]; + let mut requests = Vec::new(); + for (turn, etag, refreshed_instructions, expected) in [ + ("original", ETAG_2, Some(UPDATED), &history[..1]), + ("updated", ETAG_2, None, &history[..2]), + ("unchanged", ETAG_3, Some(""), &history[..2]), + ("cleared", ETAG_3, None, &history[..]), + ("still-cleared", ETAG_3, None, &history[..]), + ] { + if let Some(instructions) = refreshed_instructions { + models_mocks + .push(mount_models_once_with_etag(&server, catalog(instructions), etag).await); + } + let response = mount_response_once( + &server, + sse_response(sse(vec![ev_response_created(turn), ev_completed(turn)])) + .insert_header("X-Models-Etag", etag), + ) + .await; + + test.submit_text_turn(turn).await?; + let request = response.single_request(); + let dev_texts = request.message_input_texts("developer"); + let collaboration_instructions = dev_texts + .iter() + .flat_map(|text| text.split(COLLABORATION_MODE_OPEN_TAG).skip(1)) + .map(|text| { + text.split_once(COLLABORATION_MODE_CLOSE_TAG) + .expect("collaboration fragment should have a closing tag") + .0 + }) + .collect::>(); + assert_eq!(collaboration_instructions.as_slice(), expected, "{turn}"); + requests.push(request); + } + + assert_eq!( + models_mocks + .iter() + .map(|mock| mock.requests().len()) + .collect::>(), + [1, 1, 1] + ); + for pair in requests.windows(2) { + let previous_input = pair[0].input(); + assert_eq!( + pair[1].input().get(..previous_input.len()), + Some(previous_input.as_slice()) + ); + assert_eq!(pair[0].instructions_text(), pair[1].instructions_text()); + } + + Ok(()) +} + +#[test_case(None; "missing instructions")] +#[test_case(Some(""); "explicit empty suppresses fallback")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn empty_collaboration_guidance_clears_prior_instructions_once( + empty_instructions: Option<&str>, +) -> Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; - let _req1 = mount_sse_once( - &server, - sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), - ) - .await; - let req2 = mount_sse_once( - &server, - sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]), - ) - .await; - let model_slug = "catalog-collaboration-clear-model"; + let empty_model_slug = "catalog-collaboration-empty-model"; let default_text = "catalog default instructions"; + let fallback_text = "legacy fallback instructions"; + let fallback = empty_instructions.map(|_| fallback_text); let model = - model_with_collaboration_messages(model_slug, Some(default_text), /*plan*/ None); + model_with_collaboration_messages(model_slug, Some(default_text), empty_instructions); + let empty_model = + model_with_collaboration_messages(empty_model_slug, empty_instructions, empty_instructions); let mut builder = test_codex() .with_model(model_slug) .with_config(move |config| { config.model_catalog = Some(ModelsResponse { - models: vec![model], + models: vec![model, empty_model], }); }); - let test = builder.build(&server).await?; + let test = builder.build_with_auto_env(&server).await?; - core_test_support::submit_thread_settings( - &test.codex, - ThreadSettingsOverrides { - collaboration_mode: Some(collab_mode_for_model( - ModeKind::Default, - model_slug, - /*instructions*/ None, - )), - ..Default::default() - }, - ) - .await?; - test.submit_text_turn("default turn").await?; + for (turn, mode, model, clear_count) in [ + ("default", ModeKind::Default, model_slug, 0), + ("clear", ModeKind::Plan, model_slug, 1), + ("still-clear", ModeKind::Plan, model_slug, 1), + ("change-model", ModeKind::Plan, empty_model_slug, 1), + ("change-mode", ModeKind::Default, empty_model_slug, 1), + ] { + let response = mount_sse_once( + &server, + sse(vec![ev_response_created(turn), ev_completed(turn)]), + ) + .await; + core_test_support::submit_thread_settings( + &test.codex, + ThreadSettingsOverrides { + collaboration_mode: Some(collab_mode_for_model(mode, model, fallback)), + ..Default::default() + }, + ) + .await?; + test.submit_text_turn(turn).await?; - core_test_support::submit_thread_settings( - &test.codex, - ThreadSettingsOverrides { - collaboration_mode: Some(collab_mode_for_model( - ModeKind::Plan, - model_slug, - /*instructions*/ None, - )), - ..Default::default() - }, - ) - .await?; - test.submit_text_turn("plan turn").await?; - - let dev_texts = developer_texts(&req2.single_request().input()); - assert_eq!( - count_messages_containing(&dev_texts, &collab_xml(default_text)), - 1 - ); - assert_eq!(count_messages_containing(&dev_texts, &collab_xml("")), 1); + let dev_texts = response.single_request().message_input_texts("developer"); + assert_eq!( + ( + count_messages_containing(&dev_texts, &collab_xml(default_text)), + count_messages_containing(&dev_texts, &collab_xml("")), + count_messages_containing(&dev_texts, fallback_text), + ), + (1, clear_count, 0), + "{turn}" + ); + } Ok(()) } @@ -593,7 +711,7 @@ async fn user_turn_overrides_collaboration_instructions_after_override() -> Resu } #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn collaboration_mode_update_ignores_instruction_changes_within_same_mode() -> Result<()> { +async fn collaboration_mode_update_appends_instruction_changes_within_same_mode() -> Result<()> { skip_if_no_network!(Ok(())); let server = start_mock_server().await; @@ -651,7 +769,7 @@ async fn collaboration_mode_update_ignores_instruction_changes_within_same_mode( let first_text = collab_xml(first_text); let second_text = collab_xml(second_text); assert_eq!(count_messages_containing(&dev_texts, &first_text), 1); - assert_eq!(count_messages_containing(&dev_texts, &second_text), 0); + assert_eq!(count_messages_containing(&dev_texts, &second_text), 1); Ok(()) } @@ -910,6 +1028,104 @@ async fn resume_replays_collaboration_instructions() -> Result<()> { Ok(()) } +#[test_case(json!("default"); "mode only")] +#[test_case(json!({"mode": "default", "model": "catalog-collaboration-resume-model"}); "mode and model")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn cold_resume_refreshes_legacy_collaboration_snapshot_once( + legacy_snapshot: Value, +) -> Result<()> { + skip_if_no_network!(Ok(())); + + const MODEL: &str = "catalog-collaboration-resume-model"; + const ORIGINAL: &str = "original collaboration instructions"; + const CURRENT: &str = "current collaboration instructions"; + let builder_with_instructions = |instructions: &str| { + let model = + model_with_collaboration_messages(MODEL, Some(instructions), /*plan*/ None); + test_codex().with_model(MODEL).with_config(move |config| { + config.model_catalog = Some(ModelsResponse { + models: vec![model], + }); + }) + }; + let server = start_mock_server().await; + let initial_response = mount_sse_once( + &server, + sse(vec![ + ev_response_created("initial"), + ev_assistant_message("initial-message", "original response"), + ev_completed("initial"), + ]), + ) + .await; + let initial = builder_with_instructions(ORIGINAL) + .build_with_auto_env(&server) + .await?; + initial.submit_text_turn("before upgrading").await?; + + let home = initial.home.clone(); + let rollout_path = initial + .session_configured + .rollout_path + .clone() + .expect("initial session should have a rollout path"); + initial.codex.shutdown_and_wait().await?; + let mut rewrote_snapshot = false; + let legacy_rollout = std::fs::read_to_string(&rollout_path)? + .lines() + .map(|original_line| { + let mut line = serde_json::from_str::(original_line)?; + if let RolloutItem::WorldState(world_state) = &mut line.item + && let Some(snapshot) = world_state.state.get_mut("collaboration_mode") + { + *snapshot = legacy_snapshot.clone(); + rewrote_snapshot = true; + serde_json::to_string(&line) + } else { + Ok(original_line.to_string()) + } + }) + .collect::, _>>()? + .join("\n"); + assert!( + rewrote_snapshot, + "initial rollout should record collaboration mode" + ); + std::fs::write(&rollout_path, format!("{legacy_rollout}\n"))?; + + let resumed = builder_with_instructions(CURRENT) + .resume(&server, home, rollout_path) + .await?; + let mut previous_input = initial_response.single_request().input(); + for turn in ["first-resumed", "second-resumed"] { + let response = mount_sse_once( + &server, + sse(vec![ev_response_created(turn), ev_completed(turn)]), + ) + .await; + resumed.submit_text_turn(turn).await?; + let request = response.single_request(); + let input = request.input(); + assert_eq!( + input.get(..previous_input.len()), + Some(previous_input.as_slice()) + ); + let dev_texts = request.message_input_texts("developer"); + assert_eq!( + ( + count_messages_containing(&dev_texts, &collab_xml(ORIGINAL)), + count_messages_containing(&dev_texts, &collab_xml(CURRENT)), + count_messages_containing(&dev_texts, COLLABORATION_MODE_OPEN_TAG), + ), + (1, 1, 2), + "{turn}" + ); + previous_input = input; + } + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn empty_collaboration_instructions_are_ignored() -> Result<()> { skip_if_no_network!(Ok(()));