openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-10-28 18:58:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Allow replacing dynamic tools between model steps (#57553)

Browse Source 
## What changed

Add `CodexThread::replace_dynamic_tools` to replace the tool catalog without starting a turn. Updates apply to subsequent model steps and future turns while preserving in-flight calls; an empty list clears the catalog.

Require host opt-in through `allow_persistent_dynamic_tool_updates` for persistent threads. Persist replacements before publishing them, restore the latest catalog on resume, and preserve replacements and explicit clears through compaction. Keep explicit resume overrides separate from the persisted catalog.

## Testing

Add coverage for pending calls during replacement, replacement and clearing across restart, compaction recovery, resume overrides, submission ordering after caller cancellation, host opt-in, and persistence failures.

GitOrigin-RevId: fd2634906895a9b72ae7892ddd9435dced45eccb
This commit is contained in:
Steve Coffey
2026-10-27 19:50:41 +00:00
committed by copyberry
parent 0d6838318f
commit f088e0c4ff
50 changed files with 976 additions and 105 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/app-server-protocol/schema/precomputed/app-server-exports-stable.json.zst
Generated
BIN
View File
Binary file not shown.
codex-rs/app-server-protocol/src/protocol/thread_history.rs
+1
View File
@@ -491,6 +491,7 @@ impl ThreadHistoryBuilder {
| RolloutItem::RealtimeItem(_)
| RolloutItem::RetainedContext(_)
| RolloutItem::SecurityRiskScore(_)
| RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SessionMeta(_) => {}
}
}
codex-rs/app-server-protocol/src/protocol/thread_history_projection.rs
+2 -1
View File
@@ -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 { .. }
codex-rs/core-api/src/lib.rs
+1
View File
@@ -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;
codex-rs/core/src/agent/control/spawn.rs
+2 -1
View File
@@ -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
codex-rs/core/src/agent/control_tests.rs
+5 -2
View File
@@ -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,
codex-rs/core/src/agent/history.rs
+6 -2
View File
@@ -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(_)
codex-rs/core/src/codex_thread.rs
+18
View File
@@ -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<codex_protocol::dynamic_tools::DynamicToolSpec>,
) -> 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.
codex-rs/core/src/config/mod.rs
+5
View File
@@ -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<ExtraConfig>,
@@ -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),
codex-rs/core/src/session/dynamic_tools_tests.rs
+143
View File
@@ -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(), &[]);
}
codex-rs/core/src/session/extension_interruption_tests.rs
+6 -37
View File
@@ -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);
codex-rs/core/src/session/handlers.rs
+4
View File
@@ -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,
codex-rs/core/src/session/mod.rs
+11 -2
View File
@@ -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(),
codex-rs/core/src/session/rollout_reconstruction.rs
+6 -2
View File
@@ -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 { .. }
codex-rs/core/src/session/rollout_reconstruction_tests.rs
+1
View File
@@ -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,
codex-rs/core/src/session/session.rs
+17 -9
View File
@@ -166,8 +166,8 @@ pub(crate) struct SessionConfiguration {
pub(super) thread_source: Option<ThreadSource>,
/// 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<shell::Shell>,
}
@@ -707,12 +707,9 @@ impl Session {
state.session_configuration.originator.clone()
}
pub(crate) async fn dynamic_tools(&self) -> Vec<DynamicToolSpec> {
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::<ReasoningEffortPin>() {
@@ -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 { .. }
codex-rs/core/src/session/tests.rs
+121 -11
View File
@@ -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<ToolSpec>) {
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<Session>, turn_context: Arc<TurnContext>) -> ToolCallRuntime {
async fn test_tool_runtime(
session: Arc<Session>,
turn_context: Arc<TurnContext>,
) -> 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<Session>,
turn_context: &TurnContext,
rx_event: async_channel::Receiver<Event>,
session_source: SessionSource,
) -> (crate::CodexThread, async_channel::Receiver<Submission>) {
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(),
codex-rs/core/src/session/thread_settings.rs
+39
View File
@@ -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<codex_protocol::dynamic_tools::DynamicToolSpec>,
) -> 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;
codex-rs/core/src/session/turn.rs
+2
View File
@@ -1935,6 +1935,7 @@ pub(crate) async fn built_tools(
mcp: &Arc<codex_mcp::McpBinding>,
step_store: &ExtensionData,
prepared_recommendations: PreparedToolRecommendations,
dynamic_tools: &[codex_protocol::dynamic_tools::DynamicToolSpec],
) -> CodexResult<Arc<ToolRouter>> {
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,
)?))
}
codex-rs/core/src/session/turn_context.rs
-4
View File
@@ -368,7 +368,6 @@ pub struct TurnContext {
pub(crate) windows_sandbox_level: WindowsSandboxLevel,
pub(crate) available_models: Vec<ModelPreset>,
pub(crate) final_output_json_schema: Option<Value>,
pub(crate) dynamic_tools: Arc<[DynamicToolSpec]>,
pub(crate) turn_metadata_state: Arc<TurnMetadataState>,
pub(crate) extension_data: Arc<codex_extension_api::ExtensionData>,
pub(crate) turn_timing_state: Arc<TurnTimingState>,
@@ -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()),
codex-rs/core/src/state/session.rs
+5
View File
@@ -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<String>,
/// 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,
codex-rs/core/src/stream_events_utils_tests.rs
+5 -4
View File
@@ -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<Session>, turn_context: Arc<TurnContext>) -> HandleOutputCtx {
async fn output_context(session: Arc<Session>, turn_context: Arc<TurnContext>) -> 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"}]});
codex-rs/core/src/tools/router_tests.rs
+7 -7
View File
@@ -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!(
codex-rs/core/src/tools/spec_plan.rs
+2 -1
View File
@@ -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<ToolRouter> {
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,
codex-rs/core/src/tools/spec_plan_tests.rs
+4
View File
@@ -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");
codex-rs/core/tests/suite/scenarios.rs
+3
View File
@@ -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;
codex-rs/core/tests/suite/scenarios_dynamic_tools.rs
+261
View File
@@ -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(())
}
codex-rs/core/tests/suite/snapshots/all__suite__scenarios__dynamic_tools__dynamic_tools_update_preserves_pending_calls_and_survives_resume.snap
+112
View File
@@ -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: "<PROMPT_CACHE_KEY 1>"
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:
<MODEL_INSTRUCTIONS>
01:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <COLLABORATION_MODE>
02:message/developer:
<MULTI_AGENT_ROLE>
03:message/developer:
<MULTI_AGENT_MODE>
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:
<MODEL_INSTRUCTIONS>
01:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <COLLABORATION_MODE>
02:message/developer:
<MULTI_AGENT_ROLE>
03:message/developer:
<MULTI_AGENT_MODE>
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:
<MODEL_INSTRUCTIONS>
01:message/developer[2]:
[01] <PERMISSIONS_INSTRUCTIONS>
[02] <COLLABORATION_MODE>
02:message/developer:
<MULTI_AGENT_ROLE>
03:message/developer:
<MULTI_AGENT_MODE>
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"}
codex-rs/external-agent-migration/src/sessions/append.rs
+2
View File
@@ -246,6 +246,7 @@ fn source_model_items(items: &[RolloutItem]) -> Option<Vec<SourceModelItem<'_>>>
| 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<Vec<&ResponseItem>> {
| RolloutItem::TokenUsageRecord(_)
| RolloutItem::RealtimeItem(_)
| RolloutItem::RetainedContext(_)
| RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SecurityRiskScore(_) => {}
RolloutItem::ResponseItem(response_item) => model_items.push(&response_item.item),
RolloutItem::EventMsg(
codex-rs/history/src/compaction_resume_metadata.rs
+15 -2
View File
@@ -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<Box<DynamicToolsSnapshot>>,
/// Selected runtime, including when the companion turn context is absent.
pub multi_agent_version: Option<MultiAgentVersion>,
/// Turn identity used to admit continuations after cold resume.
@@ -24,6 +29,13 @@ pub struct CompactionResumeMetadata {
pub previous_turn_settings: Option<PreviousTurnSettings>,
}
/// 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<DynamicToolSpec>,
}
/// 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<MultiAgentVersio
.resume_metadata
.as_ref()
.and_then(|metadata| metadata.multi_agent_version),
RolloutItem::SessionMeta(_)
RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SessionMeta(_)
| RolloutItem::ResponseItem(_)
| RolloutItem::InterAgentCommunication(_)
| RolloutItem::InterAgentCommunicationMetadata { .. }
codex-rs/history/src/lib.rs
+32 -11
View File
@@ -8,6 +8,7 @@ pub use heartbeat::UserInputOrigin;
mod compaction_resume_metadata;
pub use codex_protocol::turn_input::TurnAttribution;
pub use compaction_resume_metadata::CompactionResumeMetadata;
pub use compaction_resume_metadata::DynamicToolsSnapshot;
pub use compaction_resume_metadata::PreviousTurnSettings;
pub use compaction_resume_metadata::resume_multi_agent_version;
@@ -214,6 +215,11 @@ impl Borrow<ResponseItem> for ResponseItemEnvelope {
#[derive(Debug, Clone)]
pub enum RolloutItem {
SessionMeta(Box<SessionMetaLine>),
/// Complete dynamic-tool catalog for subsequent model steps, including an empty catalog.
DynamicToolsUpdated {
thread_id: ThreadId,
tools: Vec<DynamicToolSpec>,
},
ResponseItem(ResponseItemEnvelope),
InterAgentCommunication(InterAgentCommunication),
InterAgentCommunicationMetadata {
@@ -446,17 +452,7 @@ impl InitialHistory {
}
pub fn get_dynamic_tools(&self) -> Option<Vec<DynamicToolSpec>> {
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<SelectedCapabilityRoot> {
@@ -538,6 +534,31 @@ fn session_cwd_from_items(items: &[RolloutItem]) -> Option<PathBuf> {
})
}
/// 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
codex-rs/history/src/rollout_payload.rs
+19
View File
@@ -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<RolloutItemWire<'_>> 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 {
codex-rs/history/src/tests.rs
+81 -1
View File
@@ -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<DynamicToolSpec>| -> Result<RolloutItem> {
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<RolloutItem> = 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(())
}
codex-rs/memories/write/src/phase1.rs
+2 -1
View File
@@ -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(_)
codex-rs/memories/write/src/rollout_input.rs
+2 -1
View File
@@ -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(_)
codex-rs/protocol/src/protocol.rs
+8
View File
@@ -778,6 +778,13 @@ pub enum Op {
reply: Option<oneshot::Sender<CodexResult<()>>>,
},
/// Replace the dynamic tool catalog without starting a turn.
#[debug("ReplaceDynamicTools")]
ReplaceDynamicTools {
tools: Vec<DynamicToolSpec>,
reply: oneshot::Sender<CodexResult<()>>,
},
/// 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",
codex-rs/rollout/src/list.rs
+3
View File
@@ -1059,6 +1059,7 @@ async fn read_head_summary(path: &Path, head_limit: usize) -> io::Result<HeadTai
}
RolloutItem::RetainedContext(_)
| RolloutItem::WorldState(_)
| RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SecurityRiskScore(_) => {
// Not included in `head`; skip.
}
@@ -1138,6 +1139,7 @@ pub async fn read_head_for_summary(path: &Path) -> io::Result<Vec<serde_json::Va
| RolloutItem::WorldState(_)
| RolloutItem::RealtimeItem(_)
| RolloutItem::RetainedContext(_)
| RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SecurityRiskScore(_)
| RolloutItem::EventMsg(_) => {}
}
@@ -1232,6 +1234,7 @@ pub async fn read_session_meta_line(path: &Path) -> io::Result<SessionMetaLine>
| RolloutItem::WorldState(_)
| RolloutItem::RealtimeItem(_)
| RolloutItem::RetainedContext(_)
| RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SecurityRiskScore(_)
| RolloutItem::EventMsg(_) => {}
}
codex-rs/rollout/src/metadata.rs
+2
View File
@@ -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,
}),
codex-rs/rollout/src/persistence_metrics.rs
+1
View File
@@ -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(),
codex-rs/rollout/src/policy.rs
+1
View File
@@ -45,6 +45,7 @@ pub fn persisted_rollout_item(
| RolloutItem::WorldState(_)
| RolloutItem::RetainedContext(_)
| RolloutItem::SecurityRiskScore(_)
| RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SessionMeta(_) => Some(Cow::Borrowed(item)),
}
}
codex-rs/rollout/src/recorder.rs
+2 -1
View File
@@ -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 { .. }
codex-rs/rollout/src/search.rs
+2 -1
View File
@@ -335,7 +335,8 @@ fn conversation_text_from_item(item: &RolloutItem) -> Option<String> {
Some(text)
}
}
RolloutItem::SessionMeta(_)
RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SessionMeta(_)
| RolloutItem::TurnContext(_)
| RolloutItem::EventMsg(_)
| RolloutItem::ResponseItem(_)
codex-rs/state/src/extract.rs
+4 -1
View File
@@ -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,
codex-rs/state/src/runtime/threads.rs
+1
View File
@@ -1404,6 +1404,7 @@ pub(super) fn extract_memory_mode(items: &[RolloutItem]) -> Option<String> {
| RolloutItem::WorldState(_)
| RolloutItem::RealtimeItem(_)
| RolloutItem::RetainedContext(_)
| RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::SecurityRiskScore(_)
| RolloutItem::TokenUsageRecord(_)
| RolloutItem::EventMsg(_) => None,
codex-rs/thread-manager-sample/src/main.rs
+1
View File
@@ -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,
codex-rs/thread-store/src/local/model_context_tests.rs
+1
View File
@@ -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,
codex-rs/thread-store/src/local/rollout_migration/canonicalizer.rs
+1
View File
@@ -290,6 +290,7 @@ impl LegacyRolloutCanonicalizer {
}
item @ (RolloutItem::InterAgentCommunicationMetadata { .. }
| RolloutItem::TurnContext(_)
| RolloutItem::DynamicToolsUpdated { .. }
| RolloutItem::TokenUsageRecord(_)
| RolloutItem::RealtimeItem(_)
| RolloutItem::RetainedContext(_)
codex-rs/thread-store/src/local/rollout_migration/rollback_plan.rs
+3 -1
View File
@@ -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 { .. })
codex-rs/thread-store/src/local/rollout_migration/rollback_replay.rs
+2 -1
View File
@@ -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 { .. }
codex-rs/thread-store/src/thread_metadata_sync.rs
+2 -1
View File
@@ -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(_)