agent/codex
2
0
Fork 0
mirror of https://github.com/pchuan98/codex.git synced 2026-07-01 00:31:56 +08:00
Code Issues Packages Projects Releases Wiki Activity

[codex] Remove redundant SQLite dynamic tool storage (#24819)

Browse Source 
## Why

Dynamic tools are defined at thread start and already stored in rollout
`SessionMeta`, which restores resumed and forked sessions. Persisting
the same tools through SQLite creates a second runtime persistence path
that is unnecessary prework for the explicit namespace refactor.

## What changed

- Restore missing thread-start dynamic tools directly from rollout
history, including when SQLite is enabled.
- Remove SQLite dynamic-tool reads, writes, backfill, and thread
metadata patch plumbing.
- Add SQLite-enabled resume integration coverage that verifies a
rollout-defined dynamic tool is still sent after resume.

## Compatibility

The existing `thread_dynamic_tools` table is intentionally not dropped
even though it's now unused. Older Codex binaries are allowed to open
databases migrated by newer binaries and still reference this table;
dropping it would break that mixed-version path. See
[here](https://github.com/openai/codex/blob/main/codex-rs/state/src/migrations.rs#L10-L11).

## Verification

- `just test -p codex-state -p codex-rollout -p codex-thread-store`
- `just test -p codex-core --test all
resume_restores_dynamic_tools_from_rollout_with_sqlite_enabled`
This commit is contained in:
sayan-oai
2026-05-27 17:57:32 -07:00
committed by GitHub
Unverified
parent 0db49a7e6a
commit 304d15cab0
9 changed files with 101 additions and 260 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/core/src/session/mod.rs
+2 -30
View File
@@ -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::<LocalThreadStore>()
{
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
};
codex-rs/core/tests/suite/sqlite_state.rs
+99 -38
View File
@@ -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(())
}
codex-rs/rollout/src/metadata.rs
-20
View File
@@ -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) => {
codex-rs/rollout/src/state_db.rs
-47
View File
@@ -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<Vec<DynamicToolSpec>> {
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()
);
}
}
codex-rs/state/src/runtime.rs
-1
View File
@@ -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;
codex-rs/state/src/runtime/threads.rs
-102
View File
@@ -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<Option<Vec<DynamicToolSpec>>> {
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::<Value>(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<Option<Vec<DynamicToolSpec>>> {
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<String> {
items.iter().rev().find_map(|item| match item {
RolloutItem::SessionMeta(meta_line) => meta_line.meta.memory_mode.clone(),
codex-rs/thread-store/src/local/update_thread_metadata.rs
-9
View File
@@ -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<T: serde::Serialize>(value: &T) -> String {
codex-rs/thread-store/src/thread_metadata_sync.rs
-7
View File
@@ -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 {
codex-rs/thread-store/src/types.rs
-6
View File
@@ -529,8 +529,6 @@ pub struct ThreadMetadataPatch {
pub git_info: Option<GitInfoPatch>,
/// Thread memory behavior.
pub memory_mode: Option<MemoryMode>,
/// Dynamic tools available to this thread.
pub dynamic_tools: Option<Vec<DynamicToolSpec>>,
}
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()
}
}