diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst index f57f363805..065fefedd0 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history.rs b/codex-rs/app-server-protocol/src/protocol/thread_history.rs index b63b84895b..c087791283 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history.rs @@ -491,6 +491,7 @@ impl ThreadHistoryBuilder { | RolloutItem::RealtimeItem(_) | RolloutItem::RetainedContext(_) | RolloutItem::SecurityRiskScore(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SessionMeta(_) => {} } } diff --git a/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs b/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs index 32d61bf82c..9f6f345e12 100644 --- a/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs +++ b/codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs @@ -82,7 +82,8 @@ pub fn project_rollout_line(line: &RolloutLine) -> ThreadHistoryChangeSet { }], ..Default::default() }, - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::InterAgentCommunicationMetadata { .. } diff --git a/codex-rs/core-api/src/lib.rs b/codex-rs/core-api/src/lib.rs index 5c9fcdd84b..97cb3f90bd 100644 --- a/codex-rs/core-api/src/lib.rs +++ b/codex-rs/core-api/src/lib.rs @@ -105,6 +105,7 @@ pub use codex_extension_api::empty_extension_registry; pub use codex_features::Feature; pub use codex_features::Features; pub use codex_history::InitialHistory; +pub use codex_history::latest_dynamic_tools; pub use codex_home::CodexHomeUserInstructionsProvider; pub use codex_image_generation_extension::install as install_image_generation_extension; pub use codex_login::AuthHeaders; diff --git a/codex-rs/core/src/agent/control/spawn.rs b/codex-rs/core/src/agent/control/spawn.rs index 8b752ea199..16323358e1 100644 --- a/codex-rs/core/src/agent/control/spawn.rs +++ b/codex-rs/core/src/agent/control/spawn.rs @@ -742,7 +742,8 @@ impl LocalAgentControl { if multi_agent_version == MultiAgentVersion::V2 && config.features.enabled(Feature::MultiAgentV2DynamicTools) { - start_options.dynamic_tools = parent_thread.session.dynamic_tools().await; + start_options.dynamic_tools = + parent_thread.session.dynamic_tools().await.to_vec(); } } None diff --git a/codex-rs/core/src/agent/control_tests.rs b/codex-rs/core/src/agent/control_tests.rs index 59027230a4..01d22c744b 100644 --- a/codex-rs/core/src/agent/control_tests.rs +++ b/codex-rs/core/src/agent/control_tests.rs @@ -2614,8 +2614,8 @@ async fn spawn_agent_without_fork_from_paginated_parent_stays_fresh_and_paginate Vec::new() }; assert_eq!( - child_thread.session.dynamic_tools().await, - expected_dynamic_tools + child_thread.session.dynamic_tools().await.as_ref(), + expected_dynamic_tools.as_slice() ); assert!( !history_contains_text( @@ -2661,6 +2661,7 @@ async fn spawn_agent_fork_sanitizes_inherited_compaction_metadata() { let (parent_thread_id, parent_thread) = harness.start_paginated_thread().await; let parent_resume_metadata = codex_history::CompactionResumeMetadata { + dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some("parent-turn".to_string()), turn_attribution: Some(codex_history::TurnAttribution { @@ -3818,6 +3819,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc compaction_response_id: None, latest_token_usage_record: None, resume_metadata: Some(codex_history::CompactionResumeMetadata { + dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, turn_attribution: None, @@ -3952,6 +3954,7 @@ async fn spawn_agent_full_fork_legacy_compaction_rebuilds_child_instructions_onc assert_eq!( inherited_state, Some(codex_history::CompactionResumeMetadata { + dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, turn_attribution: None, diff --git a/codex-rs/core/src/agent/history.rs b/codex-rs/core/src/agent/history.rs index 0549c0c74d..9cba9c4260 100644 --- a/codex-rs/core/src/agent/history.rs +++ b/codex-rs/core/src/agent/history.rs @@ -82,7 +82,10 @@ fn keep_forked_rollout_item(item: &RolloutItem, preserve_context_baselines: bool RolloutItem::TurnContext(_) | RolloutItem::WorldState(_) => preserve_context_baselines, // Child threads inherit model context, not the parent's cumulative usage state. RolloutItem::TokenUsageRecord(_) => false, - RolloutItem::Compacted(_) | RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) => true, + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::Compacted(_) + | RolloutItem::EventMsg(_) + | RolloutItem::SessionMeta(_) => true, } } @@ -417,7 +420,8 @@ pub(super) async fn prepare_forked_history( true } RolloutItem::RealtimeItem(_) => false, - RolloutItem::EventMsg(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::EventMsg(_) | RolloutItem::SessionMeta(_) | RolloutItem::TurnContext(_) | RolloutItem::InterAgentCommunication(_) diff --git a/codex-rs/core/src/codex_thread.rs b/codex-rs/core/src/codex_thread.rs index f93cc4a4b2..63fef3ceed 100644 --- a/codex-rs/core/src/codex_thread.rs +++ b/codex-rs/core/src/codex_thread.rs @@ -719,6 +719,24 @@ impl CodexThread { self.session.update_settings(updates).await.map(|_| ()) } + /// Replaces tools for subsequent model steps and future turns, preserving in-flight calls. + /// An empty list removes all dynamic tools. Persistent threads require host opt-in + /// so every reader can restore successful updates on resume. + /// + /// Once queued, stopping the caller does not cancel the update. + /// A persistence error can leave the durable result indeterminate; retry the same replacement + /// before continuing. + pub async fn replace_dynamic_tools( + &self, + tools: Vec, + ) -> CodexResult<()> { + let (reply, result) = oneshot::channel(); + self.io + .submit(Op::ReplaceDynamicTools { tools, reply }) + .await?; + result.await.unwrap_or(Err(CodexErr::InternalAgentDied)) + } + /// Persists current settings without emitting a live settings event. /// /// Serializes snapshot capture and persistence with accepted settings updates. diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index c2f12e60e5..a6640ad139 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -1013,6 +1013,10 @@ pub struct Config { /// When true, session is not persisted on disk. Default to `false` pub ephemeral: bool, + /// Enable persistent dynamic tool updates only when every reader, including rollback + /// versions, can restore the updated tools and preserve them through compaction. + pub allow_persistent_dynamic_tool_updates: bool, + /// Optional extra configuration fields for the thread. pub extra_config: Option, @@ -4627,6 +4631,7 @@ impl Config { application_auth_route_config: None, history, ephemeral: ephemeral.unwrap_or_default(), + allow_persistent_dynamic_tool_updates: false, extra_config: None, bypass_hook_trust, file_opener: cfg.file_opener.unwrap_or(UriBasedFileOpener::VsCode), diff --git a/codex-rs/core/src/session/dynamic_tools_tests.rs b/codex-rs/core/src/session/dynamic_tools_tests.rs new file mode 100644 index 0000000000..0aaa535412 --- /dev/null +++ b/codex-rs/core/src/session/dynamic_tools_tests.rs @@ -0,0 +1,143 @@ +use super::*; +use pretty_assertions::assert_eq; + +#[tokio::test] +async fn dynamic_tools_updates_preserve_submission_order_after_caller_stops_waiting() { + let (mut session, turn_context, rx_event) = + make_session_and_context_with_auth_and_config_and_rx( + CodexAuth::from_api_key("Test API Key"), + Vec::new(), + |config| config.allow_persistent_dynamic_tool_updates = true, + ) + .await; + let rollout_path = + attach_thread_persistence(Arc::get_mut(&mut session).expect("unique session")).await; + let (thread, rx_sub) = make_thread_for_session( + Arc::clone(&session), + &turn_context, + rx_event, + SessionSource::Exec, + ) + .await; + let replacement = vec![DynamicToolSpec::Function( + codex_protocol::dynamic_tools::DynamicToolFunctionSpec { + name: "check_release".into(), + description: "Check the release.".into(), + input_schema: json!({"type": "object", "properties": {}}), + defer_loading: false, + }, + )]; + let persistence = thread_settings::acquire_persistence_lock(&session).await; + let mut first = Box::pin(tokio::task::unconstrained( + thread.replace_dynamic_tools(Vec::new()), + )); + assert!(futures::poll!(first.as_mut()).is_pending()); + drop(first); + let mut submissions = Box::pin(tokio::task::unconstrained(submission_loop( + Arc::clone(&session), + Arc::clone(&turn_context.config), + rx_sub, + ))); + assert!(futures::poll!(submissions.as_mut()).is_pending()); + let mut last = Box::pin(tokio::task::unconstrained( + thread.replace_dynamic_tools(replacement.clone()), + )); + assert!(futures::poll!(last.as_mut()).is_pending()); + let mut shutdown = Box::pin(tokio::task::unconstrained(thread.submit(Op::Shutdown))); + // Shutdown cannot overtake the accepted replacements while persistence is blocked. + assert!(futures::poll!(shutdown.as_mut()).is_pending()); + assert_eq!(session.dynamic_tools().await.as_ref(), &[]); + drop(persistence); + let ((), updated, stopped) = tokio::join!(submissions, last, shutdown); + updated.expect("replace dynamic tools before shutdown"); + stopped.expect("submit shutdown"); + assert_eq!( + session.dynamic_tools().await.as_ref(), + replacement.as_slice() + ); + let restored = RolloutRecorder::get_rollout_history(&rollout_path) + .await + .expect("read persisted catalog"); + let catalogs: Vec<_> = restored + .get_rollout_items() + .iter() + .filter_map(|item| match item { + RolloutItem::DynamicToolsUpdated { tools, .. } => Some(tools.clone()), + _ => None, + }) + .collect(); + assert_eq!(catalogs, vec![Vec::new(), replacement.clone()]); + assert_eq!(restored.get_dynamic_tools(), Some(replacement)); + assert!(matches!( + thread + .replace_dynamic_tools(Vec::new()) + .await + .unwrap_err() + .details(), + codex_protocol::error::CodexErrorDetails::InternalAgentDied + )); +} + +#[tokio::test] +async fn dynamic_tools_update_requires_host_opt_in_only_for_persistent_threads() { + let (mut session, _) = make_session_and_context().await; + let initial_tools = vec![DynamicToolSpec::Function( + codex_protocol::dynamic_tools::DynamicToolFunctionSpec { + name: "check_release".into(), + description: "Check the release.".into(), + input_schema: json!({"type": "object", "properties": {}}), + defer_loading: false, + }, + )]; + session + .replace_dynamic_tools(initial_tools.clone()) + .await + .unwrap(); + let path = attach_thread_persistence(&mut session).await; + let error = session.replace_dynamic_tools(Vec::new()).await.unwrap_err(); + assert!(matches!( + error.details(), + codex_protocol::error::CodexErrorDetails::UnsupportedOperation(_) + )); + assert_eq!( + session.dynamic_tools().await.as_ref(), + initial_tools.as_slice() + ); + session.flush_rollout().await.unwrap(); + let history = RolloutRecorder::get_rollout_history(&path).await.unwrap(); + assert_eq!(history.get_dynamic_tools(), Some(initial_tools)); +} + +#[tokio::test] +async fn dynamic_tools_update_keeps_catalog_when_persistence_fails() { + let (mut session, _, _rx) = make_session_and_context_with_auth_and_config_and_rx( + CodexAuth::from_api_key("Test API Key"), + Vec::new(), + |config| config.allow_persistent_dynamic_tool_updates = true, + ) + .await; + attach_thread_persistence(Arc::get_mut(&mut session).expect("unique session")).await; + session + .live_thread() + .expect("live thread") + .shutdown() + .await + .expect("close persistence"); + let replacement = vec![codex_protocol::dynamic_tools::DynamicToolSpec::Function( + codex_protocol::dynamic_tools::DynamicToolFunctionSpec { + name: "check_release".into(), + description: "Check the release.".into(), + input_schema: json!({"type": "object", "properties": {}}), + defer_loading: false, + }, + )]; + let error = session + .replace_dynamic_tools(replacement) + .await + .unwrap_err(); + assert!(matches!( + error.details(), + codex_protocol::error::CodexErrorDetails::Fatal(_) + )); + assert_eq!(session.dynamic_tools().await.as_ref(), &[]); +} diff --git a/codex-rs/core/src/session/extension_interruption_tests.rs b/codex-rs/core/src/session/extension_interruption_tests.rs index 39e038bc7a..319da2a21c 100644 --- a/codex-rs/core/src/session/extension_interruption_tests.rs +++ b/codex-rs/core/src/session/extension_interruption_tests.rs @@ -2,16 +2,10 @@ #![cfg(test)] use super::*; -use crate::session::SessionIo; -use crate::session::completed_session_loop_termination; use crate::session::tests::HeldStepTask; use crate::session::tests::make_session_and_context_with_rx; +use crate::session::tests::make_thread_for_session; use crate::state::TaskKind; -use crate::thread_startup_metadata::ThreadStartupMetadata; -use codex_protocol::config_types::ApprovalsReviewer; -use codex_protocol::protocol::AgentStatus; -use codex_protocol::protocol::AskForApproval; -use codex_protocol::protocol::SessionConfiguredEvent; use codex_protocol::protocol::SessionSource; use opentelemetry::trace::TracerProvider; use opentelemetry_sdk::trace::InMemorySpanExporter; @@ -19,7 +13,6 @@ use opentelemetry_sdk::trace::SdkTracerProvider; use pretty_assertions::assert_eq; use std::time::Duration; use tokio::sync::Notify; -use tokio::sync::watch; use tracing::Instrument; use tracing_subscriber::prelude::*; @@ -136,37 +129,13 @@ async fn interrupt_if_no_pending_input_handles_cancelled_submission() { .expect("active task") .cancellation_token .clone(); - let (tx_sub, rx_sub) = async_channel::bounded(/*cap*/ 1); - let thread = CodexThread::new( + let (thread, rx_sub) = make_thread_for_session( Arc::clone(&session), - SessionIo { - tx_sub, - rx_event, - agent_status: watch::channel(AgentStatus::PendingInit).1, - session_loop_termination: completed_session_loop_termination(), - }, - ThreadStartupMetadata::from(&SessionConfiguredEvent { - session_id: session.session_id(), - thread_id: session.thread_id(), - forked_from_id: None, - parent_thread_id: None, - thread_source: None, - thread_name: None, - model: "test".to_string(), - model_provider_id: "test".to_string(), - service_tier: None, - approval_policy: AskForApproval::Never, - approvals_reviewer: ApprovalsReviewer::User, - permission_profile: turn_context.permission_profile(), - active_permission_profile: None, - cwd: turn_context.config.cwd.clone(), - reasoning_effort: None, - network_proxy: None, - rollout_path: None, - }), - /*rollout_path*/ None, + &turn_context, + rx_event, SessionSource::Cli, - ); + ) + .await; thread.submit(Op::Interrupt).await.expect("fill the queue"); let interrupt = thread.interrupt_if_no_pending_input(&turn_context.sub_id); diff --git a/codex-rs/core/src/session/handlers.rs b/codex-rs/core/src/session/handlers.rs index 5ded792689..128bc7d5bf 100644 --- a/codex-rs/core/src/session/handlers.rs +++ b/codex-rs/core/src/session/handlers.rs @@ -643,6 +643,10 @@ pub(super) async fn submission_loop( } false } + Op::ReplaceDynamicTools { tools, reply } => { + let _ = reply.send(sess.replace_dynamic_tools(tools).await); + false + } Op::TurnSettings { turn_id, update, diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 90ec0c9318..bc2f4c24f5 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -353,6 +353,7 @@ use codex_git_utils::get_git_repo_root; use codex_history::CodexHarnessMetadata; use codex_history::CompactedItem; use codex_history::CompactionResumeMetadata; +use codex_history::DynamicToolsSnapshot; use codex_history::InitialHistory; pub(crate) use codex_history::PreviousTurnSettings; use codex_history::ResponseItemEnvelope; @@ -845,8 +846,10 @@ impl Session { .or_else(|| conversation_history.get_base_instructions().map(|s| s.text)) .unwrap_or_else(|| render_model_instructions(&model_info)); - // Dynamic tools are defined at thread start and persisted in rollout session metadata. - let dynamic_tools = if dynamic_tools.is_empty() { + // A fork records its initial catalog in the new thread's creation metadata. + let dynamic_tools = if dynamic_tools.is_empty() + && matches!(conversation_history, InitialHistory::Forked(_)) + { conversation_history.get_dynamic_tools().unwrap_or_default() } else { dynamic_tools @@ -4073,6 +4076,7 @@ impl Session { settings = Arc::new(inherited_settings); } } + let dynamic_tools = self.dynamic_tools().await; let token_budget = token_budget::resolve_token_budget( turn_context.configured_token_budget.as_ref(), turn_context.use_model_token_budget_defaults, @@ -4200,6 +4204,7 @@ impl Session { &mcp, &extension_data, prepared_recommendations, + &dynamic_tools, ) .await?; Ok::<_, CodexErr>((executor_hook_sources, mcp, tool_router, extension_data)) @@ -4550,6 +4555,10 @@ impl Session { compaction_response_id: metadata.compaction_response_id, latest_token_usage_record: state.latest_token_usage_record.clone(), resume_metadata: Some(CompactionResumeMetadata { + dynamic_tools: Some(Box::new(DynamicToolsSnapshot { + thread_id: self.thread_id(), + tools: state.persisted_dynamic_tools.to_vec(), + })), multi_agent_version: self.multi_agent_version(), last_started_turn_id: state.last_started_turn_id.clone(), turn_attribution: state.turn_attribution.clone(), diff --git a/codex-rs/core/src/session/rollout_reconstruction.rs b/codex-rs/core/src/session/rollout_reconstruction.rs index 7c8ae88b5d..1205827911 100644 --- a/codex-rs/core/src/session/rollout_reconstruction.rs +++ b/codex-rs/core/src/session/rollout_reconstruction.rs @@ -242,7 +242,8 @@ impl ContextManager { .rev() .filter_map(|item| match item { RolloutItem::EventMsg(EventMsg::TurnStarted(event)) => Some(event.turn_id.as_str()), - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::InterAgentCommunicationMetadata { .. } @@ -439,6 +440,7 @@ impl ContextManager { active_segment.counts_as_user_turn = true; } RolloutItem::EventMsg(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SessionMeta(_) | RolloutItem::RealtimeItem(_) | RolloutItem::RetainedContext(_) @@ -553,6 +555,7 @@ impl ContextManager { | RolloutItem::WorldState(_) | RolloutItem::SecurityRiskScore(_) | RolloutItem::TokenUsageRecord(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SessionMeta(_) => {} } } @@ -586,7 +589,8 @@ impl ContextManager { }; baseline.apply_merge_patch(&world_state.state); } - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::InterAgentCommunicationMetadata { .. } diff --git a/codex-rs/core/src/session/rollout_reconstruction_tests.rs b/codex-rs/core/src/session/rollout_reconstruction_tests.rs index 86430a174f..c102a49743 100644 --- a/codex-rs/core/src/session/rollout_reconstruction_tests.rs +++ b/codex-rs/core/src/session/rollout_reconstruction_tests.rs @@ -1622,6 +1622,7 @@ async fn bounded_replay_matches_full_replay_after_empty_turn_compactions( compaction_response_id: None, latest_token_usage_record: None, resume_metadata: current.then(|| codex_history::CompactionResumeMetadata { + dynamic_tools: None, multi_agent_version: None, last_started_turn_id: Some(format!("wake-{window_number}")), turn_attribution: None, diff --git a/codex-rs/core/src/session/session.rs b/codex-rs/core/src/session/session.rs index 59c926b9f8..f979ceef90 100644 --- a/codex-rs/core/src/session/session.rs +++ b/codex-rs/core/src/session/session.rs @@ -166,8 +166,8 @@ pub(crate) struct SessionConfiguration { pub(super) thread_source: Option, /// Effective originator used for this thread's Responses requests and analytics events. pub(super) originator: String, - /// Fixed at thread startup and shared with turn snapshots. - pub(super) dynamic_tools: Arc<[DynamicToolSpec]>, + /// Active tools used to build the next model request. + pub(crate) dynamic_tools: Arc<[DynamicToolSpec]>, pub(super) user_shell_override: Option, } @@ -707,12 +707,9 @@ impl Session { state.session_configuration.originator.clone() } - pub(crate) async fn dynamic_tools(&self) -> Vec { - let dynamic_tools = { - let state = self.state.lock().await; - Arc::clone(&state.session_configuration.dynamic_tools) - }; - dynamic_tools.to_vec() + pub(crate) async fn dynamic_tools(&self) -> Arc<[DynamicToolSpec]> { + let state = self.state.lock().await; + Arc::clone(&state.session_configuration.dynamic_tools) } pub(crate) async fn responses_metadata( @@ -1576,6 +1573,16 @@ impl Session { &config.features, ), ); + if matches!(initial_history, InitialHistory::Resumed(_)) { + state.persisted_dynamic_tools = initial_history + .get_dynamic_tools() + .unwrap_or_default() + .into(); + if state.session_configuration.dynamic_tools.is_empty() { + state.session_configuration.dynamic_tools = + Arc::clone(&state.persisted_dynamic_tools); + } + } state.base_instructions_provenance = base_instructions_provenance.clone(); // Restore a provided effort baseline before startup prewarm can establish a new one. if let Some(pin) = thread_extension_data.remove::() { @@ -1711,7 +1718,8 @@ impl Session { } } RolloutItem::EventMsg(event) => mcp_runtime.observe_event(event), - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::InterAgentCommunicationMetadata { .. } diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index a8525d297d..579d109872 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -330,6 +330,8 @@ impl StepContext { } } +#[path = "dynamic_tools_tests.rs"] +mod dynamic_tools_tests; mod guardian_tests; fn user_message(text: &str) -> ResponseItem { @@ -1054,7 +1056,8 @@ async fn preview_session_start_hooks( ) } -pub(crate) fn tool_registry_for_test_step( +pub(crate) async fn tool_registry_for_test_step( + session: &Session, step_context: &StepContext, ) -> (ToolRegistry, Vec) { let mut registry = crate::tools::spec_plan::build_core_tool_registry( @@ -1071,14 +1074,18 @@ pub(crate) fn tool_registry_for_test_step( &mut registry, Vec::new(), Vec::new(), - &step_context.turn.dynamic_tools, + session.dynamic_tools().await.as_ref(), ); (registry, hosted_specs) } -fn test_tool_runtime(session: Arc, turn_context: Arc) -> ToolCallRuntime { +async fn test_tool_runtime( + session: Arc, + turn_context: Arc, +) -> ToolCallRuntime { let step_context = StepContext::for_test(Arc::clone(&turn_context)); - let (registry, hosted_specs) = tool_registry_for_test_step(step_context.as_ref()); + let (registry, hosted_specs) = + tool_registry_for_test_step(&session, step_context.as_ref()).await; let router = Arc::new(ToolRouter::from_registry( step_context.turn.as_ref(), step_context.turn.model_info(), @@ -4113,16 +4120,46 @@ async fn record_initial_history_reconstructs_forked_transcript() { ); } +#[test_case::test_case(false; "replacement")] +#[test_case::test_case(true; "cleared")] #[tokio::test] -async fn start_new_context_window_persists_checkpoint_state() { +async fn start_new_context_window_persists_checkpoint_state(clear_tools: bool) { let (mut session, turn_context, _rx) = make_session_and_context_with_auth_and_config_and_rx( CodexAuth::from_api_key("Test API Key"), Vec::new(), - |_| {}, + |config| config.allow_persistent_dynamic_tool_updates = true, ) .await; + let dynamic_tools = vec![codex_protocol::dynamic_tools::DynamicToolSpec::Function( + codex_protocol::dynamic_tools::DynamicToolFunctionSpec { + name: "check_release".into(), + description: "Check the release.".into(), + input_schema: json!({"type": "object", "properties": {}}), + defer_loading: false, + }, + )]; + let mut initial_tools = dynamic_tools.clone(); + if let codex_protocol::dynamic_tools::DynamicToolSpec::Function(tool) = &mut initial_tools[0] { + tool.name = "original_tool".into(); + } + session + .state + .lock() + .await + .session_configuration + .dynamic_tools = initial_tools.into(); let rollout_path = attach_thread_persistence(Arc::get_mut(&mut session).expect("unique session")).await; + session + .replace_dynamic_tools(dynamic_tools.clone()) + .await + .unwrap(); + let dynamic_tools = if clear_tools { + session.replace_dynamic_tools(Vec::new()).await.unwrap(); + Vec::new() + } else { + dynamic_tools + }; let thread_id = ThreadId::new(); let token_usage_record = TokenUsageRecord { thread_id, @@ -4165,6 +4202,29 @@ async fn start_new_context_window_persists_checkpoint_state() { else { panic!("expected resumed rollout history"); }; + let checkpoint = resumed + .history + .iter() + .rposition(|item| matches!(item, RolloutItem::Compacted(_))) + .unwrap(); + assert_eq!( + codex_history::latest_dynamic_tools(&resumed.history[checkpoint..]), + Some(dynamic_tools.as_slice()) + ); + // A crash can leave only the compaction record, without its following companions. + let mut scan = codex_rollout::ModelContextScan::default(); + for item in resumed.history[..=checkpoint].iter().rev() { + if scan.push(item.clone()) == codex_rollout::ModelContextScanProgress::Complete { + break; + } + } + let mut recovered = scan.finish(); + assert!(matches!(recovered.as_slice(), [RolloutItem::Compacted(_)])); + recovered.insert(0, resumed.history[0].clone()); + assert_eq!( + codex_history::latest_dynamic_tools(&recovered), + Some(dynamic_tools.as_slice()) + ); let persisted_world_state = resumed .history .iter() @@ -4178,7 +4238,8 @@ async fn start_new_context_window_persists_checkpoint_state() { assert_eq!(persisted_world_state.state, expected_snapshot); let persisted_compacted = resumed.history.iter().rev().find_map(|item| match item { RolloutItem::Compacted(compacted) => Some(compacted), - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::InterAgentCommunicationMetadata { .. } @@ -4278,7 +4339,8 @@ async fn record_initial_history_assigns_and_persists_id_for_forked_response_item }; let persisted_item = resumed.history.iter().find_map(|item| match item { RolloutItem::ResponseItem(response_item) => Some(response_item), - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::InterAgentCommunicationMetadata { .. } | RolloutItem::Compacted(_) @@ -4999,7 +5061,7 @@ async fn open_thread_persistence(session: &mut Session) -> PathBuf { thread_source: None, originator: "test_originator".to_string(), base_instructions: BaseInstructions::default(), - dynamic_tools: Vec::new(), + dynamic_tools: session.dynamic_tools().await.to_vec(), selected_capability_roots: Vec::new(), multi_agent_version: None, history_mode: Default::default(), @@ -6299,6 +6361,10 @@ async fn compaction_persists_resume_metadata_and_companion_records() { session.multi_agent_version = std::sync::OnceLock::from(MultiAgentVersion::V2); let expected = CompactionResumeMetadata { + dynamic_tools: Some(Box::new(DynamicToolsSnapshot { + thread_id: session.thread_id(), + tools: Vec::new(), + })), multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some(turn_context.sub_id.clone()), turn_attribution: Some(codex_history::TurnAttribution { @@ -9455,6 +9521,49 @@ pub(crate) async fn make_session_and_context_with_rx() -> ( make_session_and_context_with_dynamic_tools_and_rx(Vec::new()).await } +// Wrap an existing session without starting a submission loop, so tests control its queue. +pub(super) async fn make_thread_for_session( + session: Arc, + turn_context: &TurnContext, + rx_event: async_channel::Receiver, + session_source: SessionSource, +) -> (crate::CodexThread, async_channel::Receiver) { + let rollout_path = session.current_rollout_path().await.expect("rollout path"); + let event = SessionConfiguredEvent { + session_id: session.session_id(), + thread_id: session.thread_id(), + forked_from_id: None, + parent_thread_id: None, + thread_source: None, + thread_name: None, + model: "test".into(), + model_provider_id: "test".into(), + service_tier: None, + approval_policy: AskForApproval::Never, + approvals_reviewer: ApprovalsReviewer::User, + permission_profile: turn_context.permission_profile(), + active_permission_profile: None, + cwd: turn_context.config.cwd.clone(), + reasoning_effort: None, + network_proxy: None, + rollout_path: rollout_path.clone(), + }; + let (tx_sub, rx_sub) = async_channel::bounded(/*cap*/ 1); + let thread = crate::CodexThread::new( + session, + SessionIo { + tx_sub, + rx_event, + agent_status: watch::channel(AgentStatus::PendingInit).1, + session_loop_termination: completed_session_loop_termination(), + }, + crate::ThreadStartupMetadata::from(&event), + rollout_path, + session_source, + ); + (thread, rx_sub) +} + #[tokio::test] async fn cancelled_step_capture_finishes_warning_delivery() { struct WarningProvider { @@ -13183,7 +13292,7 @@ async fn tool_calls_reopen_mailbox_delivery_for_current_turn() { sess: Arc::clone(&sess), step_context: StepContext::for_test(Arc::clone(&tc)), turn_store: Arc::new(codex_extension_api::ExtensionData::new(tc.sub_id.clone())), - tool_runtime: test_tool_runtime(Arc::clone(&sess), Arc::clone(&tc)), + tool_runtime: test_tool_runtime(Arc::clone(&sess), Arc::clone(&tc)).await, defer_tool_dispatch: false, cancellation_token: CancellationToken::new(), }; @@ -13282,7 +13391,8 @@ async fn abort_review_task_emits_exited_then_aborted_and_records_history() { async fn fatal_tool_error_stops_turn_and_reports_error() { let (session, turn_context, _rx) = make_session_and_context_with_rx().await; let step_context = StepContext::for_test(Arc::clone(&turn_context)); - let (registry, hosted_specs) = tool_registry_for_test_step(step_context.as_ref()); + let (registry, hosted_specs) = + tool_registry_for_test_step(&session, step_context.as_ref()).await; let router = ToolRouter::from_registry( step_context.turn.as_ref(), step_context.turn.model_info(), diff --git a/codex-rs/core/src/session/thread_settings.rs b/codex-rs/core/src/session/thread_settings.rs index f07d921dc2..c3a82eff0f 100644 --- a/codex-rs/core/src/session/thread_settings.rs +++ b/codex-rs/core/src/session/thread_settings.rs @@ -8,15 +8,54 @@ use crate::WithTurnExtensionData; use crate::config::ConstraintResult; use codex_history::RolloutItem; use codex_protocol::capabilities::SelectedCapabilityRoot; +use codex_protocol::error::CodexErr; +use codex_protocol::error::Result as CodexResult; use codex_protocol::protocol::Event; use codex_protocol::protocol::EventMsg; use codex_protocol::protocol::ThreadSettingsAppliedEvent; use codex_protocol::protocol::ThreadSettingsOverrides; use codex_protocol::protocol::ThreadSettingsSnapshot; +use codex_thread_store::ThreadStoreError; use codex_thread_store::ThreadStoreResult; +use std::sync::Arc; use tokio::sync::SemaphorePermit; impl Session { + /// Persist before publishing so a failed write cannot change the running catalog. + pub(crate) async fn replace_dynamic_tools( + &self, + tools: Vec, + ) -> CodexResult<()> { + let _settings_guard = acquire_persistence_lock(self).await; + if let Some(live_thread) = self.live_thread() { + let config = self.get_config().await; + if !config.allow_persistent_dynamic_tool_updates { + return Err(CodexErr::UnsupportedOperation( + "The host has disabled dynamic tool updates for persistent threads.".into(), + )); + } + let persistence_error = |error: ThreadStoreError| match error { + ThreadStoreError::Unsupported { .. } => { + CodexErr::UnsupportedOperation(error.to_string()) + } + _ => CodexErr::Fatal(format!("failed to replace dynamic tools: {error}")), + }; + live_thread + .append_items(&[RolloutItem::DynamicToolsUpdated { + thread_id: self.thread_id(), + tools: tools.clone(), + }]) + .await + .map_err(persistence_error)?; + live_thread.flush().await.map_err(persistence_error)?; + } + let tools = Arc::from(tools); + let mut state = self.state.lock().await; + state.persisted_dynamic_tools = Arc::clone(&tools); + state.session_configuration.dynamic_tools = tools; + Ok(()) + } + /// Captures and flushes current settings under the shared persistence permit. pub(crate) async fn checkpoint_thread_settings(&self) -> ThreadStoreResult<()> { let _settings_guard = acquire_persistence_lock(self).await; diff --git a/codex-rs/core/src/session/turn.rs b/codex-rs/core/src/session/turn.rs index 7020ad25a3..4c1395fa19 100644 --- a/codex-rs/core/src/session/turn.rs +++ b/codex-rs/core/src/session/turn.rs @@ -1935,6 +1935,7 @@ pub(crate) async fn built_tools( mcp: &Arc, step_store: &ExtensionData, prepared_recommendations: PreparedToolRecommendations, + dynamic_tools: &[codex_protocol::dynamic_tools::DynamicToolSpec], ) -> CodexResult> { let all_mcp_tools = mcp.tools(); let connector_snapshot = mcp.config().connector_snapshot.clone(); @@ -2005,6 +2006,7 @@ pub(crate) async fn built_tools( apps_enabled, step_store, tool_suggest_candidates.as_ref(), + dynamic_tools, )?)) } diff --git a/codex-rs/core/src/session/turn_context.rs b/codex-rs/core/src/session/turn_context.rs index b31eeca5d5..a25dd60102 100644 --- a/codex-rs/core/src/session/turn_context.rs +++ b/codex-rs/core/src/session/turn_context.rs @@ -368,7 +368,6 @@ pub struct TurnContext { pub(crate) windows_sandbox_level: WindowsSandboxLevel, pub(crate) available_models: Vec, pub(crate) final_output_json_schema: Option, - pub(crate) dynamic_tools: Arc<[DynamicToolSpec]>, pub(crate) turn_metadata_state: Arc, pub(crate) extension_data: Arc, pub(crate) turn_timing_state: Arc, @@ -452,7 +451,6 @@ impl TurnContext { #[allow(deprecated)] cwd: self.cwd.clone(), final_output_json_schema: None, - dynamic_tools: self.dynamic_tools.clone(), turn_metadata_state, extension_data, turn_timing_state: Arc::new(TurnTimingState::default()), @@ -864,7 +862,6 @@ impl TurnContext { windows_sandbox_level: self.windows_sandbox_level, available_models, final_output_json_schema: self.final_output_json_schema.clone(), - dynamic_tools: self.dynamic_tools.clone(), turn_metadata_state: self.turn_metadata_state.clone(), extension_data: Arc::clone(&self.extension_data), turn_timing_state: Arc::clone(&self.turn_timing_state), @@ -1149,7 +1146,6 @@ impl Session { windows_sandbox_level: session_configuration.windows_sandbox_level, available_models, final_output_json_schema: None, - dynamic_tools: session_configuration.dynamic_tools.clone(), turn_metadata_state, extension_data, turn_timing_state: Arc::new(TurnTimingState::default()), diff --git a/codex-rs/core/src/state/session.rs b/codex-rs/core/src/state/session.rs index d122db22d8..ebfc5817cb 100644 --- a/codex-rs/core/src/state/session.rs +++ b/codex-rs/core/src/state/session.rs @@ -1,11 +1,13 @@ //! Session-wide mutable state. +use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::models::BaseInstructionsProvenance; #[cfg(test)] use codex_protocol::models::ResponseItem; use codex_protocol::openai_models::ReasoningEffort; use std::collections::HashSet; use std::collections::VecDeque; +use std::sync::Arc; use super::AdditionalContextStore; use super::auto_compact_window::AutoCompactWindow; @@ -68,6 +70,8 @@ impl ReasoningEffortPin { /// Persistent, session-scoped state previously stored directly on `Session`. pub(crate) struct SessionState { pub(crate) session_configuration: SessionConfiguration, + /// Catalog to carry across compaction, independent of an explicit resume override. + pub(crate) persisted_dynamic_tools: Arc<[DynamicToolSpec]>, /// Plugin selection of the last admitted task; settings updates take effect on the next task. pub(crate) active_disabled_plugin_ids: Vec, /// Persisted origin of the session base instructions, when known. @@ -125,6 +129,7 @@ impl SessionState { ) -> Self { Self { active_disabled_plugin_ids: Vec::new(), + persisted_dynamic_tools: Arc::clone(&session_configuration.dynamic_tools), session_configuration, base_instructions_provenance: None, history, diff --git a/codex-rs/core/src/stream_events_utils_tests.rs b/codex-rs/core/src/stream_events_utils_tests.rs index d402653316..a7b47e1370 100644 --- a/codex-rs/core/src/stream_events_utils_tests.rs +++ b/codex-rs/core/src/stream_events_utils_tests.rs @@ -295,9 +295,10 @@ async fn handle_non_tool_response_item_runs_turn_item_contributors_only_when_req assert_eq!(text, "hello world"); } -fn output_context(session: Arc, turn_context: Arc) -> HandleOutputCtx { +async fn output_context(session: Arc, turn_context: Arc) -> HandleOutputCtx { let step_context = StepContext::for_test(Arc::clone(&turn_context)); - let (registry, hosted_specs) = tool_registry_for_test_step(step_context.as_ref()); + let (registry, hosted_specs) = + tool_registry_for_test_step(&session, step_context.as_ref()).await; let router = Arc::new(ToolRouter::from_registry( step_context.turn.as_ref(), step_context.turn.model_info(), @@ -325,7 +326,7 @@ async fn handle_output_item_done_returns_contributed_last_agent_message() { let mut builder = codex_extension_api::ExtensionRegistryBuilder::new(); builder.turn_item_contributor(Arc::new(RewriteAgentMessageContributor)); session.services.extensions = Arc::new(builder.build()); - let mut ctx = output_context(Arc::new(session), Arc::new(turn_context)); + let mut ctx = output_context(Arc::new(session), Arc::new(turn_context)).await; let item = assistant_output_text("original assistant text"); let output = handle_output_item_done(&mut ctx, item, /*previously_active_item*/ None) @@ -352,7 +353,7 @@ async fn direct_results_keep_their_own_records_when_call_ids_repeat() { }, ) .await; - let mut ctx = output_context(Arc::clone(&session), turn_context); + let mut ctx = output_context(Arc::clone(&session), turn_context).await; let mut pending = Vec::new(); for step in ["read", "write"] { let arguments = json!({"plan": [{"step": step, "status": "in_progress"}]}); diff --git a/codex-rs/core/src/tools/router_tests.rs b/codex-rs/core/src/tools/router_tests.rs index aff6d41660..498fe97057 100644 --- a/codex-rs/core/src/tools/router_tests.rs +++ b/codex-rs/core/src/tools/router_tests.rs @@ -164,14 +164,14 @@ fn test_tool_router( #[tokio::test] async fn parallel_support_does_not_match_namespaced_local_tool_names() -> anyhow::Result<()> { - let (_, turn) = make_session_and_context().await; + let (session, turn) = make_session_and_context().await; let turn = Arc::new(turn); let step_context = StepContext::for_test(Arc::clone(&turn)); let router = test_tool_router( step_context.as_ref(), Vec::new(), Vec::new(), - &turn.dynamic_tools, + session.dynamic_tools().await.as_ref(), ); let parallel_tool_name = ["exec_command"] @@ -313,7 +313,7 @@ fn build_tool_call_normalizes_default_function_and_custom_namespaces() -> anyhow #[tokio::test] async fn mcp_parallel_support_uses_handler_data() -> anyhow::Result<()> { - let (_, turn) = make_session_and_context().await; + let (session, turn) = make_session_and_context().await; let turn = Arc::new(turn); let step_context = StepContext::for_test(Arc::clone(&turn)); let router = test_tool_router( @@ -354,7 +354,7 @@ async fn mcp_parallel_support_uses_handler_data() -> anyhow::Result<()> { }, ], Vec::new(), - &turn.dynamic_tools, + session.dynamic_tools().await.as_ref(), ); let call = ToolCall { @@ -415,14 +415,14 @@ async fn mcp_parallel_support_uses_handler_data() -> anyhow::Result<()> { #[tokio::test] async fn tools_without_handlers_do_not_support_parallel() -> anyhow::Result<()> { - let (_, turn) = make_session_and_context().await; + let (session, turn) = make_session_and_context().await; let turn = Arc::new(turn); let step_context = StepContext::for_test(Arc::clone(&turn)); let router = test_tool_router( step_context.as_ref(), Vec::new(), Vec::new(), - &turn.dynamic_tools, + session.dynamic_tools().await.as_ref(), ); assert!(!router.tool_supports_parallel(&ToolCall { @@ -573,7 +573,7 @@ async fn extension_tool_executors_are_model_visible_and_dispatchable() -> anyhow &session, &codex_extension_api::ExtensionData::new(turn.sub_id.clone()), ), - &turn.dynamic_tools, + session.dynamic_tools().await.as_ref(), ); assert!( diff --git a/codex-rs/core/src/tools/spec_plan.rs b/codex-rs/core/src/tools/spec_plan.rs index 192deff7e1..2602fcb9bb 100644 --- a/codex-rs/core/src/tools/spec_plan.rs +++ b/codex-rs/core/src/tools/spec_plan.rs @@ -127,6 +127,7 @@ pub(crate) fn build_tool_router( apps_enabled: bool, step_store: &ExtensionData, tool_suggest_candidates: Option<&crate::tools::router::ToolSuggestCandidates>, + dynamic_tools: &[DynamicToolSpec], ) -> CodexResult { let default_agent_type_description = crate::agent::role::spawn_tool_spec::build(&std::collections::BTreeMap::new()); @@ -170,7 +171,7 @@ pub(crate) fn build_tool_router( extension_tool_executors(session, step_store), &mut registry, ); - append_dynamic_tool_runtimes(&turn_context.dynamic_tools, &mut registry); + append_dynamic_tool_runtimes(dynamic_tools, &mut registry); let hosted_specs = hosted_model_tool_specs( turn_context, model_info, diff --git a/codex-rs/core/src/tools/spec_plan_tests.rs b/codex-rs/core/src/tools/spec_plan_tests.rs index 1ecd9f77f1..58351c7cac 100644 --- a/codex-rs/core/src/tools/spec_plan_tests.rs +++ b/codex-rs/core/src/tools/spec_plan_tests.rs @@ -601,6 +601,7 @@ async fn reviewer_tool_policy_exclude_optional_core_tools() { /*apps_enabled*/ false, &turn.extension_data, /*tool_suggest_candidates*/ None, + session.dynamic_tools().await.as_ref(), ) .expect("build internal Guardian tool router"); @@ -653,6 +654,7 @@ async fn reviewer_tool_policy_respect_managed_shell_restrictions() { /*apps_enabled*/ false, &turn.extension_data, /*tool_suggest_candidates*/ None, + session.dynamic_tools().await.as_ref(), ) .expect("build internal Guardian tool router"); @@ -689,6 +691,7 @@ async fn reviewer_tool_policy_preserve_code_mode() { /*apps_enabled*/ false, &turn.extension_data, /*tool_suggest_candidates*/ None, + session.dynamic_tools().await.as_ref(), ) .expect("build internal Guardian tool router"); @@ -759,6 +762,7 @@ async fn reviewer_tool_policy_require_managed_secondary_environments() { /*apps_enabled*/ false, &turn.extension_data, /*tool_suggest_candidates*/ None, + session.dynamic_tools().await.as_ref(), ) .expect("build internal Guardian tool router"); diff --git a/codex-rs/core/tests/suite/scenarios.rs b/codex-rs/core/tests/suite/scenarios.rs index ff0c226da5..b7e8c56a8b 100644 --- a/codex-rs/core/tests/suite/scenarios.rs +++ b/codex-rs/core/tests/suite/scenarios.rs @@ -102,6 +102,9 @@ use tokio::sync::oneshot; const ONE_PIXEL_PNG_BASE64: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR4nGP4z8DwHwAFAAH/iZk9HQAAAABJRU5ErkJggg=="; +#[path = "scenarios_dynamic_tools.rs"] +mod dynamic_tools; + #[path = "scenarios_retained_tool_outputs.rs"] mod retained_tool_outputs; diff --git a/codex-rs/core/tests/suite/scenarios_dynamic_tools.rs b/codex-rs/core/tests/suite/scenarios_dynamic_tools.rs new file mode 100644 index 0000000000..38ececca0d --- /dev/null +++ b/codex-rs/core/tests/suite/scenarios_dynamic_tools.rs @@ -0,0 +1,261 @@ +//! Dynamic tool replacement takes effect between model requests, including after restart. + +use super::*; +use core_test_support::responses; +use pretty_assertions::assert_eq; + +fn configure_dynamic_tool_scenario(config: &mut Config) { + config.model = Some("gpt-6-astra".into()); + configure_scenario_catalog(config); + config.include_environment_context = false; + config + .features + .disable(Feature::CodeMode) + .expect("disable code mode"); + let model = config + .model_catalog + .as_mut() + .expect("scenario model catalog") + .models + .iter_mut() + .find(|model| model.slug == "gpt-6-astra") + .expect("Astra model"); + model.use_responses_lite = false; + model.tool_mode = Some(ToolMode::Direct); +} + +fn tool(name: &str) -> DynamicToolSpec { + DynamicToolSpec::Function(DynamicToolFunctionSpec { + name: name.to_string(), + description: format!("Check {name}."), + input_schema: json!({"type": "object", "properties": {}, "additionalProperties": false}), + defer_loading: false, + }) +} + +fn has_tool(body: serde_json::Value, name: &str) -> bool { + body["tools"] + .as_array() + .unwrap_or_else(|| panic!("request has no tools: {body}")) + .iter() + .any(|tool| tool["name"] == name) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn dynamic_tools_update_preserves_pending_calls_and_survives_resume() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let enable_updates = |config: &mut Config| config.allow_persistent_dynamic_tool_updates = true; + let mut builder = test_codex() + .with_config(configure_dynamic_tool_scenario) + .with_config(enable_updates); + let mut test = builder.build_with_auto_env(&server).await?; + let first = responses::mount_sse_once( + &server, + sse(vec![ + responses::ev_function_call("release-call", "release", "{}"), + ev_completed("first"), + ]), + ) + .await; + let (reply, updated) = tokio::sync::oneshot::channel(); + test.codex + .submit(Op::ReplaceDynamicTools { + tools: vec![tool("release")], + reply, + }) + .await?; + test.codex + .start_or_steer_turn(TurnInputRequest::user_input(vec![text( + "Check the release and deployment.", + )])) + .await?; + updated.await??; + let call = wait_for_event_match(&test.codex, |event| match event { + EventMsg::DynamicToolCallRequest(call) => Some(call.clone()), + _ => None, + }) + .await; + assert_eq!(call.tool, "release"); + test.codex + .replace_dynamic_tools(vec![tool("deployment")]) + .await?; + let second = responses::mount_sse_once( + &server, + sse(vec![ + responses::ev_function_call("deployment-call", "deployment", "{}"), + ev_completed("second"), + ]), + ) + .await; + test.codex + .submit(Op::DynamicToolResponse { + id: call.call_id, + response: DynamicToolResponse { + content_items: vec![DynamicToolCallOutputContentItem::InputText { + text: "Release ready.".into(), + }], + success: true, + }, + }) + .await?; + let call = wait_for_event_match(&test.codex, |event| match event { + EventMsg::DynamicToolCallRequest(call) => Some(call.clone()), + _ => None, + }) + .await; + assert_eq!(call.tool, "deployment"); + let third = responses::mount_sse_once( + &server, + sse(vec![ + ev_assistant_message("ready", "Both checks passed."), + ev_completed("third"), + ]), + ) + .await; + test.codex + .submit(Op::DynamicToolResponse { + id: call.call_id, + response: DynamicToolResponse { + content_items: vec![DynamicToolCallOutputContentItem::InputText { + text: "Deployment ready.".into(), + }], + success: true, + }, + }) + .await?; + wait_for_event(&test.codex, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + assert!(has_tool(first.single_request().body_json(), "release")); + assert!(!has_tool(second.single_request().body_json(), "release")); + assert!(has_tool(second.single_request().body_json(), "deployment")); + assert_eq!( + second + .single_request() + .function_call_output_content_and_success("release-call"), + Some((Some("Release ready.".into()), None)) + ); + assert_eq!( + third + .single_request() + .function_call_output_content_and_success("deployment-call"), + Some((Some("Deployment ready.".into()), None)) + ); + + builder = builder.with_config(configure_dynamic_tool_scenario); + test = builder.restart_with_auto_env(&server, &test).await?; + let restored = responses::mount_sse_once( + &server, + sse(vec![ + ev_assistant_message("restored", "The deployment tool is available."), + ev_completed("restored"), + ]), + ) + .await; + test.submit_text_turn("Which check is available now?") + .await?; + assert!(has_tool( + restored.single_request().body_json(), + "deployment" + )); + assert!(!has_tool(restored.single_request().body_json(), "release")); + + builder = builder + .with_config(configure_dynamic_tool_scenario) + .with_config(enable_updates); + test = builder.restart_with_auto_env(&server, &test).await?; + test.codex.replace_dynamic_tools(Vec::new()).await?; + builder = builder.with_config(configure_dynamic_tool_scenario); + test = builder.restart_with_auto_env(&server, &test).await?; + let restarted = responses::mount_sse_once( + &server, + sse(vec![ + ev_assistant_message("done", "The checks are complete."), + ev_completed("restarted"), + ]), + ) + .await; + test.submit_text_turn("Summarize the checks.").await?; + assert!(!has_tool(restarted.single_request().body_json(), "release")); + assert!(!has_tool( + restarted.single_request().body_json(), + "deployment" + )); + let requests = responses::received_responses_requests(&server).await; + insta::assert_snapshot!(context_snapshot::format_request_history_snapshot( + "A pending release call survives replacement with deployment. The next step uses deployment. Replacement and clearing each survive restart.", + &requests, + &ContextSnapshotOptions::default() + .rewrite_known_segments() + .include_request_settings(), + )); + test.codex.shutdown_and_wait().await?; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn dynamic_tools_resume_override_stays_local_after_compaction() -> Result<()> { + skip_if_no_network!(Ok(())); + let server = start_mock_server().await; + let test = test_codex() + .with_config(configure_dynamic_tool_scenario) + .with_config(|config| { + config.features.enable(Feature::TokenBudget).unwrap(); + }) + .build_with_auto_env(&server) + .await?; + let start = async |initial_history, dynamic_tools| { + test.thread_manager + .start_thread(StartThreadOptions { + initial_history, + dynamic_tools, + history_mode: Some(codex_protocol::protocol::ThreadHistoryMode::Legacy), + ..test.start_thread_options().await + }) + .await + }; + let initial = start(codex_history::InitialHistory::New, vec![tool("release")]).await?; + initial.thread.ensure_rollout_materialized().await; + initial.thread.shutdown_and_wait().await?; + let path = initial.thread.rollout_path().expect("rollout path"); + let load_history = || codex_rollout::RolloutRecorder::get_rollout_history(&path); + let check_tools = async |thread: &codex_core::CodexThread, present: &str, absent: &str| { + let response = responses::mount_sse_once(&server, responses::sse_completed(present)).await; + thread + .start_or_steer_turn(TurnInputRequest::user_input(vec![text( + "Check the release.", + )])) + .await?; + wait_for_event(thread, |event| matches!(event, EventMsg::TurnComplete(_))).await; + let request = response.single_request(); + assert!(has_tool(request.body_json(), present)); + assert!(!has_tool(request.body_json(), absent)); + Ok::<_, anyhow::Error>(()) + }; + + let overridden = start(load_history().await?, vec![tool("deployment")]) + .await? + .thread; + check_tools(&overridden, "deployment", "release").await?; + overridden.submit(Op::Compact).await?; + wait_for_event(&overridden, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + overridden.shutdown_and_wait().await?; + + let history = load_history().await?; + assert!( + history + .get_rollout_items() + .iter() + .any(|item| matches!(item, codex_history::RolloutItem::Compacted(_))), + "resume must read the persisted compaction checkpoint" + ); + let restored = start(history, Vec::new()).await?.thread; + check_tools(&restored, "release", "deployment").await?; + restored.shutdown_and_wait().await?; + Ok(()) +} diff --git a/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__dynamic_tools__dynamic_tools_update_preserves_pending_calls_and_survives_resume.snap b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__dynamic_tools__dynamic_tools_update_preserves_pending_calls_and_survives_resume.snap new file mode 100644 index 0000000000..cdbd3f8b2e --- /dev/null +++ b/codex-rs/core/tests/suite/snapshots/all__suite__scenarios__dynamic_tools__dynamic_tools_update_preserves_pending_calls_and_survives_resume.snap @@ -0,0 +1,112 @@ +--- +source: core/tests/suite/scenarios_dynamic_tools.rs +assertion_line: 173 +expression: "context_snapshot::format_request_history_snapshot(\"A pending release call survives replacement with deployment. The next step uses deployment. Replacement and clearing each survive restart.\",\n&requests,\n&ContextSnapshotOptions::default().rewrite_known_segments().include_request_settings(),)" +--- +Scenario: A pending release call survives replacement with deployment. The next step uses deployment. Replacement and clearing each survive restart. + +## Window 1 +Settings: + include: ["reasoning.encrypted_content"] + model: "gpt-6-astra" + parallel_tool_calls: true + prompt_cache_key: "" + reasoning: {"effort":"low"} + store: false + stream: true + text: {"verbosity":"low"} + tool_choice: "auto" + tools (10; hash=39aa00a8736bb135): + - function/exec_command: Runs a command in a PTY, returning output or a session ID for ongoing interaction.; args=[cmd, justification, login, max_output_tokens, prefix_rule, sandbox_permissions, shell, tty, workdir, yield_time_ms]; hash=c04108e51a8548c5 + - function/write_stdin: Writes characters to an existing unified exec session and returns recent output.; args=[chars, max_output_tokens, session_id, yield_time_ms]; hash=7fb62d8a3e2df4e2 + - function/request_user_input: Request user input for one to three short questions and wait for the response. This tool is only ...; args=[questions]; hash=64a729b25c424fc6 + - function/request_user_input_async: Ask the user one or more questions during ongoing work. Use this tool only to request missing inf...; args=[questions]; hash=d6d74f33db330162 + - namespace/clock: Tools for reading and waiting on time.; hash=a80435c7a3a3943f + - function/curr_time + - function/sleep + - custom/apply_patch: The `apply_patch` tool can be used to edit files. This is a FREEFORM tool, so do not wrap the pat...; hash=dea9f12b0111a4ba + - function/view_image: View a local image file from the filesystem when visual inspection is needed. Use this for images...; args=[detail, path]; hash=3b766d5bdec5d50e + - namespace/collaboration: Tools for spawning and managing sub-agents.; hash=2c8d6295978dbe63 + - function/followup_task + - function/interrupt_agent + - function/list_agents + - function/send_message + - function/spawn_agent + - function/wait_agent + - function/release: Check release.; args=[]; hash=4c6f91a1a21d2a65 + - web_search; hash=f103e327c6870a46 +-- request 1 (turn) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] +02:message/developer: + +03:message/developer: + +04:message/user: + Check the release and deployment. +05:configuration_update:reasoning={"effort":"low"} + +## Window 2 (after request 1) +Settings: relative to window 1 + tools (10; hash=f9e31c2361b9bc7d): + - removed function/release + + function/deployment: Check deployment.; args=[]; hash=2e9a71e9a20e43ad +-- request 2 (turn) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] +02:message/developer: + +03:message/developer: + +04:message/user: + Check the release and deployment. +05:configuration_update:reasoning={"effort":"low"} +06:function_call/release:{} +07:function_call_output:Release ready. +-- request 3 (turn) -- +08:function_call/deployment:{} +09:function_call_output:Deployment ready. +-- request 4 (turn) -- +10:message/assistant: + Both checks passed. +11:message/user: + Which check is available now? +12:configuration_update:reasoning={"effort":"low"} + +## Window 3 (after request 4) +Settings: relative to window 2 + tools (9; hash=4f2e62909d4cf2eb): + - removed function/deployment +-- request 5 (turn) -- +00:message/developer: + +01:message/developer[2]: + [01] + [02] +02:message/developer: + +03:message/developer: + +04:message/user: + Check the release and deployment. +05:configuration_update:reasoning={"effort":"low"} +06:function_call/release:{} +07:function_call_output:Release ready. +08:function_call/deployment:{} +09:function_call_output:Deployment ready. +10:message/assistant: + Both checks passed. +11:message/user: + Which check is available now? +12:configuration_update:reasoning={"effort":"low"} +13:message/assistant: + The deployment tool is available. +14:message/user: + Summarize the checks. +15:configuration_update:reasoning={"effort":"low"} diff --git a/codex-rs/external-agent-migration/src/sessions/append.rs b/codex-rs/external-agent-migration/src/sessions/append.rs index b77481afd3..61f329cc19 100644 --- a/codex-rs/external-agent-migration/src/sessions/append.rs +++ b/codex-rs/external-agent-migration/src/sessions/append.rs @@ -246,6 +246,7 @@ fn source_model_items(items: &[RolloutItem]) -> Option>> | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) | RolloutItem::RetainedContext(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SecurityRiskScore(_) | RolloutItem::WorldState(_) => return None, RolloutItem::EventMsg(_) => {} @@ -263,6 +264,7 @@ fn history_model_items(items: &[RolloutItem]) -> Option> { | RolloutItem::TokenUsageRecord(_) | RolloutItem::RealtimeItem(_) | RolloutItem::RetainedContext(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SecurityRiskScore(_) => {} RolloutItem::ResponseItem(response_item) => model_items.push(&response_item.item), RolloutItem::EventMsg( diff --git a/codex-rs/history/src/compaction_resume_metadata.rs b/codex-rs/history/src/compaction_resume_metadata.rs index bd8eb7a267..ca22476e1a 100644 --- a/codex-rs/history/src/compaction_resume_metadata.rs +++ b/codex-rs/history/src/compaction_resume_metadata.rs @@ -1,6 +1,8 @@ //! Defines resume metadata stored directly on a compaction. use crate::RolloutItem; +use codex_protocol::ThreadId; +use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::protocol::MultiAgentVersion; use codex_protocol::turn_input::CyberAccessProgram; use codex_protocol::turn_input::TurnAttribution; @@ -11,9 +13,12 @@ use serde::Serialize; /// Resume metadata that is not represented by the companion records after a compaction. /// /// Presence distinguishes compactions that explicitly persisted these values from older -/// compactions that did not. Each field is authoritative, including an intentionally absent value. +/// compactions that did not. #[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)] pub struct CompactionResumeMetadata { + /// Catalog owned by the producing thread; absent in older checkpoints. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub dynamic_tools: Option>, /// Selected runtime, including when the companion turn context is absent. pub multi_agent_version: Option, /// Turn identity used to admit continuations after cold resume. @@ -24,6 +29,13 @@ pub struct CompactionResumeMetadata { pub previous_turn_settings: Option, } +/// A thread's dynamic-tool catalog, including an empty catalog after a clear. +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, JsonSchema)] +pub struct DynamicToolsSnapshot { + pub thread_id: ThreadId, + pub tools: Vec, +} + /// Previous user-turn settings used to reconstruct context changes after resume. #[derive(Serialize, Deserialize, Clone, Debug, PartialEq, Eq, JsonSchema)] pub struct PreviousTurnSettings { @@ -43,7 +55,8 @@ pub fn resume_multi_agent_version(item: &RolloutItem) -> Option for ResponseItemEnvelope { #[derive(Debug, Clone)] pub enum RolloutItem { SessionMeta(Box), + /// Complete dynamic-tool catalog for subsequent model steps, including an empty catalog. + DynamicToolsUpdated { + thread_id: ThreadId, + tools: Vec, + }, ResponseItem(ResponseItemEnvelope), InterAgentCommunication(InterAgentCommunication), InterAgentCommunicationMetadata { @@ -446,17 +452,7 @@ impl InitialHistory { } pub fn get_dynamic_tools(&self) -> Option> { - match self { - Self::New | Self::Cleared => None, - Self::Resumed(resumed) => resumed.history.iter().find_map(|item| match item { - RolloutItem::SessionMeta(meta_line) => meta_line.meta.dynamic_tools.clone(), - _ => None, - }), - Self::Forked(items) => items.iter().find_map(|item| match item { - RolloutItem::SessionMeta(meta_line) => meta_line.meta.dynamic_tools.clone(), - _ => None, - }), - } + latest_dynamic_tools(self.get_rollout_items()).map(<[_]>::to_vec) } pub fn get_selected_capability_roots(&self) -> Vec { @@ -538,6 +534,31 @@ fn session_cwd_from_items(items: &[RolloutItem]) -> Option { }) } +/// Returns the latest catalog, preserving explicit clears and falling back to creation metadata. +pub fn latest_dynamic_tools(items: &[RolloutItem]) -> Option<&[DynamicToolSpec]> { + let metadata = items.iter().find_map(|item| match item { + RolloutItem::SessionMeta(meta) => Some(&meta.meta), + _ => None, + }); + items + .iter() + .rev() + .find_map(|item| { + let (thread_id, tools) = match item { + RolloutItem::DynamicToolsUpdated { thread_id, tools } => (thread_id, tools), + RolloutItem::Compacted(compacted) => { + let snapshot = compacted.resume_metadata.as_ref()?.dynamic_tools.as_ref()?; + (&snapshot.thread_id, &snapshot.tools) + } + _ => return None, + }; + metadata + .is_none_or(|meta| meta.id == *thread_id) + .then_some(tools.as_slice()) + }) + .or_else(|| metadata.and_then(|meta| meta.dynamic_tools.as_deref())) +} + /// Returns a thread's latest plugin selection, with a turn-context fallback. /// /// Forked history may contain ancestor snapshots, and compaction may append a diff --git a/codex-rs/history/src/rollout_payload.rs b/codex-rs/history/src/rollout_payload.rs index 370e8c8c82..301c7d3c06 100644 --- a/codex-rs/history/src/rollout_payload.rs +++ b/codex-rs/history/src/rollout_payload.rs @@ -34,6 +34,9 @@ pub(super) enum RolloutItemWire<'a> { SessionMeta { payload: Cow<'a, SessionMetaLine>, }, + DynamicToolsUpdated { + payload: DynamicToolsUpdatedPayload<'a>, + }, ResponseItem { payload: Cow<'a, ResponseItem>, #[serde(default, skip_serializing_if = "Option::is_none")] @@ -71,12 +74,24 @@ pub(super) enum RolloutItemWire<'a> { }, } +#[derive(Serialize, Deserialize, JsonSchema)] +pub(super) struct DynamicToolsUpdatedPayload<'a> { + thread_id: codex_protocol::ThreadId, + tools: Cow<'a, [codex_protocol::dynamic_tools::DynamicToolSpec]>, +} + impl<'a> From<&'a RolloutItem> for RolloutItemWire<'a> { fn from(item: &'a RolloutItem) -> Self { match item { RolloutItem::SessionMeta(payload) => Self::SessionMeta { payload: Cow::Borrowed(payload), }, + RolloutItem::DynamicToolsUpdated { thread_id, tools } => Self::DynamicToolsUpdated { + payload: DynamicToolsUpdatedPayload { + thread_id: *thread_id, + tools: Cow::Borrowed(tools), + }, + }, RolloutItem::ResponseItem(envelope) => Self::ResponseItem { payload: Cow::Borrowed(&envelope.item), metadata: envelope.metadata.as_ref().map(Cow::Borrowed), @@ -169,6 +184,10 @@ impl From> for RolloutItem { RolloutItemWire::SessionMeta { payload } => { Self::SessionMeta(Box::new(payload.into_owned())) } + RolloutItemWire::DynamicToolsUpdated { payload } => Self::DynamicToolsUpdated { + thread_id: payload.thread_id, + tools: payload.tools.into_owned(), + }, RolloutItemWire::ResponseItem { payload, metadata } => { if let ( ResponseItem::Message { diff --git a/codex-rs/history/src/tests.rs b/codex-rs/history/src/tests.rs index 2f6424b6b0..272de15d87 100644 --- a/codex-rs/history/src/tests.rs +++ b/codex-rs/history/src/tests.rs @@ -584,6 +584,7 @@ fn compacted_replacement_history_stores_metadata_in_an_aligned_sidecar() -> Resu #[test] fn compacted_resume_metadata_presence_round_trips_empty_values() -> Result<()> { let resume_metadata = CompactionResumeMetadata { + dynamic_tools: None, multi_agent_version: None, last_started_turn_id: None, turn_attribution: None, @@ -695,6 +696,7 @@ fn compacted_metadata_remains_compatible_with_legacy_response_item_readers() -> compaction_response_id: None, latest_token_usage_record: None, resume_metadata: Some(CompactionResumeMetadata { + dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: Some("turn-1".to_string()), turn_attribution: None, @@ -848,7 +850,7 @@ fn rollout_item_variants_preserve_existing_payload_shapes() -> Result<()> { fn rollout_item_schema_matches_tagged_payload_and_sibling_metadata() -> Result<()> { let schema = serde_json::to_value(schemars::schema_for!(RolloutItem))?; let variants = schema["oneOf"].as_array().expect("rollout variants"); - assert_eq!(variants.len(), 12); + assert_eq!(variants.len(), 13); for variant in variants { let required = variant["required"].as_array().expect("required fields"); @@ -1051,3 +1053,81 @@ fn multi_agent_version_uses_compaction_metadata_without_turn_context() -> Result ); Ok(()) } + +#[test] +fn dynamic_tools_updates_survive_replay_and_explicit_clear() -> Result<()> { + let tool = codex_protocol::dynamic_tools::DynamicToolSpec::Function( + codex_protocol::dynamic_tools::DynamicToolFunctionSpec { + name: "check_release".into(), + description: "Check the release.".into(), + input_schema: json!({"type": "object", "properties": {}}), + defer_loading: false, + }, + ); + let thread_id = ThreadId::new(); + let mut items = vec![RolloutItem::SessionMeta(Box::new(SessionMetaLine { + meta: SessionMeta { + id: thread_id, + dynamic_tools: Some(vec![tool.clone()]), + ..SessionMeta::default() + }, + git: None, + }))]; + let mut replacement = tool.clone(); + if let codex_protocol::dynamic_tools::DynamicToolSpec::Function(tool) = &mut replacement { + tool.name = "check_deployment".into(); + } + let checkpoint = |thread_id: ThreadId, tools: Vec| -> Result { + Ok(RolloutItem::Compacted(Box::new(serde_json::from_value( + json!({ + "message": "summary", + "replacement_history": [], + "window_number": 1, + "resume_metadata": { + "dynamic_tools": {"thread_id": thread_id, "tools": tools} + } + }), + )?))) + }; + items.push(checkpoint(ThreadId::new(), vec![replacement.clone()])?); + assert_eq!( + latest_dynamic_tools(&items), + Some(std::slice::from_ref(&tool)) + ); + for expected in [vec![replacement], Vec::new()] { + for item in [ + checkpoint(thread_id, expected.clone())?, + RolloutItem::DynamicToolsUpdated { + thread_id, + tools: expected.clone(), + }, + ] { + items.push(checkpoint(thread_id, vec![tool.clone()])?); + items.push(item); + // Metadata-only writes and inherited records cannot resurrect a replaced catalog. + items.push(items[0].clone()); + items.push(RolloutItem::DynamicToolsUpdated { + thread_id: ThreadId::new(), + tools: vec![tool.clone()], + }); + items.push(checkpoint(ThreadId::new(), vec![tool.clone()])?); + let serialized = serde_json::to_vec(&items)?; + let replayed: Vec = serde_json::from_slice(&serialized)?; + assert_eq!( + InitialHistory::Forked(replayed.clone()).get_dynamic_tools(), + Some(expected.clone()) + ); + assert_eq!( + InitialHistory::Resumed(ResumedHistory { + history_revision: None, + conversation_id: thread_id, + history: Arc::new(replayed), + rollout_path: None, + }) + .get_dynamic_tools(), + Some(expected.clone()) + ); + } + } + Ok(()) +} diff --git a/codex-rs/memories/write/src/phase1.rs b/codex-rs/memories/write/src/phase1.rs index cddb20dadc..f5a5b42044 100644 --- a/codex-rs/memories/write/src/phase1.rs +++ b/codex-rs/memories/write/src/phase1.rs @@ -405,7 +405,8 @@ mod job { RolloutItem::InterAgentCommunication(communication) => { Some(communication.to_model_input_item()) } - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::InterAgentCommunicationMetadata { .. } | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) diff --git a/codex-rs/memories/write/src/rollout_input.rs b/codex-rs/memories/write/src/rollout_input.rs index e0f9e643d1..fa86e25305 100644 --- a/codex-rs/memories/write/src/rollout_input.rs +++ b/codex-rs/memories/write/src/rollout_input.rs @@ -112,7 +112,8 @@ pub(crate) fn serialize_tiered_input( let item = match item { RolloutItem::ResponseItem(item) => sanitize_response_item_for_memories(&item.item), RolloutItem::InterAgentCommunication(message) => Some(message.to_model_input_item()), - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::InterAgentCommunicationMetadata { .. } | RolloutItem::Compacted(_) | RolloutItem::TurnContext(_) diff --git a/codex-rs/protocol/src/protocol.rs b/codex-rs/protocol/src/protocol.rs index 696549842a..1e3efd5da3 100644 --- a/codex-rs/protocol/src/protocol.rs +++ b/codex-rs/protocol/src/protocol.rs @@ -778,6 +778,13 @@ pub enum Op { reply: Option>>, }, + /// Replace the dynamic tool catalog without starting a turn. + #[debug("ReplaceDynamicTools")] + ReplaceDynamicTools { + tools: Vec, + reply: oneshot::Sender>, + }, + /// Update only the named running turn, without changing future settings. /// The reply reports the actual publication or why it did not occur. #[debug("TurnSettings {{ turn_id: {turn_id:?}, requested_approvals_reviewer: {:?} }}", update.approvals_reviewer)] @@ -1096,6 +1103,7 @@ impl Op { Self::RecoverTurn { .. } => "recover_turn", Self::SuspendTurnAndShutdown { .. } => "suspend_turn_and_shutdown", Self::ThreadSettings { .. } => "thread_settings", + Self::ReplaceDynamicTools { .. } => "replace_dynamic_tools", Self::TurnSettings { .. } => "turn_settings", Self::InterAgentCommunication { .. } => "inter_agent_communication", Self::ExecApproval { .. } => "exec_approval", diff --git a/codex-rs/rollout/src/list.rs b/codex-rs/rollout/src/list.rs index 4742dd864a..22d1399840 100644 --- a/codex-rs/rollout/src/list.rs +++ b/codex-rs/rollout/src/list.rs @@ -1059,6 +1059,7 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result { // Not included in `head`; skip. } @@ -1138,6 +1139,7 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result {} } @@ -1232,6 +1234,7 @@ pub async fn read_session_meta_line(path: &Path) -> io::Result | RolloutItem::WorldState(_) | RolloutItem::RealtimeItem(_) | RolloutItem::RetainedContext(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SecurityRiskScore(_) | RolloutItem::EventMsg(_) => {} } diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index 93f967217b..21becec430 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -84,6 +84,7 @@ pub fn builder_from_items( | RolloutItem::RealtimeItem(_) | RolloutItem::TokenUsageRecord(_) | RolloutItem::RetainedContext(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SecurityRiskScore(_) | RolloutItem::EventMsg(_) => None, }) && let Some(builder) = builder_from_session_meta(session_meta, rollout_path) @@ -181,6 +182,7 @@ pub async fn extract_metadata_from_rollout( | RolloutItem::RealtimeItem(_) | RolloutItem::TokenUsageRecord(_) | RolloutItem::RetainedContext(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SecurityRiskScore(_) | RolloutItem::EventMsg(_) => None, }), diff --git a/codex-rs/rollout/src/persistence_metrics.rs b/codex-rs/rollout/src/persistence_metrics.rs index bea161d9df..2566567b8b 100644 --- a/codex-rs/rollout/src/persistence_metrics.rs +++ b/codex-rs/rollout/src/persistence_metrics.rs @@ -253,6 +253,7 @@ impl Write for CountingWriter { fn rollout_item_type(item: &RolloutItem) -> String { match item { + RolloutItem::DynamicToolsUpdated { .. } => "dynamic_tools_updated".to_string(), RolloutItem::SessionMeta(_) => "session_meta".to_string(), RolloutItem::ResponseItem(item) => response_item_type(item).to_string(), RolloutItem::InterAgentCommunication(_) => "inter_agent_communication".to_string(), diff --git a/codex-rs/rollout/src/policy.rs b/codex-rs/rollout/src/policy.rs index 93a1e3a5d8..b49049f6e9 100644 --- a/codex-rs/rollout/src/policy.rs +++ b/codex-rs/rollout/src/policy.rs @@ -45,6 +45,7 @@ pub fn persisted_rollout_item( | RolloutItem::WorldState(_) | RolloutItem::RetainedContext(_) | RolloutItem::SecurityRiskScore(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SessionMeta(_) => Some(Cow::Borrowed(item)), } } diff --git a/codex-rs/rollout/src/recorder.rs b/codex-rs/rollout/src/recorder.rs index 6d6756e810..7298c41888 100644 --- a/codex-rs/rollout/src/recorder.rs +++ b/codex-rs/rollout/src/recorder.rs @@ -2260,7 +2260,8 @@ async fn resume_candidate_matches_cwd( if let Ok((items, _, _)) = RolloutRecorder::load_rollout_items(rollout_path).await && let Some(latest_turn_context_cwd) = items.iter().rev().find_map(|item| match item { RolloutItem::TurnContext(turn_context) => Some(&turn_context.cwd), - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_) | RolloutItem::InterAgentCommunicationMetadata { .. } diff --git a/codex-rs/rollout/src/search.rs b/codex-rs/rollout/src/search.rs index 439271830a..9f782c7e59 100644 --- a/codex-rs/rollout/src/search.rs +++ b/codex-rs/rollout/src/search.rs @@ -335,7 +335,8 @@ fn conversation_text_from_item(item: &RolloutItem) -> Option { Some(text) } } - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::TurnContext(_) | RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) diff --git a/codex-rs/state/src/extract.rs b/codex-rs/state/src/extract.rs index a501301553..80ce0e086a 100644 --- a/codex-rs/state/src/extract.rs +++ b/codex-rs/state/src/extract.rs @@ -39,7 +39,9 @@ pub fn apply_rollout_item( | RolloutItem::InterAgentCommunicationMetadata { .. } => {} RolloutItem::Compacted(_) => {} RolloutItem::WorldState(_) => {} - RolloutItem::RetainedContext(_) | RolloutItem::SecurityRiskScore(_) => {} + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::RetainedContext(_) + | RolloutItem::SecurityRiskScore(_) => {} RolloutItem::RealtimeItem(_) => {} RolloutItem::TokenUsageRecord(_) => {} } @@ -73,6 +75,7 @@ pub fn rollout_item_affects_thread_metadata(item: &RolloutItem) -> bool { | RolloutItem::Compacted(_) | RolloutItem::RealtimeItem(_) | RolloutItem::RetainedContext(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SecurityRiskScore(_) | RolloutItem::TokenUsageRecord(_) | RolloutItem::WorldState(_) => false, diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index c72a9222f7..59f49001af 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -1404,6 +1404,7 @@ pub(super) fn extract_memory_mode(items: &[RolloutItem]) -> Option { | RolloutItem::WorldState(_) | RolloutItem::RealtimeItem(_) | RolloutItem::RetainedContext(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SecurityRiskScore(_) | RolloutItem::TokenUsageRecord(_) | RolloutItem::EventMsg(_) => None, diff --git a/codex-rs/thread-manager-sample/src/main.rs b/codex-rs/thread-manager-sample/src/main.rs index 56deb71fbc..1ced86ef11 100644 --- a/codex-rs/thread-manager-sample/src/main.rs +++ b/codex-rs/thread-manager-sample/src/main.rs @@ -346,6 +346,7 @@ async fn new_config( codex_home, history: History::default(), ephemeral: true, + allow_persistent_dynamic_tool_updates: false, extra_config: None, file_opener: UriBasedFileOpener::VsCode, codex_self_exe: arg0_paths.codex_self_exe, diff --git a/codex-rs/thread-store/src/local/model_context_tests.rs b/codex-rs/thread-store/src/local/model_context_tests.rs index 25fca4e638..d3feb44f2d 100644 --- a/codex-rs/thread-store/src/local/model_context_tests.rs +++ b/codex-rs/thread-store/src/local/model_context_tests.rs @@ -273,6 +273,7 @@ async fn fork_version_stops_before_older_segments_once_resolved() { unreachable!(); }; compacted.resume_metadata = Some(codex_rollout::CompactionResumeMetadata { + dynamic_tools: None, multi_agent_version: Some(MultiAgentVersion::V2), last_started_turn_id: None, turn_attribution: None, diff --git a/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs b/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs index 6a8ce712ea..5d7f336938 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs @@ -290,6 +290,7 @@ impl LegacyRolloutCanonicalizer { } item @ (RolloutItem::InterAgentCommunicationMetadata { .. } | RolloutItem::TurnContext(_) + | RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::TokenUsageRecord(_) | RolloutItem::RealtimeItem(_) | RolloutItem::RetainedContext(_) diff --git a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs index 1a884e8350..c7a6b3a0ea 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs @@ -159,7 +159,9 @@ impl RollbackPlanner { self.pending_delivery_boundary = None; match &line.item { - RolloutItem::SessionMeta(_) => self.record_boundaries[index] = None, + RolloutItem::DynamicToolsUpdated { .. } | RolloutItem::SessionMeta(_) => { + self.record_boundaries[index] = None + } RolloutItem::ResponseItem(response) => { if matches!(&response.item, ResponseItem::Message { role, .. } if role == "assistant") || matches!(&response.item, ResponseItem::FunctionCall { .. }) diff --git a/codex-rs/thread-store/src/local/rollout_migration/rollback_replay.rs b/codex-rs/thread-store/src/local/rollout_migration/rollback_replay.rs index ac23e9c577..92a0c6c0b2 100644 --- a/codex-rs/thread-store/src/local/rollout_migration/rollback_replay.rs +++ b/codex-rs/thread-store/src/local/rollout_migration/rollback_replay.rs @@ -71,7 +71,8 @@ impl ModelReplayPlanner { RolloutItem::ResponseItem(response) if rollback::counts_as_boundary(&response.item) => { ReplayRecord::UserBoundary } - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::ResponseItem(_) | RolloutItem::EventMsg(_) | RolloutItem::InterAgentCommunicationMetadata { .. } diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index 7a75565a68..b81862a4a9 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -338,7 +338,8 @@ impl ThreadMetadataSync { update.approval_mode = Some(settings.approval_policy); update.permission_profile = Some(settings.permission_profile.clone()); } - RolloutItem::SessionMeta(_) + RolloutItem::DynamicToolsUpdated { .. } + | RolloutItem::SessionMeta(_) | RolloutItem::EventMsg(_) | RolloutItem::ResponseItem(_) | RolloutItem::InterAgentCommunication(_)