Centralize SQLite connection configuration (#34808)

## What changed

- Add `SqliteConfig` to own the resolved SQLite home, runtime database paths,
  and shared read/write and read-only pool settings.
- Pass the configuration through state and local thread-store consumers,
  replacing standalone path helpers and duplicated connection setup.

GitOrigin-RevId: 7351b08da94b5c8b6c0bbe492f86aeeca0699d3c
This commit is contained in:
Adam Perry @ OpenAI authored and copyberry committed 2026-07-22 18:45:47 +00:00
1 parent a59a419afa
commit 946ed315a4
35 files changed
+445 -397

No files matched your search

+2
View File
@@ -3859,6 +3859,7 @@ dependencies = [
"codex-otel", "codex-otel",
"codex-protocol", "codex-protocol",
"codex-state", "codex-state",
"codex-utils-absolute-path",
"codex-utils-path", "codex-utils-path",
"pretty_assertions", "pretty_assertions",
"regex", "regex",
@@ -4087,6 +4088,7 @@ dependencies = [
"codex-protocol", "codex-protocol",
"codex-rollout", "codex-rollout",
"codex-state", "codex-state",
"codex-utils-absolute-path",
"codex-utils-path", "codex-utils-path",
"futures", "futures",
"pretty_assertions", "pretty_assertions",
+10 -2
View File
@@ -1196,8 +1196,16 @@ async fn init_sqlite_state_db_with_fresh_start_on_corruption(
} }
Err(err) => err, Err(err) => err,
}; };
let database_path = codex_state::runtime_db_path_for_corruption_error(&err) let database_path =
.unwrap_or_else(|| codex_state::state_db_path(config.sqlite_home.as_path())); codex_state::runtime_db_path_for_corruption_error(&err).unwrap_or_else(|| {
codex_state::SqliteConfig::from_sqlite_home(
codex_utils_absolute_path::AbsolutePathBuf::resolve_path_against_base(
config.sqlite_home.clone(),
&config.codex_home,
),
)
.state_db_path()
});
if !codex_state::is_sqlite_corruption_error(&err) if !codex_state::is_sqlite_corruption_error(&err)
&& !sqlite_home_is_blocking_file(database_path.as_path()) && !sqlite_home_is_blocking_file(database_path.as_path())
{ {
@@ -58,6 +58,7 @@ use codex_thread_store::CreateThreadParams as StoreCreateThreadParams;
use codex_thread_store::InMemoryThreadStore; use codex_thread_store::InMemoryThreadStore;
use codex_thread_store::ThreadPersistenceMetadata; use codex_thread_store::ThreadPersistenceMetadata;
use codex_thread_store::ThreadStore; use codex_thread_store::ThreadStore;
use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use tempfile::TempDir; use tempfile::TempDir;
use tokio::time::timeout; use tokio::time::timeout;
@@ -404,7 +405,9 @@ fn assert_no_local_persistence_artifacts(codex_home: &Path) -> Result<()> {
"non-local thread persistence should not create archived rollout sessions" "non-local thread persistence should not create archived rollout sessions"
); );
assert!( assert!(
!codex_state::state_db_path(codex_home).exists(), !codex_state::SqliteConfig::new_for_testing(codex_home.abs())
.state_db_path()
.exists(),
"non-local thread persistence should not create local thread sqlite" "non-local thread persistence should not create local thread sqlite"
); );
@@ -80,6 +80,7 @@ use codex_thread_store::ThreadMetadataPatch;
use codex_thread_store::ThreadPersistenceMetadata; use codex_thread_store::ThreadPersistenceMetadata;
use codex_thread_store::ThreadStore; use codex_thread_store::ThreadStore;
use codex_thread_store::UpdateThreadMetadataParams; use codex_thread_store::UpdateThreadMetadataParams;
use codex_utils_absolute_path::test_support::PathExt;
use core_test_support::responses; use core_test_support::responses;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use serde_json::Value; use serde_json::Value;
@@ -467,15 +468,14 @@ async fn thread_search_occurrences_reads_paginated_projection() -> Result<()> {
let codex_home = TempDir::new()?; let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let thread_id = codex_protocol::ThreadId::default(); let thread_id = codex_protocol::ThreadId::default();
let state_db = codex_state::StateRuntime::init( let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs());
codex_home.path().to_path_buf(), let state_db =
"mock_provider".to_string(), codex_state::StateRuntime::init(sqlite.home().to_path_buf(), "mock_provider".to_string())
) .await?;
.await?;
let store = LocalThreadStore::new( let store = LocalThreadStore::new(
LocalThreadStoreConfig { LocalThreadStoreConfig {
codex_home: codex_home.path().to_path_buf(), codex_home: codex_home.path().to_path_buf(),
sqlite_home: codex_home.path().to_path_buf(), sqlite,
default_model_provider_id: "mock_provider".to_string(), default_model_provider_id: "mock_provider".to_string(),
}, },
Some(state_db), Some(state_db),
@@ -1445,15 +1445,14 @@ async fn paginated_history_lists_use_projected_turns_and_items() -> Result<()> {
let codex_home = TempDir::new()?; let codex_home = TempDir::new()?;
MockResponsesConfig::new(&server.uri()).write(codex_home.path())?; MockResponsesConfig::new(&server.uri()).write(codex_home.path())?;
let thread_id = codex_protocol::ThreadId::default(); let thread_id = codex_protocol::ThreadId::default();
let state_db = codex_state::StateRuntime::init( let sqlite = codex_state::SqliteConfig::new_for_testing(codex_home.path().abs());
codex_home.path().to_path_buf(), let state_db =
"mock_provider".to_string(), codex_state::StateRuntime::init(sqlite.home().to_path_buf(), "mock_provider".to_string())
) .await?;
.await?;
let store = LocalThreadStore::new( let store = LocalThreadStore::new(
LocalThreadStoreConfig { LocalThreadStoreConfig {
codex_home: codex_home.path().to_path_buf(), codex_home: codex_home.path().to_path_buf(),
sqlite_home: codex_home.path().to_path_buf(), sqlite,
default_model_provider_id: "mock_provider".to_string(), default_model_provider_id: "mock_provider".to_string(),
}, },
Some(state_db), Some(state_db),
+7 -1
View File
@@ -2165,7 +2165,13 @@ async fn state_check(config: &Config) -> DoctorCheck {
path_readiness(&mut details, "log dir", &config.log_dir); path_readiness(&mut details, "log dir", &config.log_dir);
path_readiness(&mut details, "sqlite home", &config.sqlite_home); path_readiness(&mut details, "sqlite home", &config.sqlite_home);
let mut integrity_failures = Vec::new(); let mut integrity_failures = Vec::new();
for db in codex_state::runtime_db_paths(&config.sqlite_home) { let sqlite = codex_state::SqliteConfig::from_sqlite_home(
codex_utils_absolute_path::AbsolutePathBuf::resolve_path_against_base(
&config.sqlite_home,
&config.codex_home,
),
);
for db in sqlite.runtime_db_paths() {
path_readiness(&mut details, db.label, &db.path); path_readiness(&mut details, db.label, &db.path);
sqlite_integrity_detail(&mut details, &mut integrity_failures, db.label, &db.path).await; sqlite_integrity_detail(&mut details, &mut integrity_failures, db.label, &db.path).await;
} }
+25 -14
View File
@@ -91,9 +91,15 @@ impl RolloutScan {
} }
pub(super) async fn thread_inventory_check(config: &Config) -> DoctorCheck { pub(super) async fn thread_inventory_check(config: &Config) -> DoctorCheck {
let sqlite = codex_state::SqliteConfig::from_sqlite_home(
codex_utils_absolute_path::AbsolutePathBuf::resolve_path_against_base(
&config.sqlite_home,
&config.codex_home,
),
);
thread_inventory_check_for_roots( thread_inventory_check_for_roots(
config.codex_home.as_path(), config.codex_home.as_path(),
config.sqlite_home.as_path(), &sqlite,
config.model_provider_id.as_str(), config.model_provider_id.as_str(),
) )
.await .await
@@ -101,11 +107,11 @@ pub(super) async fn thread_inventory_check(config: &Config) -> DoctorCheck {
async fn thread_inventory_check_for_roots( async fn thread_inventory_check_for_roots(
codex_home: &Path, codex_home: &Path,
sqlite_home: &Path, sqlite: &codex_state::SqliteConfig,
default_provider: &str, default_provider: &str,
) -> DoctorCheck { ) -> DoctorCheck {
let scan = scan_rollout_files(codex_home).await; let scan = scan_rollout_files(codex_home).await;
let state_db_path = codex_state::state_db_path(sqlite_home); let state_db_path = sqlite.state_db_path();
let mut details = vec![ let mut details = vec![
format!("default model provider: {default_provider}"), format!("default model provider: {default_provider}"),
@@ -779,7 +785,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -826,7 +832,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -876,7 +882,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -902,7 +908,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -932,7 +938,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -959,7 +965,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -1013,7 +1019,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -1045,7 +1051,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -1067,7 +1073,7 @@ mod tests {
let check = thread_inventory_check_for_roots( let check = thread_inventory_check_for_roots(
fixture.codex_home.path(), fixture.codex_home.path(),
fixture.sqlite_home.path(), &fixture.sqlite(),
"test-provider", "test-provider",
) )
.await; .await;
@@ -1315,6 +1321,10 @@ mod tests {
} }
} }
fn sqlite(&self) -> codex_state::SqliteConfig {
codex_state::SqliteConfig::new_for_testing(self.sqlite_home.path().abs())
}
fn write_rollout(&self, archived: bool, timestamp: &str, thread_id: &str) -> PathBuf { fn write_rollout(&self, archived: bool, timestamp: &str, thread_id: &str) -> PathBuf {
let root = if archived { let root = if archived {
self.codex_home.path().join("archived_sessions") self.codex_home.path().join("archived_sessions")
@@ -1348,8 +1358,9 @@ mod tests {
} }
async fn insert_thread_row(&self, id: &str, rollout_path: &Path, archived: bool) { async fn insert_thread_row(&self, id: &str, rollout_path: &Path, archived: bool) {
let state_db_path = codex_state::state_db_path(self.sqlite_home.path()); let sqlite = self.sqlite();
let pool = codex_state::SqliteConfig::new_for_testing(self.sqlite_home.path().abs()) let state_db_path = sqlite.state_db_path();
let pool = sqlite
.open_read_write_pool(&state_db_path) .open_read_write_pool(&state_db_path)
.await .await
.expect("sqlite pool"); .expect("sqlite pool");
+4 -2
View File
@@ -26,8 +26,8 @@ use codex_execpolicy::ExecPolicyCheckCommand;
use codex_responses_api_proxy::Args as ResponsesApiProxyArgs; use codex_responses_api_proxy::Args as ResponsesApiProxyArgs;
use codex_rollout_trace::REDUCED_STATE_FILE_NAME; use codex_rollout_trace::REDUCED_STATE_FILE_NAME;
use codex_rollout_trace::replay_bundle; use codex_rollout_trace::replay_bundle;
use codex_state::SqliteConfig;
use codex_state::StateRuntime; use codex_state::StateRuntime;
use codex_state::memories_db_path;
use codex_tui::AppExitInfo; use codex_tui::AppExitInfo;
use codex_tui::Cli as TuiCli; use codex_tui::Cli as TuiCli;
use codex_tui::ExitReason; use codex_tui::ExitReason;
@@ -2052,7 +2052,9 @@ async fn run_debug_clear_memories_command(
.build() .build()
.await?; .await?;
let memories_path = memories_db_path(config.sqlite_home.as_path()); let sqlite_home =
AbsolutePathBuf::resolve_path_against_base(&config.sqlite_home, &config.codex_home);
let memories_path = SqliteConfig::from_sqlite_home(sqlite_home).memories_db_path();
let cleared_memories_db = let cleared_memories_db =
StateRuntime::clear_memory_data_in_sqlite_home(config.sqlite_home.as_path()).await?; StateRuntime::clear_memory_data_in_sqlite_home(config.sqlite_home.as_path()).await?;
+7 -6
View File
@@ -96,6 +96,7 @@ fn backup_folder(backups: &[RuntimeDbBackup]) -> Option<&Path> {
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use std::path::PathBuf; use std::path::PathBuf;
use tempfile::TempDir; use tempfile::TempDir;
@@ -103,8 +104,9 @@ mod tests {
#[tokio::test] #[tokio::test]
async fn backup_backs_up_only_failed_database_file() -> std::io::Result<()> { async fn backup_backs_up_only_failed_database_file() -> std::io::Result<()> {
let temp_dir = TempDir::new()?; let temp_dir = TempDir::new()?;
let state_path = codex_state::state_db_path(temp_dir.path()); let sqlite = codex_state::SqliteConfig::new_for_testing(temp_dir.path().abs());
let failed_db_path = codex_state::logs_db_path(temp_dir.path()); let state_path = sqlite.state_db_path();
let failed_db_path = sqlite.logs_db_path();
tokio::fs::write(state_path.as_path(), b"state").await?; tokio::fs::write(state_path.as_path(), b"state").await?;
tokio::fs::write(failed_db_path.as_path(), b"logs").await?; tokio::fs::write(failed_db_path.as_path(), b"logs").await?;
@@ -130,10 +132,9 @@ mod tests {
let temp_dir = TempDir::new()?; let temp_dir = TempDir::new()?;
let sqlite_home = temp_dir.path().join("sqlite-home"); let sqlite_home = temp_dir.path().join("sqlite-home");
tokio::fs::write(sqlite_home.as_path(), b"not-a-directory").await?; tokio::fs::write(sqlite_home.as_path(), b"not-a-directory").await?;
let startup_error = LocalStateDbStartupError::new( let sqlite = codex_state::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
codex_state::state_db_path(sqlite_home.as_path()), let startup_error =
"File exists".to_string(), LocalStateDbStartupError::new(sqlite.state_db_path(), "File exists".to_string());
);
assert!(is_auto_backup_recoverable(&startup_error)); assert!(is_auto_backup_recoverable(&startup_error));
let backups = backup_files_for_fresh_start(&startup_error).await?; let backups = backup_files_for_fresh_start(&startup_error).await?;
+4 -6
View File
@@ -2,8 +2,6 @@ use std::path::Path;
use anyhow::Result; use anyhow::Result;
use codex_state::StateRuntime; use codex_state::StateRuntime;
use codex_state::memories_db_path;
use codex_state::state_db_path;
use codex_utils_absolute_path::test_support::PathExt; use codex_utils_absolute_path::test_support::PathExt;
use predicates::str::contains; use predicates::str::contains;
use tempfile::TempDir; use tempfile::TempDir;
@@ -23,9 +21,9 @@ async fn debug_clear_memories_resets_state_and_removes_memory_dir() -> Result<()
drop(runtime); drop(runtime);
let thread_id = "00000000-0000-0000-0000-000000000123"; let thread_id = "00000000-0000-0000-0000-000000000123";
let db_path = state_db_path(codex_home.path()); let db_path = sqlite.state_db_path();
let pool = sqlite.open_read_write_pool(&db_path).await?; let pool = sqlite.open_read_write_pool(&db_path).await?;
let memories_db_path = memories_db_path(codex_home.path()); let memories_db_path = sqlite.memories_db_path();
let memories_pool = sqlite.open_read_write_pool(&memories_db_path).await?; let memories_pool = sqlite.open_read_write_pool(&memories_db_path).await?;
sqlx::query( sqlx::query(
@@ -145,8 +143,8 @@ async fn debug_clear_memories_resets_memories_db_without_state_db() -> Result<()
StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?; StateRuntime::init(codex_home.path().to_path_buf(), "test-provider".to_string()).await?;
runtime.close().await; runtime.close().await;
let db_path = state_db_path(codex_home.path()); let db_path = sqlite.state_db_path();
let memories_db_path = memories_db_path(codex_home.path()); let memories_db_path = sqlite.memories_db_path();
let memories_pool = sqlite.open_read_write_pool(&memories_db_path).await?; let memories_pool = sqlite.open_read_write_pool(&memories_db_path).await?;
sqlx::query( sqlx::query(
+3 -3
View File
@@ -437,7 +437,7 @@ impl ThreadManager {
set_thread_manager_test_mode_for_tests(/*enabled*/ true); set_thread_manager_test_mode_for_tests(/*enabled*/ true);
let auth_manager = AuthManager::from_auth_for_testing(auth); let auth_manager = AuthManager::from_auth_for_testing(auth);
let installation_id = uuid::Uuid::new_v4().to_string(); let installation_id = uuid::Uuid::new_v4().to_string();
let skills_codex_home = match AbsolutePathBuf::from_absolute_path_checked(&codex_home) { let absolute_codex_home = match AbsolutePathBuf::from_absolute_path_checked(&codex_home) {
Ok(codex_home) => codex_home, Ok(codex_home) => codex_home,
Err(err) => panic!("test codex_home should be absolute: {err}"), Err(err) => panic!("test codex_home should be absolute: {err}"),
}; };
@@ -450,7 +450,7 @@ impl ThreadManager {
)); ));
let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager))); let mcp_manager = Arc::new(McpManager::new(Arc::clone(&plugins_manager)));
let skills_service = Arc::new(SkillsService::new_with_restriction_product( let skills_service = Arc::new(SkillsService::new_with_restriction_product(
skills_codex_home, absolute_codex_home.clone(),
/*bundled_skills_enabled*/ true, /*bundled_skills_enabled*/ true,
restriction_product, restriction_product,
)); ));
@@ -459,7 +459,7 @@ impl ThreadManager {
let thread_store: Arc<dyn ThreadStore> = Arc::new(LocalThreadStore::new( let thread_store: Arc<dyn ThreadStore> = Arc::new(LocalThreadStore::new(
LocalThreadStoreConfig { LocalThreadStoreConfig {
codex_home: codex_home.clone(), codex_home: codex_home.clone(),
sqlite_home: codex_home.clone(), sqlite: codex_state::SqliteConfig::new_for_testing(absolute_codex_home),
default_model_provider_id: OPENAI_PROVIDER_ID.to_string(), default_model_provider_id: OPENAI_PROVIDER_ID.to_string(),
}, },
state_db.clone(), state_db.clone(),
+10 -3
View File
@@ -23,6 +23,7 @@ use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::UserMessageEvent; use codex_protocol::protocol::UserMessageEvent;
use codex_protocol::user_input::UserInput; use codex_protocol::user_input::UserInput;
use codex_web_search_extension::install as install_web_search_extension; use codex_web_search_extension::install as install_web_search_extension;
use core_test_support::PathExt;
use core_test_support::responses; use core_test_support::responses;
use core_test_support::responses::ev_completed; use core_test_support::responses::ev_completed;
use core_test_support::responses::ev_function_call; use core_test_support::responses::ev_function_call;
@@ -65,7 +66,9 @@ async fn new_thread_is_recorded_in_state_db() -> Result<()> {
let thread_id = test.session_configured.thread_id; let thread_id = test.session_configured.thread_id;
let rollout_path = test.codex.rollout_path().expect("rollout path"); let rollout_path = test.codex.rollout_path().expect("rollout path");
let db_path = codex_state::state_db_path(test.config.sqlite_home.as_path()); let db_path =
codex_state::SqliteConfig::new_for_testing(test.config.sqlite_home.as_path().abs())
.state_db_path();
for _ in 0..100 { for _ in 0..100 {
if tokio::fs::try_exists(&db_path).await.unwrap_or(false) { if tokio::fs::try_exists(&db_path).await.unwrap_or(false) {
@@ -419,7 +422,9 @@ async fn backfill_scans_existing_rollouts() -> Result<()> {
let test = builder.build(&server).await?; let test = builder.build(&server).await?;
let db_path = codex_state::state_db_path(test.config.sqlite_home.as_path()); let db_path =
codex_state::SqliteConfig::new_for_testing(test.config.sqlite_home.as_path().abs())
.state_db_path();
let rollout_path = test.config.codex_home.join(&rollout_rel_path); let rollout_path = test.config.codex_home.join(&rollout_rel_path);
let default_provider = test.config.model_provider_id.clone(); let default_provider = test.config.model_provider_id.clone();
@@ -470,7 +475,9 @@ async fn user_messages_persist_in_state_db() -> Result<()> {
}); });
let test = builder.build(&server).await?; let test = builder.build(&server).await?;
let db_path = codex_state::state_db_path(test.config.sqlite_home.as_path()); let db_path =
codex_state::SqliteConfig::new_for_testing(test.config.sqlite_home.as_path().abs())
.state_db_path();
for _ in 0..100 { for _ in 0..100 {
if tokio::fs::try_exists(&db_path).await.unwrap_or(false) { if tokio::fs::try_exists(&db_path).await.unwrap_or(false) {
break; break;
+1
View File
@@ -21,6 +21,7 @@ codex-git-utils = { workspace = true }
codex-otel = { workspace = true } codex-otel = { workspace = true }
codex-protocol = { workspace = true } codex-protocol = { workspace = true }
codex-state = { workspace = true } codex-state = { workspace = true }
codex-utils-absolute-path = { workspace = true }
codex-utils-path = { workspace = true } codex-utils-path = { workspace = true }
regex = { workspace = true } regex = { workspace = true }
serde = { workspace = true, features = ["derive"] } serde = { workspace = true, features = ["derive"] }
+3 -1
View File
@@ -14,6 +14,7 @@ use codex_protocol::protocol::SessionSource;
use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadHistoryMode;
pub use codex_state::LogEntry; pub use codex_state::LogEntry;
use codex_state::ThreadMetadataBuilder; use codex_state::ThreadMetadataBuilder;
use codex_utils_absolute_path::AbsolutePathBuf;
use codex_utils_path::normalize_for_path_comparison; use codex_utils_path::normalize_for_path_comparison;
use serde_json::Value; use serde_json::Value;
use std::path::Path; use std::path::Path;
@@ -216,7 +217,8 @@ fn emit_startup_warning(message: &str) {
/// Unlike [`init`], this helper does not run rollout backfill. It is for /// Unlike [`init`], this helper does not run rollout backfill. It is for
/// optional local reads from non-owning contexts such as remote app-server mode. /// optional local reads from non-owning contexts such as remote app-server mode.
pub async fn get_state_db(config: &impl RolloutConfigView) -> Option<StateDbHandle> { pub async fn get_state_db(config: &impl RolloutConfigView) -> Option<StateDbHandle> {
let state_path = codex_state::state_db_path(config.sqlite_home()); let sqlite_home = AbsolutePathBuf::try_from(config.sqlite_home()).ok()?;
let state_path = codex_state::SqliteConfig::from_sqlite_home(sqlite_home).state_db_path();
if !tokio::fs::try_exists(&state_path).await.unwrap_or(false) { if !tokio::fs::try_exists(&state_path).await.unwrap_or(false) {
codex_state::record_fallback( codex_state::record_fallback(
"get_state_db", "get_state_db",
+4 -1
View File
@@ -7,7 +7,9 @@ use clap::Parser;
use clap::ValueEnum; use clap::ValueEnum;
use codex_state::LogQuery; use codex_state::LogQuery;
use codex_state::LogRow; use codex_state::LogRow;
use codex_state::SqliteConfig;
use codex_state::StateRuntime; use codex_state::StateRuntime;
use codex_utils_absolute_path::AbsolutePathBuf;
use dirs::home_dir; use dirs::home_dir;
use owo_colors::OwoColorize; use owo_colors::OwoColorize;
@@ -136,7 +138,8 @@ fn resolve_db_path(args: &Args) -> anyhow::Result<PathBuf> {
} }
let codex_home = args.codex_home.clone().unwrap_or_else(default_codex_home); let codex_home = args.codex_home.clone().unwrap_or_else(default_codex_home);
Ok(codex_state::logs_db_path(codex_home.as_path())) let sqlite_home = AbsolutePathBuf::relative_to_current_dir(codex_home)?;
Ok(SqliteConfig::from_sqlite_home(sqlite_home).logs_db_path())
} }
fn default_codex_home() -> PathBuf { fn default_codex_home() -> PathBuf {
+1 -18
View File
@@ -63,26 +63,15 @@ pub use runtime::GoalUpdate;
pub use runtime::MemoryStore; pub use runtime::MemoryStore;
pub use runtime::RemoteControlEnrollmentRecord; pub use runtime::RemoteControlEnrollmentRecord;
pub use runtime::RuntimeDbBackup; pub use runtime::RuntimeDbBackup;
pub use runtime::RuntimeDbPath;
pub use runtime::ThreadFilterOptions; pub use runtime::ThreadFilterOptions;
pub use runtime::backup_runtime_db_for_fresh_start; pub use runtime::backup_runtime_db_for_fresh_start;
pub use runtime::goals_db_filename;
pub use runtime::goals_db_path;
pub use runtime::is_sqlite_corruption_error; pub use runtime::is_sqlite_corruption_error;
pub use runtime::logs_db_filename;
pub use runtime::logs_db_path;
pub use runtime::memories_db_filename;
pub use runtime::memories_db_path;
pub use runtime::open_thread_history_db; pub use runtime::open_thread_history_db;
pub use runtime::runtime_db_path_for_corruption_error; pub use runtime::runtime_db_path_for_corruption_error;
pub use runtime::runtime_db_paths;
pub use runtime::sqlite_error_detail_is_corruption; pub use runtime::sqlite_error_detail_is_corruption;
pub use runtime::sqlite_error_detail_is_lock; pub use runtime::sqlite_error_detail_is_lock;
pub use runtime::sqlite_integrity_check; pub use runtime::sqlite_integrity_check;
pub use runtime::state_db_filename; pub use sqlite::RuntimeDbPath;
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::DbTelemetry;
pub use telemetry::DbTelemetryHandle; pub use telemetry::DbTelemetryHandle;
pub use telemetry::install_process_db_telemetry; pub use telemetry::install_process_db_telemetry;
@@ -92,12 +81,6 @@ pub use telemetry::record_fallback;
/// Environment variable for overriding the SQLite state database home directory. /// Environment variable for overriding the SQLite state database home directory.
pub const SQLITE_HOME_ENV: &str = "CODEX_SQLITE_HOME"; pub const SQLITE_HOME_ENV: &str = "CODEX_SQLITE_HOME";
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] /// Errors encountered during DB operations. Tags: [stage]
pub const DB_ERROR_METRIC: &str = "codex.db.error"; pub const DB_ERROR_METRIC: &str = "codex.db.error";
/// Metrics on backfill process. Tags: [status] /// Metrics on backfill process. Tags: [status]
+7 -5
View File
@@ -7,7 +7,6 @@ use std::borrow::Cow;
use super::STATE_MIGRATOR; use super::STATE_MIGRATOR;
use super::repair_legacy_recency_migration_version; use super::repair_legacy_recency_migration_version;
use crate::state_db_path;
fn migrator_through(version: i64) -> Migrator { fn migrator_through(version: i64) -> Migrator {
Migrator { Migrator {
@@ -37,8 +36,9 @@ async fn agent_job_tables_are_dropped_when_upgrading() {
let _ = std::fs::remove_dir_all(sqlite_home); let _ = std::fs::remove_dir_all(sqlite_home);
}); });
let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs()); let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
let state_path = sqlite.state_db_path();
let pool = sqlite let pool = sqlite
.open_read_write_pool(&state_db_path(&sqlite_home)) .open_read_write_pool(&state_path)
.await .await
.expect("sqlite database should open"); .expect("sqlite database should open");
migrator_through(/*version*/ 15) migrator_through(/*version*/ 15)
@@ -130,8 +130,9 @@ async fn recency_migration_backfills_and_seeds_old_binary_inserts() {
let _ = std::fs::remove_dir_all(sqlite_home); let _ = std::fs::remove_dir_all(sqlite_home);
}); });
let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs()); let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
let state_path = sqlite.state_db_path();
let pool = sqlite let pool = sqlite
.open_read_write_pool(&state_db_path(&sqlite_home)) .open_read_write_pool(&state_path)
.await .await
.expect("sqlite database should open"); .expect("sqlite database should open");
migrator_through(/*version*/ 37) migrator_through(/*version*/ 37)
@@ -243,8 +244,9 @@ async fn repairs_recency_migration_that_was_applied_as_version_38() {
let _ = std::fs::remove_dir_all(sqlite_home); let _ = std::fs::remove_dir_all(sqlite_home);
}); });
let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs()); let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
let state_path = sqlite.state_db_path();
let pool = sqlite let pool = sqlite
.open_read_write_pool(&state_db_path(&sqlite_home)) .open_read_write_pool(&state_path)
.await .await
.expect("sqlite database should open"); .expect("sqlite database should open");
migrator_through(/*version*/ 37) migrator_through(/*version*/ 37)
@@ -319,7 +321,7 @@ async fn repair_recency_migration_succeeds_while_another_connection_holds_writer
let _ = std::fs::remove_dir_all(sqlite_home); let _ = std::fs::remove_dir_all(sqlite_home);
}); });
let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs()); let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
let state_path = state_db_path(&sqlite_home); let state_path = sqlite.state_db_path();
let pool = sqlite let pool = sqlite
.open_read_write_pool(&state_path) .open_read_write_pool(&state_path)
.await .await
+41 -266
View File
@@ -1,18 +1,12 @@
use crate::GOALS_DB_FILENAME;
use crate::LOGS_DB_FILENAME;
use crate::LogEntry; use crate::LogEntry;
use crate::LogQuery; use crate::LogQuery;
use crate::LogRow; use crate::LogRow;
use crate::MEMORIES_DB_FILENAME;
use crate::STATE_DB_FILENAME;
use crate::SortKey; use crate::SortKey;
use crate::SqliteConfig; use crate::SqliteConfig;
use crate::THREAD_HISTORY_DB_FILENAME;
use crate::ThreadMetadata; use crate::ThreadMetadata;
use crate::ThreadMetadataBuilder; use crate::ThreadMetadataBuilder;
use crate::ThreadsPage; use crate::ThreadsPage;
use crate::apply_rollout_item; use crate::apply_rollout_item;
use crate::migrations::repair_legacy_recency_migration_version;
use crate::migrations::runtime_goals_migrator; use crate::migrations::runtime_goals_migrator;
use crate::migrations::runtime_logs_migrator; use crate::migrations::runtime_logs_migrator;
use crate::migrations::runtime_memories_migrator; use crate::migrations::runtime_memories_migrator;
@@ -37,7 +31,6 @@ use sqlx::Row;
use sqlx::Sqlite; use sqlx::Sqlite;
use sqlx::SqliteConnection; use sqlx::SqliteConnection;
use sqlx::SqlitePool; use sqlx::SqlitePool;
use sqlx::migrate::Migrator;
use std::collections::BTreeSet; use std::collections::BTreeSet;
use std::path::Path; use std::path::Path;
use std::path::PathBuf; use std::path::PathBuf;
@@ -67,6 +60,7 @@ pub use goals::GoalStore;
pub use goals::GoalUpdate; pub use goals::GoalUpdate;
pub use memories::MemoryStore; pub use memories::MemoryStore;
pub use recovery::RuntimeDbBackup; pub use recovery::RuntimeDbBackup;
pub(super) use recovery::RuntimeDbInitError;
pub use recovery::backup_runtime_db_for_fresh_start; pub use recovery::backup_runtime_db_for_fresh_start;
pub use recovery::is_sqlite_corruption_error; pub use recovery::is_sqlite_corruption_error;
pub use recovery::runtime_db_path_for_corruption_error; pub use recovery::runtime_db_path_for_corruption_error;
@@ -84,70 +78,6 @@ pub use threads::ThreadFilterOptions;
const LOG_PARTITION_SIZE_LIMIT_BYTES: i64 = 10 * 1024 * 1024; const LOG_PARTITION_SIZE_LIMIT_BYTES: i64 = 10 * 1024 * 1024;
const LOG_PARTITION_ROW_LIMIT: i64 = 1_000; const LOG_PARTITION_ROW_LIMIT: i64 = 1_000;
#[derive(Clone, Copy)]
struct RuntimeDbSpec {
label: &'static str,
filename: &'static str,
kind: DbKind,
open_phase: &'static str,
migrate_phase: &'static str,
}
impl RuntimeDbSpec {
fn path(self, codex_home: &Path) -> PathBuf {
codex_home.join(self.filename)
}
}
const STATE_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "state DB",
filename: STATE_DB_FILENAME,
kind: DbKind::State,
open_phase: "open_state",
migrate_phase: "migrate_state",
};
const LOGS_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "log DB",
filename: LOGS_DB_FILENAME,
kind: DbKind::Logs,
open_phase: "open_logs",
migrate_phase: "migrate_logs",
};
const GOALS_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "goals DB",
filename: GOALS_DB_FILENAME,
kind: DbKind::Goals,
open_phase: "open_goals",
migrate_phase: "migrate_goals",
};
const MEMORIES_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "memories DB",
filename: MEMORIES_DB_FILENAME,
kind: DbKind::Memories,
open_phase: "open_memories",
migrate_phase: "migrate_memories",
};
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 {
pub label: &'static str,
pub path: PathBuf,
}
#[derive(Clone)] #[derive(Clone)]
pub struct StateRuntime { pub struct StateRuntime {
codex_home: PathBuf, codex_home: PathBuf,
@@ -195,17 +125,13 @@ impl StateRuntime {
let logs_migrator = runtime_logs_migrator(); let logs_migrator = runtime_logs_migrator();
let goals_migrator = runtime_goals_migrator(); let goals_migrator = runtime_goals_migrator();
let memories_migrator = runtime_memories_migrator(); let memories_migrator = runtime_memories_migrator();
let state_path = STATE_DB.path(codex_home.as_path()); let state_path = sqlite.state_db_path();
let logs_path = LOGS_DB.path(codex_home.as_path()); let logs_path = sqlite.logs_db_path();
let goals_path = GOALS_DB.path(codex_home.as_path()); let goals_path = sqlite.goals_db_path();
let memories_path = MEMORIES_DB.path(codex_home.as_path()); let memories_path = sqlite.memories_db_path();
let pool = match open_state_sqlite( let pool = match sqlite
&sqlite, .open_state_db(&state_migrator, telemetry_override)
&state_path, .await
&state_migrator,
telemetry_override,
)
.await
{ {
Ok(db) => Arc::new(db), Ok(db) => Arc::new(db),
Err(err) => { Err(err) => {
@@ -213,22 +139,20 @@ impl StateRuntime {
return Err(err); return Err(err);
} }
}; };
let logs_pool = let logs_pool = match sqlite
match open_logs_sqlite(&sqlite, &logs_path, &logs_migrator, telemetry_override).await { .open_logs_db(&logs_migrator, telemetry_override)
Ok(db) => Arc::new(db), .await
Err(err) => { {
warn!("failed to open logs db at {}: {err}", logs_path.display()); Ok(db) => Arc::new(db),
close_sqlite_pools(&[pool.as_ref()]).await; Err(err) => {
return Err(err); warn!("failed to open logs db at {}: {err}", logs_path.display());
} close_sqlite_pools(&[pool.as_ref()]).await;
}; return Err(err);
let goals_pool = match open_goals_sqlite( }
&sqlite, };
&goals_path, let goals_pool = match sqlite
&goals_migrator, .open_goals_db(&goals_migrator, telemetry_override)
telemetry_override, .await
)
.await
{ {
Ok(db) => Arc::new(db), Ok(db) => Arc::new(db),
Err(err) => { Err(err) => {
@@ -237,13 +161,9 @@ impl StateRuntime {
return Err(err); return Err(err);
} }
}; };
let memories_pool = match open_memories_sqlite( let memories_pool = match sqlite
&sqlite, .open_memories_db(&memories_migrator, telemetry_override)
&memories_path, .await
&memories_migrator,
telemetry_override,
)
.await
{ {
Ok(db) => Arc::new(db), Ok(db) => Arc::new(db),
Err(err) => { Err(err) => {
@@ -347,19 +267,15 @@ impl StateRuntime {
pub async fn clear_memory_data_in_sqlite_home(sqlite_home: &Path) -> anyhow::Result<bool> { pub async fn clear_memory_data_in_sqlite_home(sqlite_home: &Path) -> anyhow::Result<bool> {
let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(sqlite_home)?); let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(sqlite_home)?);
let memories_path = MEMORIES_DB.path(sqlite_home); let memories_path = sqlite.memories_db_path();
if !tokio::fs::try_exists(&memories_path).await? { if !tokio::fs::try_exists(&memories_path).await? {
return Ok(false); return Ok(false);
} }
let memories_migrator = runtime_memories_migrator(); let memories_migrator = runtime_memories_migrator();
let pool = open_memories_sqlite( let pool = sqlite
&sqlite, .open_memories_db(&memories_migrator, /*telemetry_override*/ None)
&memories_path, .await?;
&memories_migrator,
/*telemetry_override*/ None,
)
.await?;
memories::clear_memory_data_in_pool(&pool).await?; memories::clear_memory_data_in_pool(&pool).await?;
pool.close().await; pool.close().await;
Ok(true) Ok(true)
@@ -372,100 +288,13 @@ async fn close_sqlite_pools(pools: &[&SqlitePool]) {
} }
} }
async fn open_state_sqlite(
sqlite: &SqliteConfig,
path: &Path,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
// New state DBs should use incremental auto-vacuum, but retrofitting an
// existing DB requires a full VACUUM. Do not attempt that during process
// startup: it is maintenance work that can contend with foreground writers.
open_sqlite(sqlite, path, migrator, STATE_DB, telemetry_override).await
}
async fn open_logs_sqlite(
sqlite: &SqliteConfig,
path: &Path,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
open_sqlite(sqlite, path, migrator, LOGS_DB, telemetry_override).await
}
async fn open_goals_sqlite(
sqlite: &SqliteConfig,
path: &Path,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
open_sqlite(sqlite, path, migrator, GOALS_DB, telemetry_override).await
}
async fn open_memories_sqlite(
sqlite: &SqliteConfig,
path: &Path,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
open_sqlite(sqlite, path, migrator, MEMORIES_DB, telemetry_override).await
}
/// Open and migrate the rebuildable paginated thread-history database. /// Open and migrate the rebuildable paginated thread-history database.
pub async fn open_thread_history_db(sqlite_home: &Path) -> anyhow::Result<SqlitePool> { pub async fn open_thread_history_db(sqlite_home: &Path) -> anyhow::Result<SqlitePool> {
let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(sqlite_home)?); let sqlite = SqliteConfig::from_sqlite_home(AbsolutePathBuf::try_from(sqlite_home)?);
let migrator = runtime_thread_history_migrator(); let migrator = runtime_thread_history_migrator();
open_sqlite( sqlite
&sqlite, .open_thread_history_db(&migrator, /*telemetry_override*/ None)
thread_history_db_path(sqlite_home).as_path(),
&migrator,
THREAD_HISTORY_DB,
/*telemetry_override*/ None,
)
.await
}
async fn open_sqlite(
sqlite: &SqliteConfig,
path: &Path,
migrator: &Migrator,
spec: RuntimeDbSpec,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
let started = Instant::now();
let pool_result = sqlite
.open_read_write_pool(path)
.await .await
.map_err(anyhow::Error::from);
crate::telemetry::record_init_result(
telemetry_override,
spec.kind,
spec.open_phase,
started.elapsed(),
&pool_result,
);
let pool = pool_result
.map_err(|source| recovery::RuntimeDbInitError::new(spec.label, "open", path, source))?;
let started = Instant::now();
let migrate_result = async {
if matches!(spec.kind, DbKind::State) {
repair_legacy_recency_migration_version(&pool, migrator).await?;
}
migrator.run(&pool).await.map_err(anyhow::Error::from)
}
.await;
crate::telemetry::record_init_result(
telemetry_override,
spec.kind,
spec.migrate_phase,
started.elapsed(),
&migrate_result,
);
if let Err(source) = migrate_result {
pool.close().await;
return Err(recovery::RuntimeDbInitError::new(spec.label, "migrate", path, source).into());
}
Ok(pool)
} }
pub(super) async fn ensure_backfill_state_row_in_pool( pub(super) async fn ensure_backfill_state_row_in_pool(
@@ -496,56 +325,6 @@ ON CONFLICT(id) DO NOTHING
Ok(()) Ok(())
} }
pub fn state_db_filename() -> String {
STATE_DB.filename.to_string()
}
pub fn state_db_path(codex_home: &Path) -> PathBuf {
STATE_DB.path(codex_home)
}
pub fn logs_db_filename() -> String {
LOGS_DB.filename.to_string()
}
pub fn logs_db_path(codex_home: &Path) -> PathBuf {
LOGS_DB.path(codex_home)
}
pub fn goals_db_filename() -> String {
GOALS_DB.filename.to_string()
}
pub fn goals_db_path(codex_home: &Path) -> PathBuf {
GOALS_DB.path(codex_home)
}
pub fn memories_db_filename() -> String {
MEMORIES_DB.filename.to_string()
}
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()
.map(|spec| RuntimeDbPath {
label: spec.label,
path: spec.path(codex_home),
})
.collect()
}
/// Run SQLite's built-in integrity check against an existing database file. /// Run SQLite's built-in integrity check against an existing database file.
pub async fn sqlite_integrity_check(path: &Path) -> anyhow::Result<Vec<String>> { pub async fn sqlite_integrity_check(path: &Path) -> anyhow::Result<Vec<String>> {
let sqlite = let sqlite =
@@ -561,10 +340,8 @@ pub async fn sqlite_integrity_check(path: &Path) -> anyhow::Result<Vec<String>>
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::StateRuntime; use super::StateRuntime;
use super::open_state_sqlite;
use super::runtime_state_migrator; use super::runtime_state_migrator;
use super::sqlite_integrity_check; use super::sqlite_integrity_check;
use super::state_db_path;
use super::test_support::unique_temp_dir; use super::test_support::unique_temp_dir;
use crate::DB_INIT_METRIC; use crate::DB_INIT_METRIC;
use crate::DbTelemetry; use crate::DbTelemetry;
@@ -642,8 +419,9 @@ mod tests {
tokio::fs::create_dir_all(&codex_home) tokio::fs::create_dir_all(&codex_home)
.await .await
.expect("create codex home"); .expect("create codex home");
let path = state_db_path(codex_home.as_path()); let sqlite = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs());
let pool = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()) let path = sqlite.state_db_path();
let pool = sqlite
.open_read_write_pool(&path) .open_read_write_pool(&path)
.await .await
.expect("open sqlite db"); .expect("open sqlite db");
@@ -667,8 +445,9 @@ mod tests {
tokio::fs::create_dir_all(&codex_home) tokio::fs::create_dir_all(&codex_home)
.await .await
.expect("create codex home"); .expect("create codex home");
let state_path = state_db_path(codex_home.as_path()); let sqlite = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs());
let pool = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()) let state_path = sqlite.state_db_path();
let pool = sqlite
.open_read_write_pool(&state_path) .open_read_write_pool(&state_path)
.await .await
.expect("open state db"); .expect("open state db");
@@ -698,14 +477,10 @@ mod tests {
strict_pool.close().await; strict_pool.close().await;
let tolerant_migrator = runtime_state_migrator(); let tolerant_migrator = runtime_state_migrator();
let tolerant_pool = open_state_sqlite( let tolerant_pool = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs())
&crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()), .open_state_db(&tolerant_migrator, /*telemetry_override*/ None)
state_path.as_path(), .await
&tolerant_migrator, .expect("runtime migrator should tolerate newer applied migrations");
/*telemetry_override*/ None,
)
.await
.expect("runtime migrator should tolerate newer applied migrations");
tolerant_pool.close().await; tolerant_pool.close().await;
let _ = tokio::fs::remove_dir_all(codex_home).await; let _ = tokio::fs::remove_dir_all(codex_home).await;
+3 -2
View File
@@ -175,8 +175,9 @@ mod tests {
let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string())
.await .await
.expect("initialize runtime"); .expect("initialize runtime");
let write_pool = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()) let sqlite = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs());
.open_read_write_pool(&crate::state_db_path(codex_home.as_path())) let write_pool = sqlite
.open_read_write_pool(&sqlite.state_db_path())
.await .await
.expect("open write pool"); .expect("open write pool");
let mut write_connection = write_pool.acquire().await.expect("open write connection"); let mut write_connection = write_pool.acquire().await.expect("open write connection");
+8 -4
View File
@@ -545,7 +545,6 @@ mod tests {
use super::test_support::unique_temp_dir; use super::test_support::unique_temp_dir;
use crate::LogEntry; use crate::LogEntry;
use crate::LogQuery; use crate::LogQuery;
use crate::logs_db_path;
use crate::migrations::LOGS_MIGRATOR; use crate::migrations::LOGS_MIGRATOR;
use chrono::Utc; use chrono::Utc;
use codex_utils_absolute_path::test_support::PathExt; use codex_utils_absolute_path::test_support::PathExt;
@@ -596,7 +595,9 @@ mod tests {
.await .await
.expect("insert test logs"); .expect("insert test logs");
let logs_count = log_row_count(logs_db_path(codex_home.as_path()).as_path()).await; let logs_path =
crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()).logs_db_path();
let logs_count = log_row_count(logs_path.as_path()).await;
assert_eq!(logs_count, 1); assert_eq!(logs_count, 1);
@@ -609,7 +610,8 @@ mod tests {
tokio::fs::create_dir_all(&codex_home) tokio::fs::create_dir_all(&codex_home)
.await .await
.expect("create codex home"); .expect("create codex home");
let logs_path = logs_db_path(codex_home.as_path()); let logs_path =
crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()).logs_db_path();
let old_logs_migrator = Migrator { let old_logs_migrator = Migrator {
migrations: Cow::Owned(vec![LOGS_MIGRATOR.migrations[0].clone()]), migrations: Cow::Owned(vec![LOGS_MIGRATOR.migrations[0].clone()]),
ignore_missing: false, ignore_missing: false,
@@ -706,7 +708,9 @@ mod tests {
.await .await
.expect("initialize runtime"); .expect("initialize runtime");
let pool = open_db_pool(logs_db_path(codex_home.as_path()).as_path()).await; let logs_path =
crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()).logs_db_path();
let pool = open_db_pool(logs_path.as_path()).await;
let auto_vacuum = sqlx::query_scalar::<_, i64>("PRAGMA auto_vacuum") let auto_vacuum = sqlx::query_scalar::<_, i64>("PRAGMA auto_vacuum")
.fetch_one(&pool) .fetch_one(&pool)
.await .await
+8 -4
View File
@@ -1,12 +1,14 @@
use super::*; use super::*;
use crate::runtime::test_support::unique_temp_dir; use crate::runtime::test_support::unique_temp_dir;
use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
#[tokio::test] #[tokio::test]
async fn backup_moves_only_requested_runtime_db_files_to_backup_folder() -> std::io::Result<()> { async fn backup_moves_only_requested_runtime_db_files_to_backup_folder() -> std::io::Result<()> {
let sqlite_home = unique_temp_dir(); let sqlite_home = unique_temp_dir();
tokio::fs::create_dir_all(sqlite_home.as_path()).await?; tokio::fs::create_dir_all(sqlite_home.as_path()).await?;
let runtime_paths = super::super::runtime_db_paths(sqlite_home.as_path()); let sqlite = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
let runtime_paths = sqlite.runtime_db_paths();
let mut expected_paths = Vec::new(); let mut expected_paths = Vec::new();
for db_path in runtime_paths.iter().map(|db| db.path.as_path()) { for db_path in runtime_paths.iter().map(|db| db.path.as_path()) {
for path in sqlite_paths(db_path) { for path in sqlite_paths(db_path) {
@@ -14,7 +16,7 @@ async fn backup_moves_only_requested_runtime_db_files_to_backup_folder() -> std:
expected_paths.push(path); expected_paths.push(path);
} }
} }
let failed_db_path = super::super::logs_db_path(sqlite_home.as_path()); let failed_db_path = sqlite.logs_db_path();
let failed_paths = sqlite_paths(failed_db_path.as_path()); let failed_paths = sqlite_paths(failed_db_path.as_path());
let backups = backup_runtime_db_for_fresh_start(failed_db_path.as_path()).await?; let backups = backup_runtime_db_for_fresh_start(failed_db_path.as_path()).await?;
@@ -48,7 +50,9 @@ async fn backup_replaces_blocking_sqlite_home_file() -> std::io::Result<()> {
tokio::fs::write(sqlite_home.as_path(), b"not-a-directory").await?; tokio::fs::write(sqlite_home.as_path(), b"not-a-directory").await?;
let backups = backup_runtime_db_for_fresh_start( let backups = backup_runtime_db_for_fresh_start(
super::super::state_db_path(sqlite_home.as_path()).as_path(), crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs())
.state_db_path()
.as_path(),
) )
.await?; .await?;
@@ -79,7 +83,7 @@ async fn runtime_db_path_for_corruption_error_returns_failed_database_path() ->
{ {
let sqlite_home = unique_temp_dir(); let sqlite_home = unique_temp_dir();
tokio::fs::create_dir_all(sqlite_home.as_path()).await?; tokio::fs::create_dir_all(sqlite_home.as_path()).await?;
let path = super::super::state_db_path(sqlite_home.as_path()); let path = crate::SqliteConfig::new_for_testing(sqlite_home.as_path().abs()).state_db_path();
tokio::fs::write(path.as_path(), b"not sqlite").await?; tokio::fs::write(path.as_path(), b"not sqlite").await?;
let err = match super::super::StateRuntime::init(sqlite_home, "openai".to_string()).await { let err = match super::super::StateRuntime::init(sqlite_home, "openai".to_string()).await {
+3 -3
View File
@@ -157,7 +157,6 @@ mod tests {
use super::StateRuntime; use super::StateRuntime;
use super::test_support::unique_temp_dir; use super::test_support::unique_temp_dir;
use crate::migrations::STATE_MIGRATOR; use crate::migrations::STATE_MIGRATOR;
use crate::state_db_path;
use codex_utils_absolute_path::test_support::PathExt; use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use sqlx::migrate::Migrator; use sqlx::migrate::Migrator;
@@ -344,8 +343,9 @@ mod tests {
table_name: STATE_MIGRATOR.table_name.clone(), table_name: STATE_MIGRATOR.table_name.clone(),
create_schemas: STATE_MIGRATOR.create_schemas.clone(), create_schemas: STATE_MIGRATOR.create_schemas.clone(),
}; };
let pool = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs()) let sqlite = crate::SqliteConfig::new_for_testing(codex_home.as_path().abs());
.open_read_write_pool(&state_db_path(codex_home.as_path())) let pool = sqlite
.open_read_write_pool(&sqlite.state_db_path())
.await .await
.expect("open old state db"); .expect("open old state db");
old_state_migrator old_state_migrator
+208
View File
@@ -1,17 +1,95 @@
//! Shared SQLite connection configuration. //! Shared SQLite connection configuration.
use crate::DbTelemetry;
use crate::migrations::repair_legacy_recency_migration_version;
use crate::runtime::RuntimeDbInitError;
use crate::telemetry;
use crate::telemetry::DbKind;
use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_absolute_path::AbsolutePathBuf;
use log::LevelFilter; use log::LevelFilter;
use sqlx::ConnectOptions; use sqlx::ConnectOptions;
use sqlx::Error; use sqlx::Error;
use sqlx::SqlitePool; use sqlx::SqlitePool;
use sqlx::migrate::Migrator;
use sqlx::sqlite::SqliteAutoVacuum; use sqlx::sqlite::SqliteAutoVacuum;
use sqlx::sqlite::SqliteConnectOptions; use sqlx::sqlite::SqliteConnectOptions;
use sqlx::sqlite::SqliteJournalMode; use sqlx::sqlite::SqliteJournalMode;
use sqlx::sqlite::SqlitePoolOptions; use sqlx::sqlite::SqlitePoolOptions;
use sqlx::sqlite::SqliteSynchronous; use sqlx::sqlite::SqliteSynchronous;
use std::path::Path; use std::path::Path;
use std::path::PathBuf;
use std::time::Duration; use std::time::Duration;
use std::time::Instant;
const LOGS_DB_FILENAME: &str = "logs_2.sqlite";
const GOALS_DB_FILENAME: &str = "goals_1.sqlite";
const MEMORIES_DB_FILENAME: &str = "memories_1.sqlite";
const STATE_DB_FILENAME: &str = "state_5.sqlite";
const THREAD_HISTORY_DB_FILENAME: &str = "thread_history_1.sqlite";
#[derive(Clone, Copy)]
struct RuntimeDbSpec {
label: &'static str,
filename: &'static str,
kind: DbKind,
open_phase: &'static str,
migrate_phase: &'static str,
}
impl RuntimeDbSpec {
fn path(self, codex_home: &Path) -> PathBuf {
codex_home.join(self.filename)
}
}
const STATE_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "state DB",
filename: STATE_DB_FILENAME,
kind: DbKind::State,
open_phase: "open_state",
migrate_phase: "migrate_state",
};
const LOGS_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "log DB",
filename: LOGS_DB_FILENAME,
kind: DbKind::Logs,
open_phase: "open_logs",
migrate_phase: "migrate_logs",
};
const GOALS_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "goals DB",
filename: GOALS_DB_FILENAME,
kind: DbKind::Goals,
open_phase: "open_goals",
migrate_phase: "migrate_goals",
};
const MEMORIES_DB: RuntimeDbSpec = RuntimeDbSpec {
label: "memories DB",
filename: MEMORIES_DB_FILENAME,
kind: DbKind::Memories,
open_phase: "open_memories",
migrate_phase: "migrate_memories",
};
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 {
pub label: &'static str,
pub path: PathBuf,
}
/// Resolved configuration shared by all Codex SQLite connections. /// Resolved configuration shared by all Codex SQLite connections.
#[derive(Clone, Debug, Eq, PartialEq)] #[derive(Clone, Debug, Eq, PartialEq)]
@@ -32,6 +110,136 @@ impl SqliteConfig {
self.sqlite_home.as_path() self.sqlite_home.as_path()
} }
/// Return the path to the primary state database.
pub fn state_db_path(&self) -> PathBuf {
STATE_DB.path(self.home())
}
/// Return the path to the logs database.
pub fn logs_db_path(&self) -> PathBuf {
LOGS_DB.path(self.home())
}
/// Return the path to the goals database.
pub fn goals_db_path(&self) -> PathBuf {
GOALS_DB.path(self.home())
}
/// Return the path to the memories database.
pub fn memories_db_path(&self) -> PathBuf {
MEMORIES_DB.path(self.home())
}
/// Return the path to the paginated thread-history database.
pub fn thread_history_db_path(&self) -> PathBuf {
THREAD_HISTORY_DB.path(self.home())
}
/// Return the paths to every database managed by the state runtime.
pub fn runtime_db_paths(&self) -> Vec<RuntimeDbPath> {
RUNTIME_DBS
.iter()
.map(|spec| RuntimeDbPath {
label: spec.label,
path: spec.path(self.home()),
})
.collect()
}
pub(super) async fn open_state_db(
&self,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
// New state DBs should use incremental auto-vacuum, but retrofitting an
// existing DB requires a full VACUUM. Do not attempt that during process
// startup: it is maintenance work that can contend with foreground writers.
self.open_runtime_db(STATE_DB, migrator, telemetry_override)
.await
}
pub(super) async fn open_logs_db(
&self,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
self.open_runtime_db(LOGS_DB, migrator, telemetry_override)
.await
}
pub(super) async fn open_goals_db(
&self,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
self.open_runtime_db(GOALS_DB, migrator, telemetry_override)
.await
}
pub(super) async fn open_memories_db(
&self,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
self.open_runtime_db(MEMORIES_DB, migrator, telemetry_override)
.await
}
pub(super) async fn open_thread_history_db(
&self,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
self.open_runtime_db(THREAD_HISTORY_DB, migrator, telemetry_override)
.await
}
async fn open_runtime_db(
&self,
spec: RuntimeDbSpec,
migrator: &Migrator,
telemetry_override: Option<&dyn DbTelemetry>,
) -> anyhow::Result<SqlitePool> {
let path = spec.path(self.home());
let started = Instant::now();
let pool_result = self
.open_read_write_pool(&path)
.await
.map_err(anyhow::Error::from);
telemetry::record_init_result(
telemetry_override,
spec.kind,
spec.open_phase,
started.elapsed(),
&pool_result,
);
let pool = pool_result.map_err(|source| {
RuntimeDbInitError::new(spec.label, "open", path.as_path(), source)
})?;
let started = Instant::now();
let migrate_result = async {
if matches!(spec.kind, DbKind::State) {
repair_legacy_recency_migration_version(&pool, migrator).await?;
}
migrator.run(&pool).await.map_err(anyhow::Error::from)
}
.await;
telemetry::record_init_result(
telemetry_override,
spec.kind,
spec.migrate_phase,
started.elapsed(),
&migrate_result,
);
if let Err(source) = migrate_result {
pool.close().await;
return Err(
RuntimeDbInitError::new(spec.label, "migrate", path.as_path(), source).into(),
);
}
Ok(pool)
}
/// Open a writable Codex SQLite database, creating it if necessary. /// Open a writable Codex SQLite database, creating it if necessary.
pub async fn open_read_write_pool(&self, path: &Path) -> Result<SqlitePool, Error> { pub async fn open_read_write_pool(&self, path: &Path) -> Result<SqlitePool, Error> {
let options = SqliteConnectOptions::new() let options = SqliteConnectOptions::new()
+1
View File
@@ -21,6 +21,7 @@ codex-otel = { workspace = true }
codex-protocol = { workspace = true } codex-protocol = { workspace = true }
codex-rollout = { workspace = true } codex-rollout = { workspace = true }
codex-state = { workspace = true } codex-state = { workspace = true }
codex-utils-absolute-path = { workspace = true }
codex-utils-path = { workspace = true } codex-utils-path = { workspace = true }
futures = { workspace = true } futures = { workspace = true }
pulldown-cmark = { workspace = true } pulldown-cmark = { workspace = true }
@@ -20,7 +20,7 @@ pub(super) async fn create_thread(
})?; })?;
let config = RolloutConfig { let config = RolloutConfig {
codex_home: store.config.codex_home.clone(), codex_home: store.config.codex_home.clone(),
sqlite_home: store.config.sqlite_home.clone(), sqlite_home: store.config.sqlite.home().to_path_buf(),
cwd, cwd,
model_provider_id: params.metadata.model_provider.clone(), model_provider_id: params.metadata.model_provider.clone(),
generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled), generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled),
@@ -41,7 +41,7 @@ pub(super) async fn list_threads(
let state_db = store.state_db().await; let state_db = store.state_db().await;
let rollout_config = RolloutConfig { let rollout_config = RolloutConfig {
codex_home: store.config.codex_home.clone(), codex_home: store.config.codex_home.clone(),
sqlite_home: store.config.sqlite_home.clone(), sqlite_home: store.config.sqlite.home().to_path_buf(),
cwd: store.config.codex_home.clone(), cwd: store.config.codex_home.clone(),
model_provider_id: store.config.default_model_provider_id.clone(), model_provider_id: store.config.default_model_provider_id.clone(),
generate_memories: false, generate_memories: false,
@@ -93,7 +93,7 @@ pub(super) async fn resume_thread(
})?; })?;
let config = RolloutConfig { let config = RolloutConfig {
codex_home: store.config.codex_home.clone(), codex_home: store.config.codex_home.clone(),
sqlite_home: store.config.sqlite_home.clone(), sqlite_home: store.config.sqlite.home().to_path_buf(),
cwd, cwd,
model_provider_id: params.metadata.model_provider.clone(), model_provider_id: params.metadata.model_provider.clone(),
generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled), generate_memories: matches!(params.metadata.memory_mode, ThreadMemoryMode::Enabled),
+21 -12
View File
@@ -22,6 +22,8 @@ use codex_protocol::ThreadId;
use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadHistoryMode;
use codex_rollout::RolloutRecorder; use codex_rollout::RolloutRecorder;
use codex_rollout::StateDbHandle; use codex_rollout::StateDbHandle;
use codex_state::SqliteConfig;
use codex_utils_absolute_path::AbsolutePathBuf;
use std::collections::HashMap; use std::collections::HashMap;
use std::collections::hash_map::Entry; use std::collections::hash_map::Entry;
use std::path::PathBuf; use std::path::PathBuf;
@@ -115,16 +117,23 @@ impl LiveWriterLocks {
#[derive(Clone, Debug, PartialEq, Eq)] #[derive(Clone, Debug, PartialEq, Eq)]
pub struct LocalThreadStoreConfig { pub struct LocalThreadStoreConfig {
pub codex_home: PathBuf, pub codex_home: PathBuf,
pub sqlite_home: PathBuf, pub sqlite: SqliteConfig,
/// Provider used only when older local metadata does not contain one. /// Provider used only when older local metadata does not contain one.
pub default_model_provider_id: String, pub default_model_provider_id: String,
} }
impl LocalThreadStoreConfig { impl LocalThreadStoreConfig {
#[expect(
clippy::expect_used,
reason = "resolved Codex configuration paths must be absolute"
)]
pub fn from_config(config: &impl codex_rollout::RolloutConfigView) -> Self { pub fn from_config(config: &impl codex_rollout::RolloutConfigView) -> Self {
Self { Self {
codex_home: config.codex_home().to_path_buf(), codex_home: config.codex_home().to_path_buf(),
sqlite_home: config.sqlite_home().to_path_buf(), sqlite: SqliteConfig::from_sqlite_home(
AbsolutePathBuf::from_absolute_path_checked(config.sqlite_home())
.expect("sqlite home from resolved config should be absolute"),
),
default_model_provider_id: config.model_provider_id().to_string(), default_model_provider_id: config.model_provider_id().to_string(),
} }
} }
@@ -158,7 +167,7 @@ impl LocalThreadStore {
async fn thread_history_db(&self) -> ThreadStoreResult<&sqlx::SqlitePool> { async fn thread_history_db(&self) -> ThreadStoreResult<&sqlx::SqlitePool> {
self.thread_history_db self.thread_history_db
.get_or_try_init(|| async { .get_or_try_init(|| async {
codex_state::open_thread_history_db(self.config.sqlite_home.as_path()).await codex_state::open_thread_history_db(self.config.sqlite.home()).await
}) })
.await .await
.map_err(|err| ThreadStoreError::Internal { .map_err(|err| ThreadStoreError::Internal {
@@ -499,7 +508,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -534,7 +543,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -569,7 +578,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -637,7 +646,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -726,7 +735,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -762,7 +771,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -802,7 +811,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -846,7 +855,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -901,7 +910,7 @@ mod tests {
let external_home = TempDir::new().expect("external temp dir"); let external_home = TempDir::new().expect("external temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
+12 -12
View File
@@ -608,7 +608,7 @@ mod tests {
let active_path = let active_path =
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -767,7 +767,7 @@ mod tests {
let rollout_path = let rollout_path =
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -812,7 +812,7 @@ mod tests {
) )
.expect("session file"); .expect("session file");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -855,7 +855,7 @@ mod tests {
let rollout_path = let rollout_path =
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -896,7 +896,7 @@ mod tests {
let rollout_path = let rollout_path =
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -930,7 +930,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1066,7 +1066,7 @@ mod tests {
writeln!(file, "{meta}").expect("write session meta"); writeln!(file, "{meta}").expect("write session meta");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1120,7 +1120,7 @@ mod tests {
write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file"); write_session_file(home.path(), "2025-01-03T12-00-00", uuid).expect("session file");
let stale_path = external.path().join("missing-rollout.jsonl"); let stale_path = external.path().join("missing-rollout.jsonl");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1171,7 +1171,7 @@ mod tests {
let stale_path = write_session_file(external.path(), "2025-01-04T12-00-00", other_uuid) let stale_path = write_session_file(external.path(), "2025-01-04T12-00-00", other_uuid)
.expect("other session file"); .expect("other session file");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1270,7 +1270,7 @@ mod tests {
.path() .path()
.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")); .join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl"));
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1332,7 +1332,7 @@ mod tests {
.path() .path()
.join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl")); .join(format!("rollout-2025-01-03T12-00-00-{uuid}.jsonl"));
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1387,7 +1387,7 @@ mod tests {
let archived_path = write_archived_session_file(home.path(), "2025-01-03T12-00-00", uuid) let archived_path = write_archived_session_file(home.path(), "2025-01-03T12-00-00", uuid)
.expect("archived session file"); .expect("archived session file");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -60,7 +60,7 @@ pub(super) async fn search_threads(
let state_db = store.state_db().await; let state_db = store.state_db().await;
let rollout_config = RolloutConfig { let rollout_config = RolloutConfig {
codex_home: store.config.codex_home.clone(), codex_home: store.config.codex_home.clone(),
sqlite_home: store.config.sqlite_home.clone(), sqlite_home: store.config.sqlite.home().to_path_buf(),
cwd: store.config.codex_home.clone(), cwd: store.config.codex_home.clone(),
model_provider_id: store.config.default_model_provider_id.clone(), model_provider_id: store.config.default_model_provider_id.clone(),
generate_memories: false, generate_memories: false,
@@ -5,6 +5,7 @@ use std::path::PathBuf;
use codex_protocol::protocol::ThreadHistoryMode; use codex_protocol::protocol::ThreadHistoryMode;
use codex_rollout::ARCHIVED_SESSIONS_SUBDIR; use codex_rollout::ARCHIVED_SESSIONS_SUBDIR;
use codex_utils_absolute_path::test_support::PathExt;
use uuid::Uuid; use uuid::Uuid;
use super::LocalThreadStoreConfig; use super::LocalThreadStoreConfig;
@@ -12,7 +13,7 @@ use super::LocalThreadStoreConfig;
pub(super) fn test_config(codex_home: &Path) -> LocalThreadStoreConfig { pub(super) fn test_config(codex_home: &Path) -> LocalThreadStoreConfig {
LocalThreadStoreConfig { LocalThreadStoreConfig {
codex_home: codex_home.to_path_buf(), codex_home: codex_home.to_path_buf(),
sqlite_home: codex_home.to_path_buf(), sqlite: codex_state::SqliteConfig::new_for_testing(codex_home.abs()),
default_model_provider_id: "test-provider".to_string(), default_model_provider_id: "test-provider".to_string(),
} }
} }
@@ -32,7 +32,7 @@ pub(super) async fn next_rollout_byte_offset(
store: &LocalThreadStore, store: &LocalThreadStore,
thread_id: ThreadId, thread_id: ThreadId,
) -> ThreadStoreResult<u64> { ) -> ThreadStoreResult<u64> {
let db_path = codex_state::thread_history_db_path(store.config.sqlite_home.as_path()); let db_path = store.config.sqlite.thread_history_db_path();
if !tokio::fs::try_exists(db_path.as_path()) if !tokio::fs::try_exists(db_path.as_path())
.await .await
.map_err(thread_history_error)? .map_err(thread_history_error)?
@@ -140,7 +140,7 @@ pub(super) async fn delete_thread(
store: &LocalThreadStore, store: &LocalThreadStore,
thread_id: ThreadId, thread_id: ThreadId,
) -> ThreadStoreResult<()> { ) -> ThreadStoreResult<()> {
let db_path = codex_state::thread_history_db_path(store.config.sqlite_home.as_path()); let db_path = store.config.sqlite.thread_history_db_path();
if !tokio::fs::try_exists(db_path.as_path()) if !tokio::fs::try_exists(db_path.as_path())
.await .await
.map_err(thread_history_delete_error)? .map_err(thread_history_delete_error)?
@@ -507,7 +507,7 @@ async fn store_with_mode(history_mode: ThreadHistoryMode) -> (TempDir, LocalThre
); );
} }
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -21,6 +21,7 @@ use codex_protocol::protocol::ThreadMemoryMode;
use codex_protocol::protocol::TurnCompleteEvent; use codex_protocol::protocol::TurnCompleteEvent;
use codex_protocol::protocol::TurnStartedEvent; use codex_protocol::protocol::TurnStartedEvent;
use codex_rollout::RolloutRecorder; use codex_rollout::RolloutRecorder;
use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use tempfile::TempDir; use tempfile::TempDir;
@@ -456,7 +457,7 @@ async fn summary_items_use_final_answers_and_ignore_commentary() {
let config = test_config(home.path()); let config = test_config(home.path());
let thread_id = ThreadId::default(); let thread_id = ThreadId::default();
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -832,7 +833,11 @@ async fn jsonl_failure_does_not_create_projection_database() {
.await .await
.expect_err("JSONL append should fail"); .expect_err("JSONL append should fail");
assert!(!codex_state::thread_history_db_path(home.path()).exists()); assert!(
!codex_state::SqliteConfig::new_for_testing(home.path().abs())
.thread_history_db_path()
.exists()
);
} }
#[tokio::test] #[tokio::test]
@@ -866,7 +871,7 @@ async fn sqlite_failure_does_not_fail_durable_jsonl_write() {
let sqlite_home = home.path().join("not-a-directory"); let sqlite_home = home.path().join("not-a-directory");
fs::write(sqlite_home.as_path(), "not a directory").expect("block sqlite home"); fs::write(sqlite_home.as_path(), "not a directory").expect("block sqlite home");
let mut config = test_config(home.path()); let mut config = test_config(home.path());
config.sqlite_home = sqlite_home; config.sqlite = codex_state::SqliteConfig::new_for_testing(sqlite_home.as_path().abs());
let store = LocalThreadStore::new(config, /*state_db*/ None); let store = LocalThreadStore::new(config, /*state_db*/ None);
let thread_id = ThreadId::default(); let thread_id = ThreadId::default();
create_paginated_thread(&store, thread_id).await; create_paginated_thread(&store, thread_id).await;
@@ -1095,7 +1095,7 @@ mod tests {
let path = let path =
write_session_file(home.path(), "2025-01-03T18-30-00", uuid).expect("session file"); write_session_file(home.path(), "2025-01-03T18-30-00", uuid).expect("session file");
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1200,7 +1200,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1243,7 +1243,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1298,7 +1298,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
@@ -1356,7 +1356,7 @@ mod tests {
let home = TempDir::new().expect("temp dir"); let home = TempDir::new().expect("temp dir");
let config = test_config(home.path()); let config = test_config(home.path());
let runtime = codex_state::StateRuntime::init( let runtime = codex_state::StateRuntime::init(
config.sqlite_home.clone(), config.sqlite.home().to_path_buf(),
config.default_model_provider_id.clone(), config.default_model_provider_id.clone(),
) )
.await .await
+15 -3
View File
@@ -290,7 +290,15 @@ async fn init_state_db_for_app_server_target(
match app_server_target { match app_server_target {
AppServerTarget::Embedded => state_db::try_init(config).await.map(Some).map_err(|err| { AppServerTarget::Embedded => state_db::try_init(config).await.map(Some).map_err(|err| {
let database_path = codex_state::runtime_db_path_for_corruption_error(&err) let database_path = codex_state::runtime_db_path_for_corruption_error(&err)
.unwrap_or_else(|| codex_state::state_db_path(config.sqlite_home.as_path())); .unwrap_or_else(|| {
codex_state::SqliteConfig::from_sqlite_home(
AbsolutePathBuf::resolve_path_against_base(
&config.sqlite_home,
&config.codex_home,
),
)
.state_db_path()
});
std::io::Error::other(LocalStateDbStartupError::new( std::io::Error::other(LocalStateDbStartupError::new(
database_path, database_path,
format!("{err:#}"), format!("{err:#}"),
@@ -2027,6 +2035,7 @@ mod tests {
use codex_app_server_protocol::ThreadStartParams; use codex_app_server_protocol::ThreadStartParams;
use codex_app_server_protocol::ThreadStartResponse; use codex_app_server_protocol::ThreadStartResponse;
use codex_config::config_toml::ProjectConfig; use codex_config::config_toml::ProjectConfig;
use codex_utils_absolute_path::test_support::PathExt;
use pretty_assertions::assert_eq; use pretty_assertions::assert_eq;
use serial_test::serial; use serial_test::serial;
use tempfile::TempDir; use tempfile::TempDir;
@@ -3114,7 +3123,9 @@ mod tests {
assert_eq!( assert_eq!(
startup_error.state_db_path(), startup_error.state_db_path(),
codex_state::state_db_path(occupied_sqlite_home.as_path()).as_path() codex_state::SqliteConfig::new_for_testing(config.sqlite_home.as_path().abs())
.state_db_path()
.as_path()
); );
assert!( assert!(
startup_error startup_error
@@ -3132,7 +3143,8 @@ mod tests {
let mut config = build_config(&temp_dir).await?; let mut config = build_config(&temp_dir).await?;
let sqlite_home = temp_dir.path().join("sqlite-home"); let sqlite_home = temp_dir.path().join("sqlite-home");
std::fs::create_dir_all(&sqlite_home)?; std::fs::create_dir_all(&sqlite_home)?;
let logs_db_path = codex_state::logs_db_path(&sqlite_home); let logs_db_path =
codex_state::SqliteConfig::new_for_testing(sqlite_home.as_path().abs()).logs_db_path();
std::fs::write(&logs_db_path, "not a sqlite database")?; std::fs::write(&logs_db_path, "not a sqlite database")?;
config.sqlite_home = sqlite_home; config.sqlite_home = sqlite_home;