From 414217dc8accc5c0e2542239cbab6eaa815dc605 Mon Sep 17 00:00:00 2001 From: Owen Lin Date: Fri, 10 Jul 2026 18:22:36 +0000 Subject: [PATCH] Add dedicated storage for paginated thread history (#32234) ## Why Paginated thread history needs its own SQLite database to avoid adding lock contention to the main state store. ## What changed - Add the `thread_history_1.sqlite` path and migration scaffolding. - Create tables and pagination indexes for projected turns and items, plus a per-thread projection checkpoint. - Register the database with runtime diagnostics, Bazel inputs, and database telemetry. GitOrigin-RevId: d194310835f1df2a2a29c7827d77ca37eabbd0a3 --- codex-rs/cli/tests/debug_clear_memories.rs | 2 +- codex-rs/state/BUILD.bazel | 1 + codex-rs/state/src/lib.rs | 3 ++ codex-rs/state/src/migrations.rs | 7 ++++ codex-rs/state/src/runtime.rs | 29 ++++++++++++--- codex-rs/state/src/telemetry.rs | 2 + .../0001_thread_history.sql | 37 +++++++++++++++++++ 7 files changed, 74 insertions(+), 7 deletions(-) create mode 100644 codex-rs/state/thread_history_migrations/0001_thread_history.sql diff --git a/codex-rs/cli/tests/debug_clear_memories.rs b/codex-rs/cli/tests/debug_clear_memories.rs index e39d736ae056..be1c30da8a55 100644 --- a/codex-rs/cli/tests/debug_clear_memories.rs +++ b/codex-rs/cli/tests/debug_clear_memories.rs @@ -142,7 +142,7 @@ async fn debug_clear_memories_resets_memories_db_without_state_db() -> Result<() let codex_home = TempDir::new()?; let runtime = StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?; - drop(runtime); + runtime.close().await; let db_path = state_db_path(codex_home.path()); let memories_db_path = memories_db_path(codex_home.path()); diff --git a/codex-rs/state/BUILD.bazel b/codex-rs/state/BUILD.bazel index 8b7c454fdb01..ce4900a135f3 100644 --- a/codex-rs/state/BUILD.bazel +++ b/codex-rs/state/BUILD.bazel @@ -7,6 +7,7 @@ codex_rust_crate( "logs_migrations/**", "memory_migrations/**", "migrations/**", + "thread_history_migrations/**", ]), crate_name = "codex_state", ) diff --git a/codex-rs/state/src/lib.rs b/codex-rs/state/src/lib.rs index cdfcd29c5087..cdac70a4eb31 100644 --- a/codex-rs/state/src/lib.rs +++ b/codex-rs/state/src/lib.rs @@ -85,6 +85,8 @@ pub use runtime::sqlite_error_detail_is_lock; pub use runtime::sqlite_integrity_check; pub use runtime::state_db_filename; pub use runtime::state_db_path; +pub use runtime::thread_history_db_filename; +pub use runtime::thread_history_db_path; pub use telemetry::DbTelemetry; pub use telemetry::DbTelemetryHandle; pub use telemetry::install_process_db_telemetry; @@ -98,6 +100,7 @@ pub const LOGS_DB_FILENAME: &str = "logs_2.sqlite"; pub const GOALS_DB_FILENAME: &str = "goals_1.sqlite"; pub const MEMORIES_DB_FILENAME: &str = "memories_1.sqlite"; pub const STATE_DB_FILENAME: &str = "state_5.sqlite"; +pub const THREAD_HISTORY_DB_FILENAME: &str = "thread_history_1.sqlite"; /// Errors encountered during DB operations. Tags: [stage] pub const DB_ERROR_METRIC: &str = "codex.db.error"; diff --git a/codex-rs/state/src/migrations.rs b/codex-rs/state/src/migrations.rs index 187e1fb33d4c..f5851378dbc9 100644 --- a/codex-rs/state/src/migrations.rs +++ b/codex-rs/state/src/migrations.rs @@ -7,6 +7,7 @@ pub(crate) static STATE_MIGRATOR: Migrator = sqlx::migrate!("./migrations"); pub(crate) static LOGS_MIGRATOR: Migrator = sqlx::migrate!("./logs_migrations"); pub(crate) static GOALS_MIGRATOR: Migrator = sqlx::migrate!("./goals_migrations"); pub(crate) static MEMORIES_MIGRATOR: Migrator = sqlx::migrate!("./memory_migrations"); +pub(crate) static THREAD_HISTORY_MIGRATOR: Migrator = sqlx::migrate!("./thread_history_migrations"); /// Allow an older Codex binary to open a database that has already been /// migrated by a newer binary running in parallel. @@ -41,6 +42,12 @@ pub(crate) fn runtime_memories_migrator() -> Migrator { runtime_migrator(&MEMORIES_MIGRATOR) } +// The paginated history projector will call this when it takes ownership of opening the database. +#[allow(dead_code)] +pub(crate) fn runtime_thread_history_migrator() -> Migrator { + runtime_migrator(&THREAD_HISTORY_MIGRATOR) +} + pub(crate) async fn repair_legacy_recency_migration_version( pool: &SqlitePool, migrator: &Migrator, diff --git a/codex-rs/state/src/runtime.rs b/codex-rs/state/src/runtime.rs index 4b1d264d0b82..ad659ede5270 100644 --- a/codex-rs/state/src/runtime.rs +++ b/codex-rs/state/src/runtime.rs @@ -13,6 +13,7 @@ use crate::LogRow; use crate::MEMORIES_DB_FILENAME; use crate::STATE_DB_FILENAME; use crate::SortKey; +use crate::THREAD_HISTORY_DB_FILENAME; use crate::ThreadMetadata; use crate::ThreadMetadataBuilder; use crate::ThreadsPage; @@ -144,7 +145,16 @@ const MEMORIES_DB: RuntimeDbSpec = RuntimeDbSpec { migrate_phase: "migrate_memories", }; -const RUNTIME_DBS: [RuntimeDbSpec; 4] = [STATE_DB, LOGS_DB, GOALS_DB, MEMORIES_DB]; +const THREAD_HISTORY_DB: RuntimeDbSpec = RuntimeDbSpec { + label: "thread history DB", + filename: THREAD_HISTORY_DB_FILENAME, + kind: DbKind::ThreadHistory, + open_phase: "open_thread_history", + migrate_phase: "migrate_thread_history", +}; + +const RUNTIME_DBS: [RuntimeDbSpec; 5] = + [STATE_DB, LOGS_DB, GOALS_DB, MEMORIES_DB, THREAD_HISTORY_DB]; #[derive(Clone, Debug, Eq, PartialEq)] pub struct RuntimeDbPath { @@ -167,9 +177,9 @@ pub struct StateRuntime { impl StateRuntime { /// Initialize the state runtime using the provided Codex home and default provider. /// - /// This opens (and migrates) the SQLite databases under `codex_home`, - /// keeping logs in a dedicated file to reduce lock contention with the - /// rest of the state store. + /// This opens (and migrates) the SQLite databases under `codex_home`. + /// Logs and paginated thread history live in dedicated files to reduce + /// lock contention with the rest of the state store. pub async fn init(codex_home: PathBuf, default_provider: String) -> anyhow::Result> { Self::init_inner( codex_home, @@ -508,6 +518,14 @@ pub fn memories_db_path(codex_home: &Path) -> PathBuf { MEMORIES_DB.path(codex_home) } +pub fn thread_history_db_filename() -> String { + THREAD_HISTORY_DB.filename.to_string() +} + +pub fn thread_history_db_path(codex_home: &Path) -> PathBuf { + THREAD_HISTORY_DB.path(codex_home) +} + pub fn runtime_db_paths(codex_home: &Path) -> Vec { RUNTIME_DBS .iter() @@ -734,8 +752,7 @@ mod tests { .collect::>(); assert_eq!(phases, expected); - runtime.pool.close().await; - runtime.logs_pool.close().await; + runtime.close().await; let _ = tokio::fs::remove_dir_all(codex_home).await; } } diff --git a/codex-rs/state/src/telemetry.rs b/codex-rs/state/src/telemetry.rs index 1a9e4f995165..57a6c7131413 100644 --- a/codex-rs/state/src/telemetry.rs +++ b/codex-rs/state/src/telemetry.rs @@ -41,6 +41,7 @@ pub(crate) enum DbKind { Logs, Goals, Memories, + ThreadHistory, } impl DbKind { @@ -50,6 +51,7 @@ impl DbKind { Self::Logs => "logs", Self::Goals => "goals", Self::Memories => "memories", + Self::ThreadHistory => "thread_history", } } } diff --git a/codex-rs/state/thread_history_migrations/0001_thread_history.sql b/codex-rs/state/thread_history_migrations/0001_thread_history.sql new file mode 100644 index 000000000000..99ec54e052a8 --- /dev/null +++ b/codex-rs/state/thread_history_migrations/0001_thread_history.sql @@ -0,0 +1,37 @@ +CREATE TABLE thread_turns ( + thread_id TEXT NOT NULL, + turn_id TEXT NOT NULL, + rollout_ordinal INTEGER NOT NULL, + status TEXT NOT NULL, + error_json TEXT, + started_at INTEGER, + completed_at INTEGER, + duration_ms INTEGER, + first_user_item_id TEXT, + final_agent_item_id TEXT, + PRIMARY KEY (thread_id, turn_id) +); + +CREATE UNIQUE INDEX idx_thread_turns_page + ON thread_turns(thread_id, rollout_ordinal); + +CREATE TABLE thread_items ( + thread_id TEXT NOT NULL, + turn_id TEXT NOT NULL, + item_id TEXT NOT NULL, + rollout_ordinal INTEGER NOT NULL, + item_json TEXT NOT NULL, + PRIMARY KEY (thread_id, turn_id, item_id) +); + +CREATE UNIQUE INDEX idx_thread_items_page + ON thread_items(thread_id, rollout_ordinal); + +CREATE INDEX idx_thread_items_by_turn_page + ON thread_items(thread_id, turn_id, rollout_ordinal); + +CREATE TABLE thread_history_projection_state ( + thread_id TEXT PRIMARY KEY, + next_rollout_byte_offset INTEGER NOT NULL, + next_rollout_ordinal INTEGER NOT NULL +);