Add dedicated storage for paginated thread history

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());

codex-rs/state/BUILD.bazel

@@ -7,6 +7,7 @@ codex_rust_crate(
         "logs_migrations/**",
         "memory_migrations/**",
         "migrations/**",
+        "thread_history_migrations/**",
     ]),
     crate_name = "codex_state",
 )

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";

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,

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<Arc<Self>> {
         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<RuntimeDbPath> {
     RUNTIME_DBS
         .iter()
@@ -734,8 +752,7 @@ mod tests {
         .collect::<BTreeSet<_>>();
         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;
     }
 }

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",
         }
     }
 }

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
+);