diff --git a/codex-rs/core/src/session/mod.rs b/codex-rs/core/src/session/mod.rs index 245047c92..512f8acc7 100644 --- a/codex-rs/core/src/session/mod.rs +++ b/codex-rs/core/src/session/mod.rs @@ -549,37 +549,9 @@ impl Codex { .or_else(|| conversation_history.get_base_instructions().map(|s| s.text)) .unwrap_or_else(|| model_info.get_model_instructions(config.personality)); - // Respect thread-start tools. When missing (resumed/forked threads), read from the db - // first, then fall back to rollout-file tools. - let persisted_tools = if dynamic_tools.is_empty() { - let thread_id = match &conversation_history { - InitialHistory::Resumed(resumed) => Some(resumed.conversation_id), - InitialHistory::Forked(_) => conversation_history.forked_from_id(), - InitialHistory::New | InitialHistory::Cleared => None, - }; - match thread_id { - Some(thread_id) => { - let state_db_ctx = if config.ephemeral { - None - } else if let Some(local_store) = - thread_store.as_any().downcast_ref::() - { - local_store.state_db().await - } else { - None - }; - state_db::get_dynamic_tools(state_db_ctx.as_deref(), thread_id, "codex_spawn") - .await - } - None => None, - } - } else { - None - }; + // Dynamic tools are defined at thread start and persisted in rollout session metadata. let dynamic_tools = if dynamic_tools.is_empty() { - persisted_tools - .or_else(|| conversation_history.get_dynamic_tools()) - .unwrap_or_default() + conversation_history.get_dynamic_tools().unwrap_or_default() } else { dynamic_tools }; diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index 99b7fb0ec..8cbc46e1b 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -93,6 +93,104 @@ async fn new_thread_is_recorded_in_state_db() -> Result<()> { Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn resume_restores_dynamic_tools_from_rollout_with_sqlite_enabled() -> Result<()> { + let server = start_mock_server().await; + let mock = mount_sse_sequence( + &server, + vec![ + responses::sse(vec![ev_response_created("resp-1"), ev_completed("resp-1")]), + responses::sse(vec![ev_response_created("resp-2"), ev_completed("resp-2")]), + ], + ) + .await; + + let dynamic_tool = DynamicToolSpec { + namespace: None, + name: "resume_lookup".to_string(), + description: "Look up a value after resume.".to_string(), + input_schema: json!({ + "type": "object", + "properties": { "query": { "type": "string" } }, + "required": ["query"], + "additionalProperties": false, + }), + defer_loading: false, + }; + let mut builder = test_codex().with_config(|config| { + config + .features + .enable(Feature::Sqlite) + .expect("test config should allow feature update"); + }); + let base_test = builder.build(&server).await?; + let started = base_test + .thread_manager + .start_thread_with_tools( + base_test.config.clone(), + vec![dynamic_tool.clone()], + /*persist_extended_history*/ false, + ) + .await?; + let rollout_path = started + .session_configured + .rollout_path + .clone() + .expect("rollout path"); + + started + .thread + .submit(Op::UserInput { + environments: None, + items: vec![UserInput::Text { + text: "persist this thread".to_string(), + text_elements: Vec::new(), + }], + final_output_json_schema: None, + responsesapi_client_metadata: None, + additional_context: Default::default(), + thread_settings: Default::default(), + }) + .await?; + wait_for_event(&started.thread, |event| { + matches!(event, EventMsg::TurnComplete(_)) + }) + .await; + + let mut resume_builder = test_codex().with_config(|config| { + config + .features + .enable(Feature::Sqlite) + .expect("test config should allow feature update"); + }); + let resumed = resume_builder + .resume(&server, base_test.home.clone(), rollout_path) + .await?; + resumed.submit_turn("use the restored tool").await?; + + let requests = mock.requests(); + assert_eq!(requests.len(), 2); + let resumed_body = requests[1].body_json(); + let tools = resumed_body + .get("tools") + .and_then(serde_json::Value::as_array) + .expect("resumed request tools"); + let restored_tool = tools + .iter() + .find(|tool| tool.get("name") == Some(&json!(dynamic_tool.name.as_str()))) + .expect("dynamic tool should be restored from rollout metadata"); + assert_eq!( + restored_tool.get("description"), + Some(&json!(dynamic_tool.description.as_str())) + ); + assert_eq!( + restored_tool.get("parameters"), + Some(&dynamic_tool.input_schema) + ); + + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn backfill_scans_existing_rollouts() -> Result<()> { let server = start_mock_server().await; @@ -102,32 +200,6 @@ async fn backfill_scans_existing_rollouts() -> Result<()> { let rollout_rel_path = format!("sessions/2026/01/27/rollout-2026-01-27T12-00-00-{uuid}.jsonl"); let rollout_rel_path_for_hook = rollout_rel_path.clone(); - let dynamic_tools = vec![ - DynamicToolSpec { - namespace: Some("codex_app".to_string()), - name: "geo_lookup".to_string(), - description: "lookup a city".to_string(), - input_schema: json!({ - "type": "object", - "required": ["city"], - "properties": { "city": { "type": "string" } } - }), - defer_loading: true, - }, - DynamicToolSpec { - namespace: None, - name: "weather_lookup".to_string(), - description: "lookup weather".to_string(), - input_schema: json!({ - "type": "object", - "required": ["zip"], - "properties": { "zip": { "type": "string" } } - }), - defer_loading: false, - }, - ]; - let dynamic_tools_for_hook = dynamic_tools.clone(); - let mut builder = test_codex() .with_pre_build_hook(move |codex_home| { let rollout_path = codex_home.join(&rollout_rel_path_for_hook); @@ -150,7 +222,7 @@ async fn backfill_scans_existing_rollouts() -> Result<()> { agent_role: None, model_provider: None, base_instructions: None, - dynamic_tools: Some(dynamic_tools_for_hook), + dynamic_tools: None, memory_mode: None, }, git: None, @@ -217,17 +289,6 @@ async fn backfill_scans_existing_rollouts() -> Result<()> { assert_eq!(metadata.model_provider, default_provider); assert!(metadata.first_user_message.is_some()); - let mut stored_tools = None; - for _ in 0..40 { - stored_tools = db.get_dynamic_tools(thread_id).await?; - if stored_tools.is_some() { - break; - } - tokio::time::sleep(Duration::from_millis(25)).await; - } - let stored_tools = stored_tools.expect("dynamic tools should be stored"); - assert_eq!(stored_tools, dynamic_tools); - Ok(()) } diff --git a/codex-rs/rollout/src/metadata.rs b/codex-rs/rollout/src/metadata.rs index 2dd2df3a4..00c6e3e63 100644 --- a/codex-rs/rollout/src/metadata.rs +++ b/codex-rs/rollout/src/metadata.rs @@ -1,6 +1,5 @@ use crate::ARCHIVED_SESSIONS_SUBDIR; use crate::SESSIONS_SUBDIR; -use crate::list; use crate::list::parse_timestamp_uuid_from_filename; use crate::recorder::RolloutRecorder; use crate::state_db::normalize_cwd_for_state_db; @@ -286,25 +285,6 @@ pub(crate) async fn backfill_sessions_with_lease( continue; } stats.upserted = stats.upserted.saturating_add(1); - if let Ok(meta_line) = list::read_session_meta_line(&rollout.path).await { - if let Err(err) = runtime - .persist_dynamic_tools( - meta_line.meta.id, - meta_line.meta.dynamic_tools.as_deref(), - ) - .await - { - warn!( - "failed to backfill dynamic tools {}: {err}", - rollout.path.display() - ); - } - } else { - warn!( - "failed to read session meta for dynamic tools {}", - rollout.path.display() - ); - } } } Err(err) => { diff --git a/codex-rs/rollout/src/state_db.rs b/codex-rs/rollout/src/state_db.rs index ea087e5c6..2b6153484 100644 --- a/codex-rs/rollout/src/state_db.rs +++ b/codex-rs/rollout/src/state_db.rs @@ -8,7 +8,6 @@ use crate::sqlite_metrics; use chrono::DateTime; use chrono::Utc; use codex_protocol::ThreadId; -use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::protocol::RolloutItem; use codex_protocol::protocol::SessionSource; pub use codex_state::LogEntry; @@ -454,37 +453,6 @@ pub async fn find_rollout_path_by_id( }) } -/// Get dynamic tools for a thread id using SQLite. -pub async fn get_dynamic_tools( - context: Option<&codex_state::StateRuntime>, - thread_id: ThreadId, - stage: &str, -) -> Option> { - let ctx = context?; - match ctx.get_dynamic_tools(thread_id).await { - Ok(tools) => tools, - Err(err) => { - warn!("state db get_dynamic_tools failed during {stage}: {err}"); - None - } - } -} - -/// Persist dynamic tools for a thread id using SQLite, if none exist yet. -pub async fn persist_dynamic_tools( - context: Option<&codex_state::StateRuntime>, - thread_id: ThreadId, - tools: Option<&[DynamicToolSpec]>, - stage: &str, -) { - let Some(ctx) = context else { - return; - }; - if let Err(err) = ctx.persist_dynamic_tools(thread_id, tools).await { - warn!("state db persist_dynamic_tools failed during {stage}: {err}"); - } -} - pub async fn mark_thread_memory_mode_polluted( context: Option<&codex_state::StateRuntime>, thread_id: ThreadId, @@ -570,21 +538,6 @@ pub async fn reconcile_rollout( "state db reconcile_rollout memory_mode update failed {}: {err}", rollout_path.display() ); - return; - } - if let Ok(meta_line) = crate::list::read_session_meta_line(rollout_path).await { - persist_dynamic_tools( - Some(ctx), - meta_line.meta.id, - meta_line.meta.dynamic_tools.as_deref(), - "reconcile_rollout", - ) - .await; - } else { - warn!( - "state db reconcile_rollout missing session meta {}", - rollout_path.display() - ); } } diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index ba4102129..b08545d7a 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -33,7 +33,6 @@ use crate::telemetry::DbTelemetry; use chrono::DateTime; use chrono::Utc; use codex_protocol::ThreadId; -use codex_protocol::dynamic_tools::DynamicToolSpec; use codex_protocol::protocol::RolloutItem; use log::LevelFilter; use serde_json::Value; diff --git a/codex-rs/state/src/runtime/threads.rs b/codex-rs/state/src/runtime/threads.rs index 63bfe09f5..c67056ab5 100644 --- a/codex-rs/state/src/runtime/threads.rs +++ b/codex-rs/state/src/runtime/threads.rs @@ -74,40 +74,6 @@ WHERE id = ? AND preview = '' Ok(result.rows_affected() > 0) } - /// Get dynamic tools for a thread, if present. - pub async fn get_dynamic_tools( - &self, - thread_id: ThreadId, - ) -> anyhow::Result>> { - let rows = sqlx::query( - r#" -SELECT namespace, name, description, input_schema, defer_loading -FROM thread_dynamic_tools -WHERE thread_id = ? -ORDER BY position ASC - "#, - ) - .bind(thread_id.to_string()) - .fetch_all(self.pool.as_ref()) - .await?; - if rows.is_empty() { - return Ok(None); - } - let mut tools = Vec::with_capacity(rows.len()); - for row in rows { - let input_schema: String = row.try_get("input_schema")?; - let input_schema = serde_json::from_str::(input_schema.as_str())?; - tools.push(DynamicToolSpec { - namespace: row.try_get("namespace")?, - name: row.try_get("name")?, - description: row.try_get("description")?, - input_schema, - defer_loading: row.try_get("defer_loading")?, - }); - } - Ok(Some(tools)) - } - /// Persist or replace the directional parent-child edge for a spawned thread. pub async fn upsert_thread_spawn_edge( &self, @@ -821,54 +787,6 @@ ON CONFLICT(id) DO UPDATE SET Ok(()) } - /// Persist dynamic tools for a thread if none have been stored yet. - /// - /// Dynamic tools are defined at thread start and should not change afterward. - /// This only writes the first time we see tools for a given thread. - pub async fn persist_dynamic_tools( - &self, - thread_id: ThreadId, - tools: Option<&[DynamicToolSpec]>, - ) -> anyhow::Result<()> { - let Some(tools) = tools else { - return Ok(()); - }; - if tools.is_empty() { - return Ok(()); - } - let thread_id = thread_id.to_string(); - let mut tx = self.pool.begin().await?; - for (idx, tool) in tools.iter().enumerate() { - let position = i64::try_from(idx).unwrap_or(i64::MAX); - let input_schema = serde_json::to_string(&tool.input_schema)?; - sqlx::query( - r#" -INSERT INTO thread_dynamic_tools ( - thread_id, - position, - namespace, - name, - description, - input_schema, - defer_loading -) VALUES (?, ?, ?, ?, ?, ?, ?) -ON CONFLICT(thread_id, position) DO NOTHING - "#, - ) - .bind(thread_id.as_str()) - .bind(position) - .bind(tool.namespace.as_deref()) - .bind(tool.name.as_str()) - .bind(tool.description.as_str()) - .bind(input_schema) - .bind(tool.defer_loading) - .execute(&mut *tx) - .await?; - } - tx.commit().await?; - Ok(()) - } - /// Apply rollout items incrementally using the underlying database. pub async fn apply_rollout_items( &self, @@ -898,8 +816,6 @@ ON CONFLICT(thread_id, position) DO NOTHING if let Some(updated_at) = updated_at { metadata.updated_at = updated_at; } - // Keep the thread upsert before dynamic tools to satisfy the foreign key constraint: - // thread_dynamic_tools.thread_id -> threads.id. let upsert_result = if existing_metadata.is_none() { self.upsert_thread_with_creation_memory_mode(&metadata, new_thread_memory_mode) .await @@ -914,14 +830,6 @@ ON CONFLICT(thread_id, position) DO NOTHING { return Err(err); } - let dynamic_tools = extract_dynamic_tools(items); - if let Some(dynamic_tools) = dynamic_tools - && let Err(err) = self - .persist_dynamic_tools(builder.id, dynamic_tools.as_deref()) - .await - { - return Err(err); - } Ok(()) } @@ -1039,16 +947,6 @@ SELECT ); } -pub(super) fn extract_dynamic_tools(items: &[RolloutItem]) -> Option>> { - items.iter().find_map(|item| match item { - RolloutItem::SessionMeta(meta_line) => Some(meta_line.meta.dynamic_tools.clone()), - RolloutItem::ResponseItem(_) - | RolloutItem::Compacted(_) - | RolloutItem::TurnContext(_) - | RolloutItem::EventMsg(_) => None, - }) -} - pub(super) fn extract_memory_mode(items: &[RolloutItem]) -> Option { items.iter().rev().find_map(|item| match item { RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(), diff --git a/codex-rs/thread-store/src/local/update_thread_metadata.rs b/codex-rs/thread-store/src/local/update_thread_metadata.rs index 16f9f79e0..612328f4c 100644 --- a/codex-rs/thread-store/src/local/update_thread_metadata.rs +++ b/codex-rs/thread-store/src/local/update_thread_metadata.rs @@ -309,14 +309,6 @@ async fn apply_metadata_update( message: format!("failed to update memory mode for {thread_id}: {err}"), })?; } - if let Some(dynamic_tools) = patch.dynamic_tools { - state_db - .persist_dynamic_tools(thread_id, Some(dynamic_tools.as_slice())) - .await - .map_err(|err| ThreadStoreError::Internal { - message: format!("failed to update dynamic tools for {thread_id}: {err}"), - })?; - } Ok(()) } .await @@ -393,7 +385,6 @@ fn has_observed_metadata_facts(patch: &ThreadMetadataPatch) -> bool { || patch.sandbox_policy.is_some() || patch.token_usage.is_some() || patch.first_user_message.is_some() - || patch.dynamic_tools.is_some() } fn enum_to_string(value: &T) -> String { diff --git a/codex-rs/thread-store/src/thread_metadata_sync.rs b/codex-rs/thread-store/src/thread_metadata_sync.rs index edc78a36c..408c792da 100644 --- a/codex-rs/thread-store/src/thread_metadata_sync.rs +++ b/codex-rs/thread-store/src/thread_metadata_sync.rs @@ -61,8 +61,6 @@ impl ThreadMetadataSync { } else { None }; - let dynamic_tools = - (!params.dynamic_tools.is_empty()).then(|| params.dynamic_tools.clone()); let update = ThreadMetadataPatch { model_provider: Some(params.metadata.model_provider.clone()), created_at: Some(created_at), @@ -76,7 +74,6 @@ impl ThreadMetadataSync { cli_version: Some(env!("CARGO_PKG_VERSION").to_string()), git_info: git_info.map(git_info_patch_from_observation), memory_mode: Some(params.metadata.memory_mode), - dynamic_tools, ..Default::default() }; Self { @@ -228,9 +225,6 @@ impl ThreadMetadataSync { { update.memory_mode = Some(memory_mode); } - if let Some(dynamic_tools) = meta_line.meta.dynamic_tools.clone() { - update.dynamic_tools = Some(dynamic_tools); - } } RolloutItem::TurnContext(turn_ctx) => { if !self.cwd_seen && !turn_ctx.cwd.as_os_str().is_empty() { @@ -365,7 +359,6 @@ fn update_has_metadata_facts(update: &ThreadMetadataPatch) -> bool { || update.first_user_message.is_some() || update.git_info.is_some() || update.memory_mode.is_some() - || update.dynamic_tools.is_some() } fn git_info_patch_from_observation(git_info: GitInfo) -> GitInfoPatch { diff --git a/codex-rs/thread-store/src/types.rs b/codex-rs/thread-store/src/types.rs index ca75ca1b6..1e26e5dc7 100644 --- a/codex-rs/thread-store/src/types.rs +++ b/codex-rs/thread-store/src/types.rs @@ -529,8 +529,6 @@ pub struct ThreadMetadataPatch { pub git_info: Option, /// Thread memory behavior. pub memory_mode: Option, - /// Dynamic tools available to this thread. - pub dynamic_tools: Option>, } impl ThreadMetadataPatch { @@ -608,9 +606,6 @@ impl ThreadMetadataPatch { if next.memory_mode.is_some() { self.memory_mode = next.memory_mode; } - if next.dynamic_tools.is_some() { - self.dynamic_tools = next.dynamic_tools; - } } pub fn is_empty(&self) -> bool { @@ -636,7 +631,6 @@ impl ThreadMetadataPatch { && self.first_user_message.is_none() && self.git_info.is_none() && self.memory_mode.is_none() - && self.dynamic_tools.is_none() } }