diff --git a/codex-rs/core/src/context/world_state/top_level_tools_tests.rs b/codex-rs/core/src/context/world_state/top_level_tools_tests.rs index 0e85b553de..b182b6226b 100644 --- a/codex-rs/core/src/context/world_state/top_level_tools_tests.rs +++ b/codex-rs/core/src/context/world_state/top_level_tools_tests.rs @@ -267,6 +267,7 @@ fn initial_catalog_preserves_order_and_unchanged_catalog_emits_nothing() { let definitions = vec![ namespace("zebra", vec![declaration("lookup")]), namespace("alpha", vec![declaration("lookup")]), + namespace("exec", vec![declaration("lookup")]), ]; let state = TopLevelToolsState::new( /*messages*/ Default::default(), @@ -293,7 +294,14 @@ fn initial_catalog_preserves_order_and_unchanged_catalog_emits_nothing() { .keys() .map(String::as_str) .collect::>(), - vec!["alpha", "alpha.lookup", "zebra", "zebra.lookup"] + vec![ + "alpha", + "alpha.lookup", + "exec", + "exec.lookup", + "zebra", + "zebra.lookup" + ] ); let snapshot: TopLevelToolsSnapshot = serde_json::from_value(saved).unwrap(); let (unchanged_snapshot, updates) = state.render_diff(PreviousSectionState::Known(&snapshot)); diff --git a/codex-rs/core/tests/suite/scenarios.rs b/codex-rs/core/tests/suite/scenarios.rs index 45dca5128e..b5b8074c8c 100644 --- a/codex-rs/core/tests/suite/scenarios.rs +++ b/codex-rs/core/tests/suite/scenarios.rs @@ -4026,6 +4026,298 @@ async fn astra_async_compaction_keeps_work_added_after_its_snapshot( Ok(()) } +#[derive(Clone, Copy)] +enum ExecCatalogRefresh { + Hybrid, + Metadata, +} + +#[test_case::test_case(ExecCatalogRefresh::Hybrid; "hybrid mode")] +#[test_case::test_case(ExecCatalogRefresh::Metadata; "exec metadata refresh")] +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn astra_async_compaction_preserves_exec_catalog(change: ExecCatalogRefresh) -> Result<()> { + use codex_models_manager::manager::RefreshStrategy; + use core_test_support::responses::mount_models_once_with_etag; + + skip_if_no_network!(Ok(())); + let (release_background, background_gate) = oneshot::channel(); + let (release_foreground, foreground_gate) = oneshot::channel(); + let (streaming, _completions) = start_streaming_sse_server(vec![ + vec![compaction_chunk( + /*gate*/ None, + vec![ + ev_assistant_message("draft", "I drafted the launch notes."), + ev_completed_with_tokens("draft-response", /*total_tokens*/ 70_000), + ], + )], + vec![compaction_chunk( + Some(background_gate), + vec![ + compaction_output("ASYNC_LAUNCH_CHECKPOINT"), + ev_completed_with_tokens("compact-response", /*total_tokens*/ 3_000), + ], + )], + vec![ + compaction_chunk( + /*gate*/ None, + vec![ + ev_response_created("team-response"), + ev_assistant_message("team-check", "I will check who is available."), + ], + ), + compaction_chunk( + Some(foreground_gate), + vec![ + ev_function_call_with_namespace("team", "collaboration", "list_agents", "{}"), + ev_completed_with_tokens("team-response", /*total_tokens*/ 75_000), + ], + ), + ], + vec![compaction_chunk( + /*gate*/ None, + vec![ + ev_assistant_message("answer", "I can finish the notes myself."), + ev_completed_with_tokens("answer-response", /*total_tokens*/ 3_000), + ], + )], + ]) + .await; + let model_server = wiremock::MockServer::start().await; + let mut catalog = bundled_models_response()?; + let model = catalog + .models + .iter_mut() + .find(|model| model.slug == "gpt-6-astra") + .expect("bundled Astra model"); + model.tool_mode = Some(ToolMode::CodeModeOnly); + model.supports_image_detail_original = false; + mount_models_once_with_etag(&model_server, catalog.clone(), "\"exec-mode-v1\"").await; + let model_catalog_url = format!("{}/models", model_server.uri()); + let (_config_server, test) = async_compaction_test( + test_codex(), + streaming.uri(), + vec![ + Feature::IncrementalTools, + Feature::IncrementalExecTools, + Feature::ViewImage, + ], + move |config| { + config + .features + .disable(Feature::UnifiedImageBudget) + .expect("use the model's image detail capability"); + config.model_catalog = None; + config.model_provider.model_catalog_url = Some(model_catalog_url.clone().into()); + config.code_mode.disable_in_process_fallback = true; + }, + ) + .await?; + test.codex.ensure_rollout_materialized().await; + + test.submit_text_turn("Draft the launch notes.").await?; + let timeout = Duration::from_secs(/*secs*/ 20); + tokio::time::timeout(timeout, streaming.wait_for_request_count(/*count*/ 2)).await?; + // Refresh the same model while the prepared result is still pending. + let model = catalog + .models + .iter_mut() + .find(|model| model.slug == "gpt-6-astra") + .expect("bundled Astra model"); + match change { + ExecCatalogRefresh::Hybrid => model.tool_mode = Some(ToolMode::CodeMode), + ExecCatalogRefresh::Metadata => { + model.supports_image_detail_original = true; + let messages = model + .model_messages + .get_or_insert_default() + .tools + .get_or_insert_default() + .code_mode + .get_or_insert_default(); + messages.exec.get_or_insert_default().description = + Some("Previous exec instructions.".to_string()); + messages.mcp_typescript_preamble = Some("type PreviousResult = string;".to_string()); + } + } + mount_models_once_with_etag(&model_server, catalog.clone(), "\"exec-mode-v2\"").await; + test.thread_manager + .get_models_manager() + .list_models( + RefreshStrategy::Online, + codex_core::test_support::default_http_client_factory(), + ) + .await; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![text( + "Check who can finish the notes.", + )])) + .await?; + tokio::time::timeout(timeout, streaming.wait_for_request_count(/*count*/ 3)).await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::ItemCompleted(event) + if matches!(&event.item, TurnItem::AgentMessage(message) if message.id == "team-check")) + }) + .await; + release_background + .send(()) + .expect("release background compaction"); + // Keep foreground work gated until the prepared result is available for optional adoption. + tokio::time::timeout( + timeout, + test.codex.wait_for_background_compaction_for_test(), + ) + .await?; + release_foreground + .send(()) + .expect("release foreground work"); + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + tokio::time::timeout( + timeout, + test.codex.wait_for_background_compaction_for_test(), + ) + .await?; + let requests = streaming + .requests() + .await + .iter() + .map(|body| serde_json::from_slice::(body)) + .collect::>>()?; + assert_eq!(requests.len(), 4); + let history = test + .codex + .load_history(/*include_archived*/ false) + .await? + .items; + let input = requests[3]["input"].as_array().expect("request input"); + let compaction = input + .iter() + .position(|item| item["encrypted_content"] == "ASYNC_LAUNCH_CHECKPOINT") + .expect("prepared summary was adopted"); + let user = input + .iter() + .position(|item| { + item["role"] == "user" + && item["content"][0]["text"] == "Check who can finish the notes." + }) + .expect("retained user message"); + let assistant = input + .iter() + .position(|item| { + item["role"] == "assistant" + && item["content"][0]["text"] == "I will check who is available." + }) + .expect("retained assistant message"); + let call = input + .iter() + .position(|item| item["type"] == "function_call" && item["call_id"] == "team") + .expect("retained tool call"); + let output = input + .iter() + .position(|item| item["type"] == "function_call_output" && item["call_id"] == "team") + .expect("retained tool output"); + assert!(user < assistant && assistant < call && call < output); + let [before_adoption, after_adoption] = [&requests[2], &requests[3]].map(|request| { + request["input"] + .as_array() + .expect("request input") + .iter() + .filter(|item| item["type"] == "additional_tools") + .map(|item| item["tools"].clone()) + .collect::>() + }); + assert_eq!(after_adoption, before_adoption); + let (exec_positions, exec_declarations): (Vec<_>, Vec<_>) = input + .iter() + .enumerate() + .filter_map(|(index, item)| { + core_test_support::responses::namespace_child_tool(item, "functions", "exec") + .map(|declaration| (index, declaration)) + }) + .unzip(); + let (adoption_index, adopted) = history + .windows(2) + .enumerate() + .find_map(|(index, pair)| match pair { + [RolloutItem::Compacted(_), RolloutItem::WorldState(baseline)] if baseline.full => { + Some((index, &baseline.state["top_level_tools"])) + } + _ => None, + }) + .expect("adopted full checkpoint"); + let changed_hashes: &[&str] = match change { + ExecCatalogRefresh::Hybrid => &["/functions.exec"], + ExecCatalogRefresh::Metadata => &["/tools/functions.exec", "/exec/Tool: tools.view_image"], + }; + for path in changed_hashes { + let live = history[..adoption_index] + .iter() + .rev() + .find_map(|item| match item { + RolloutItem::WorldState(patch) if !patch.full => { + patch.state.get("top_level_tools")?.pointer(path) + } + _ => None, + }) + .and_then(serde_json::Value::as_str) + .expect("live exec update before adoption"); + assert_eq!( + adopted.pointer(path).and_then(serde_json::Value::as_str), + Some(live), + "{path}" + ); + } + if matches!(change, ExecCatalogRefresh::Metadata) { + let initial_exec = requests[0]["input"] + .as_array() + .expect("initial input") + .iter() + .find_map(|item| { + core_test_support::responses::namespace_child_tool(item, "functions", "exec") + }) + .expect("initial exec declaration"); + assert_eq!(exec_declarations, vec![initial_exec]); + assert!(exec_positions[0] < compaction && compaction < user); + let is_view_image_update = |item: &&serde_json::Value| { + item["role"] == "developer" + && item["content"][0]["text"] + .as_str() + .is_some_and(|text| text.lines().any(|line| line == "### `view_image`")) + }; + let live_update = requests[2]["input"] + .as_array() + .expect("input before adoption") + .iter() + .find(is_view_image_update) + .expect("view_image redefinition during background compaction"); + assert!( + live_update["content"][0]["text"] + .as_str() + .expect("view_image update text") + .contains("detail?: \"high\" | \"original\"") + ); + assert_eq!( + input + .iter() + .filter(is_view_image_update) + .collect::>(), + vec![live_update] + ); + assert!(input[compaction + 1..].contains(live_update)); + return Ok(()); + } + assert_eq!(exec_declarations.len(), 2); + assert_ne!(exec_declarations[0], exec_declarations[1]); + assert!( + exec_positions[0] < compaction + && compaction < exec_positions[1] + && exec_positions[1] < user + ); + Ok(()) +} + #[derive(PartialEq)] enum PendingCompactionOutcome { Absent, diff --git a/codex-rs/core/tests/suite/scenarios_incremental_tools_resume.rs b/codex-rs/core/tests/suite/scenarios_incremental_tools_resume.rs index 9bfb5c80bb..d057713f06 100644 --- a/codex-rs/core/tests/suite/scenarios_incremental_tools_resume.rs +++ b/codex-rs/core/tests/suite/scenarios_incremental_tools_resume.rs @@ -1,11 +1,15 @@ -//! Request-history coverage for adopting incremental tools when initial context is rebuilt. +//! Request-history coverage for incremental tool modes across resume, fork, and window rebuilds. use anyhow::Result; +use codex_core::ForkSnapshot; +use codex_core::StartThreadOptions; +use codex_core::TurnInputRequest; use codex_features::Feature; use codex_login::CodexAuth; use codex_protocol::openai_models::ToolMode; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::Op; +use codex_protocol::user_input::UserInput; use core_test_support::context_snapshot; use core_test_support::context_snapshot::ContextSnapshotOptions; use core_test_support::responses; @@ -13,6 +17,8 @@ use core_test_support::skip_if_no_network; use core_test_support::test_codex::TestCodexBuilder; use core_test_support::test_codex::test_codex; use core_test_support::wait_for_event; +use core_test_support::wait_for_successful_turn; +use pretty_assertions::assert_eq; use serde_json::json; use test_case::test_case; @@ -215,3 +221,296 @@ async fn disable_incremental_tools_at_next_window() -> Result<()> { ); Ok(()) } + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn exec_tools_follow_current_settings_across_history() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = responses::start_mock_server().await; + let replies = (0..13) + .map(|index| { + let mut events = Vec::new(); + if index == 9 { + events.push(json!({"type": "response.output_item.done", "item": {"type": "compaction", "encrypted_content": "COMPACTED_HISTORY"}})); + } + events.push(responses::ev_completed(&format!("lifecycle-{index}"))); + responses::sse(events) + }) + .collect(); + let mock = responses::mount_sse_sequence(&server, replies).await; + let session_builder = |enabled, mode, planning, changed_catalog| { + builder(/*incremental_tools*/ true, /*token_budget*/ false) + .with_model_info_override("gpt-5.4", move |model| { + model.use_responses_lite = true; + model.tool_mode = Some(mode); + }) + .with_config(move |config| { + config + .features + .set_enabled(Feature::IncrementalExecTools, enabled) + .expect("set incremental exec tools"); + config.update_plan_enabled = planning; + if changed_catalog { + config.features.disable(Feature::ShellTool).unwrap(); + } + }) + }; + let initial = session_builder( + /*enabled*/ true, + ToolMode::CodeModeOnly, + /*planning*/ false, + /*changed_catalog*/ false, + ) + .build_with_auto_env(&server) + .await?; + initial + .submit_turn("Begin with an incremental exec catalog.") + .await?; + let before_updates = session_builder( + /*enabled*/ false, + ToolMode::CodeModeOnly, + /*planning*/ false, + /*changed_catalog*/ false, + ) + .restart_with_auto_env(&server, &initial) + .await?; + before_updates + .submit_turn("Disable granular updates before any catalog change.") + .await?; + let enabled = session_builder( + /*enabled*/ true, + ToolMode::CodeModeOnly, + /*planning*/ false, + /*changed_catalog*/ false, + ) + .restart_with_auto_env(&server, &before_updates) + .await?; + enabled + .submit_turn("Re-enable granular updates with the same catalog.") + .await?; + let mixed = session_builder( + /*enabled*/ true, + ToolMode::CodeModeOnly, + /*planning*/ true, + /*changed_catalog*/ true, + ) + .restart_with_auto_env(&server, &enabled) + .await?; + mixed + .submit_turn("Add planning and remove shell tools.") + .await?; + let disabled_after_changes = session_builder( + /*enabled*/ false, + ToolMode::CodeModeOnly, + /*planning*/ true, + /*changed_catalog*/ true, + ) + .restart_with_auto_env(&server, &mixed) + .await?; + disabled_after_changes + .submit_turn("Disable granular updates while retaining those catalog changes.") + .await?; + let hybrid = session_builder( + /*enabled*/ true, + ToolMode::CodeMode, + /*planning*/ false, + /*changed_catalog*/ false, + ) + .restart_with_auto_env(&server, &disabled_after_changes) + .await?; + hybrid + .submit_turn("Resume in hybrid mode with the catalog flag enabled.") + .await?; + let returned = session_builder( + /*enabled*/ false, + ToolMode::CodeModeOnly, + /*planning*/ false, + /*changed_catalog*/ false, + ) + .restart_with_auto_env(&server, &hybrid) + .await?; + returned + .submit_turn("Return to CodeModeOnly with planning still removed.") + .await?; + + returned.codex.flush_rollout().await?; + let mut fork_config = returned.config.clone(); + fork_config.update_plan_enabled = true; + let forked = returned + .thread_manager + .fork_legacy_thread( + ForkSnapshot::Interrupted, + StartThreadOptions::new(fork_config), + returned + .codex + .rollout_path() + .expect("persisted parent rollout"), + ) + .await?; + forked + .thread + .start_or_steer_turn(TurnInputRequest::user_input(vec![UserInput::Text { + text: "Fork with planning enabled and granular updates disabled.".to_string(), + text_elements: Vec::new(), + }])) + .await?; + wait_for_successful_turn(&forked.thread).await; + forked.thread.shutdown_and_wait().await?; + + let disabled = session_builder( + /*enabled*/ false, + ToolMode::CodeModeOnly, + /*planning*/ true, + /*changed_catalog*/ false, + ) + .restart_with_auto_env(&server, &returned) + .await?; + disabled + .submit_turn("Resume with planning enabled and granular updates disabled.") + .await?; + disabled.codex.submit(Op::Compact).await?; + wait_for_successful_turn(&disabled.codex).await; + disabled + .submit_turn("Rebuild the complete catalog with granular updates disabled.") + .await?; + let enabled = session_builder( + /*enabled*/ true, + ToolMode::CodeModeOnly, + /*planning*/ false, + /*changed_catalog*/ false, + ) + .restart_with_auto_env(&server, &disabled) + .await?; + enabled + .submit_turn("Enable granular updates and remove planning.") + .await?; + let resumed = session_builder( + /*enabled*/ true, + ToolMode::CodeModeOnly, + /*planning*/ true, + /*changed_catalog*/ false, + ) + .restart_with_auto_env(&server, &enabled) + .await?; + resumed + .submit_turn("Resume with planning enabled and granular updates enabled.") + .await?; + + let requests = mock.requests(); + let declarations = requests + .iter() + .map(|request| { + request + .inputs_of_type("additional_tools") + .iter() + .filter_map(|item| responses::namespace_child_tool(item, "functions", "exec")) + .cloned() + .collect::>() + }) + .collect::>(); + assert_eq!( + declarations.iter().map(Vec::len).collect::>(), + vec![ + 1, // Initial catalog. + 1, // Disabling before any delta keeps the existing declaration. + 2, // Enabling initializes a full catalog and its comparison state. + 2, // Added and removed entries use developer messages. + 2, // Disabling keeps those messages without a forced declaration. + 3, // Hybrid mode uses ordinary declarations even with the flag enabled. + 4, // Returning with the flag disabled uses a whole declaration. + 5, // Fork adds planning through a whole declaration. + 5, // Resume adds planning through the same whole declaration. + 5, // Compaction samples the retained history. + 1, // Full replacement uses the current disabled flag. + 2, // Enabling initializes a full catalog before further deltas. + 2, // Resume adds planning in a developer message. + ] + ); + for (before, after) in [(0, 1), (3, 4)] { + assert_eq!(declarations[before], declarations[after]); + assert!( + requests[after] + .input() + .starts_with(&requests[before].input()) + ); + } + assert_eq!( + declarations[0].last(), + declarations[6].last(), + "returning from hybrid restores the complete current catalog" + ); + assert_eq!( + declarations[7].last(), + declarations[8].last(), + "fork and resume publish the same complete catalog while granular updates are disabled" + ); + let has_update_plan = [6, 8].map(|index| { + declarations[index].last().expect("full exec declaration")["description"] + .as_str() + .expect("exec description") + .lines() + .any(|line| line == "### `update_plan`") + }); + assert_eq!(has_update_plan, [false, true]); + assert_eq!( + declarations[10].last(), + declarations[8].last(), + "compaction while disabled preserves the complete current catalog" + ); + assert_eq!(declarations[11].last(), declarations[0].last()); + assert_eq!(declarations[12], declarations[11]); + let exec_definition_updates = requests + .iter() + .map(|request| { + request + .message_input_texts("developer") + .into_iter() + .filter(|text| { + text.starts_with( + "Use these definitions and namespace instructions with `functions.exec`.", + ) + }) + .collect::>() + }) + .collect::>(); + assert_eq!( + exec_definition_updates + .iter() + .map(Vec::len) + .collect::>(), + vec![0, 0, 0, 1, 1, 1, 1, 1, 1, 1, 0, 0, 1] + ); + for index in [3, 12] { + let update = &exec_definition_updates[index][0]; + assert_eq!( + update + .lines() + .filter(|line| *line != "## Namespace: functions") + .filter(|line| line.starts_with("## ") || line.starts_with("### ")) + .collect::>(), + vec!["### `update_plan`"] + ); + } + assert_eq!( + exec_definition_updates[3], exec_definition_updates[4], + "disabling granular updates retains the previously published definitions" + ); + assert_eq!( + exec_definition_updates[8], exec_definition_updates[9], + "compaction receives retained definitions without repeating them" + ); + let removal_notices = requests + .iter() + .map(|request| { + request + .message_input_texts("developer") + .into_iter() + .filter(|text| text.contains("- Tool: tools.exec_command")) + .collect::>() + }) + .collect::>(); + assert_eq!(removal_notices[3].len(), 1); + assert_eq!(removal_notices[3], removal_notices[4]); + assert!(removal_notices[10].is_empty()); + assert!(removal_notices[12].is_empty()); + Ok(()) +}