diff --git a/codex-rs/Cargo.lock b/codex-rs/Cargo.lock index b00ef1baf3..645acaabd1 100644 --- a/codex-rs/Cargo.lock +++ b/codex-rs/Cargo.lock @@ -3486,6 +3486,7 @@ dependencies = [ "futures", "rustix 1.1.4", "serde", + "tokio", ] [[package]] diff --git a/codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs b/codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs index 1a447050c3..d6a115bbe9 100644 --- a/codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs +++ b/codex-rs/app-server/src/request_processors/feedback_log_snapshot_tests.rs @@ -1,5 +1,6 @@ -//! Exercises the real SQLite writer, feedback ring buffer, and snapshot ordering. +//! Exercises SQLite log collection, legacy storage, and feedback snapshot ordering. use super::*; +use codex_core::config::ConfigBuilder; use codex_state::SqliteConfig; use codex_state::StateRuntime; use codex_utils_absolute_path::test_support::PathExt; @@ -21,7 +22,8 @@ async fn feedback_includes_flush_and_query_failures_in_the_same_submission() -> Scenario::Corruption, ] { let home = tempfile::tempdir()?; - let sqlite = SqliteConfig::new_for_testing(home.path().abs()); + let config = core_test_support::load_default_config_for_test(&home).await; + let sqlite = &config.sqlite; let runtime = StateRuntime::init(sqlite.clone(), "test-provider".to_string()).await?; let pool = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?; let feedback = CodexFeedback::new(); @@ -47,14 +49,16 @@ async fn feedback_includes_flush_and_query_failures_in_the_same_submission() -> let snapshot = feedback.snapshot(Some(thread_id)); tracing::info_span!("thread", thread_id = %thread_id) .in_scope(|| tracing::info!("latest log")); - let (snapshot, sqlite_logs) = collect_feedback_logs( + let (snapshot, sqlite_logs, legacy_logs) = collect_feedback_logs( &feedback, Some(&writer), Some(&runtime), + &config, snapshot, &[thread_id], ) .await; + assert_eq!(legacy_logs, None); assert_eq!(sqlite_logs.is_some(), scenario == Scenario::Healthy); let logs = String::from_utf8(snapshot.log_attachment(sqlite_logs).buffer)?; assert!(logs.contains("persisted log")); @@ -80,3 +84,104 @@ async fn feedback_includes_flush_and_query_failures_in_the_same_submission() -> } Ok(()) } + +#[tokio::test] +async fn feedback_preserves_legacy_logs_when_the_directory_is_unset_or_current_logs_are_empty() +-> anyhow::Result<()> { + for configured_log_dir in [false, true] { + let home = tempfile::tempdir()?; + let sqlite_home = home.path().join("sqlite"); + let mut overrides = vec![( + "sqlite_home".to_string(), + toml::Value::String(sqlite_home.to_string_lossy().into_owned()), + )]; + if configured_log_dir { + overrides.push(( + "log_dir".to_string(), + toml::Value::String(home.path().join("log").to_string_lossy().into_owned()), + )); + } + let config = ConfigBuilder::default() + .codex_home(home.path().to_path_buf()) + .loader_overrides(codex_config::LoaderOverrides::without_managed_config_for_tests()) + .cli_overrides(overrides) + .build() + .await?; + let legacy_sqlite = SqliteConfig::new_for_testing(sqlite_home.abs()); + let legacy = StateRuntime::init(legacy_sqlite.clone(), "test-provider".to_string()).await?; + let pool = legacy_sqlite + .open_read_write_pool(&legacy_sqlite.logs_db_path()) + .await?; + let thread_id = ThreadId::new(); + let now = chrono::Utc::now(); + let timestamp = now.format("%Y-%m-%dT%H:%M:%S.000000Z"); + for (id, body) in [ + (thread_id, "legacy history"), + (ThreadId::new(), "unrelated history"), + ] { + sqlx::query( + "INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) + VALUES (?, 0, 'INFO', 'test', ?, ?)", + ) + .bind(now.timestamp()) + .bind(body) + .bind(id.to_string()) + .execute(&pool) + .await?; + } + pool.close().await; + legacy.close().await; + let legacy_before = std::fs::read(legacy_sqlite.logs_db_path())?; + + let runtime = + StateRuntime::init(config.sqlite.clone(), "test-provider".to_string()).await?; + let pool = config + .sqlite + .open_read_write_pool(&config.sqlite.logs_db_path()) + .await?; + let feedback = CodexFeedback::new(); + for current_has_logs in [false, true] { + if current_has_logs { + sqlx::query( + "INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) + VALUES (?, 0, 'INFO', 'test', 'current history', ?)", + ) + .bind(now.timestamp()) + .bind(thread_id.to_string()) + .execute(&pool) + .await?; + } + let (_, current_logs, legacy_logs) = collect_feedback_logs( + &feedback, + /*log_db*/ None, + Some(&runtime), + &config, + feedback.snapshot(Some(thread_id)), + &[thread_id], + ) + .await; + let mut expected_current = String::new(); + if !configured_log_dir { + expected_current.push_str(&format!("{timestamp} INFO legacy history\n")); + } + if current_has_logs { + expected_current.push_str(&format!("{timestamp} INFO current history\n")); + } + assert_eq!( + current_logs, + (!expected_current.is_empty()).then(|| expected_current.into_bytes()) + ); + assert_eq!( + legacy_logs, + (configured_log_dir && !current_has_logs) + .then(|| format!("{timestamp} INFO legacy history\n").into_bytes()) + ); + } + if configured_log_dir { + assert_eq!(std::fs::read(legacy_sqlite.logs_db_path())?, legacy_before); + } + pool.close().await; + runtime.close().await; + } + Ok(()) +} diff --git a/codex-rs/app-server/src/request_processors/feedback_processor.rs b/codex-rs/app-server/src/request_processors/feedback_processor.rs index 0e33d9a9cc..b14279d188 100644 --- a/codex-rs/app-server/src/request_processors/feedback_processor.rs +++ b/codex-rs/app-server/src/request_processors/feedback_processor.rs @@ -8,6 +8,7 @@ use codex_connectors::ConnectorDirectoryCacheKey; use codex_connectors::connector_runtime_cache_path; use codex_feedback::CODEX_APP_DIRECTORY_CACHE_ATTACHMENT_FILENAME; use codex_feedback::CODEX_APPS_TOOLS_CACHE_ATTACHMENT_FILENAME; +use codex_feedback::FeedbackAttachment; use codex_feedback::FeedbackSnapshot; #[cfg(target_os = "windows")] use codex_feedback::WINDOWS_SANDBOX_LOG_ATTACHMENT_FILENAME; @@ -190,14 +191,22 @@ impl FeedbackRequestProcessor { feedback_index = Some(index); } extra_attachments.extend(failures.attachment); - let (snapshot, sqlite_feedback_logs) = collect_feedback_logs( + let (snapshot, sqlite_feedback_logs, legacy_logs) = collect_feedback_logs( &self.feedback, self.log_db.as_ref(), state_db_ctx.as_ref(), + &self.config, snapshot, &feedback_thread_ids, ) .await; + if let Some(buffer) = legacy_logs { + extra_attachments.push(FeedbackAttachment { + filename: "codex-logs-legacy.log".to_string(), + content_type: Some("text/plain".to_string()), + buffer, + }); + } (snapshot, sqlite_feedback_logs, state_db_ctx) } else { (snapshot, None, None) @@ -548,13 +557,15 @@ async fn collect_feedback_logs( feedback: &CodexFeedback, log_db: Option<&LogDbLayer>, state_db: Option<&StateDbHandle>, + config: &Config, mut snapshot: FeedbackSnapshot, thread_ids: &[ThreadId], -) -> (FeedbackSnapshot, Option>) { +) -> (FeedbackSnapshot, Option>, Option>) { if let Some(log_db) = log_db { log_db.flush().await; } let mut sqlite_logs = None; + let mut legacy_logs = None; if let Some(state_db) = state_db && !thread_ids.is_empty() { @@ -576,6 +587,21 @@ async fn collect_feedback_logs( ); } } + // Earlier versions stored logs in sqlite_home/logs_2.sqlite. + // Keep this temporary read path so feedback can include logs from before the directory change. + // Read the old database when log_dir is unset or the current query returns no logs. + if config.configured_log_dir().is_none() || sqlite_logs.is_none() { + match state_db + .query_legacy_feedback_logs_for_threads(&thread_id_refs) + .await + { + Ok(logs) if logs.is_empty() => {} + Ok(logs) => legacy_logs = Some(logs), + Err(err) => { + tracing::warn!("failed to query legacy feedback logs from sqlite: {err}") + } + } + } } if log_db.is_some_and(LogDbLayer::has_write_failure) { sqlite_logs = None; @@ -583,7 +609,7 @@ async fn collect_feedback_logs( // new logs might arrive to the buffer while collecting metadata e.g. if log db flush fails snapshot.refresh_logs(feedback); - (snapshot, sqlite_logs) + (snapshot, sqlite_logs, legacy_logs) } #[cfg(test)] diff --git a/codex-rs/app-server/tests/suite/logging.rs b/codex-rs/app-server/tests/suite/logging.rs index ce75967b80..ce6590ed2c 100644 --- a/codex-rs/app-server/tests/suite/logging.rs +++ b/codex-rs/app-server/tests/suite/logging.rs @@ -28,9 +28,8 @@ use codex_app_server_protocol::TurnStartResponse; use codex_app_server_protocol::UserInput; use codex_features::Feature; use codex_state::LogQuery; -use codex_state::SqliteConfig; use codex_state::StateRuntime; -use codex_utils_absolute_path::test_support::PathExt; +use core_test_support::load_default_config_for_test; use core_test_support::responses; use core_test_support::skip_if_no_network; use core_test_support::skip_if_wine_exec; @@ -145,7 +144,7 @@ async fn submission_metadata_reaches_persisted_and_feedback_logs() -> Result<()> .await??; let state = StateRuntime::init( - SqliteConfig::new_for_testing(codex_home.path().abs()), + load_default_config_for_test(&codex_home).await.sqlite, "mock_provider".into(), ) .await?; @@ -368,7 +367,7 @@ supports_websockets = false .send_response(RequestId::String(barrier.into()), json!({})) .await?; let state = StateRuntime::init( - SqliteConfig::new_for_testing(codex_home.path().abs()), + load_default_config_for_test(&codex_home).await.sqlite, "mock_provider".into(), ) .await?; @@ -580,7 +579,7 @@ async fn sqlite_log_metrics_exports_do_not_create_log_cycles() -> Result<()> { } let state = codex_state::StateRuntime::init( - codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()), + load_default_config_for_test(&codex_home).await.sqlite, "test-provider".to_string(), ) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/feedback.rs b/codex-rs/app-server/tests/suite/v2/feedback.rs index afe5d57ed3..aa573d49ef 100644 --- a/codex-rs/app-server/tests/suite/v2/feedback.rs +++ b/codex-rs/app-server/tests/suite/v2/feedback.rs @@ -98,8 +98,6 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures use app_test_support::MockResponsesConfig; use codex_app_server_protocol::ClientNotification; - use codex_state::SqliteConfig; - use codex_utils_absolute_path::test_support::PathExt; use tokio::io::AsyncReadExt; use tokio_rustls::TlsAcceptor; use tokio_rustls::rustls; @@ -189,9 +187,13 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures })); let models = wiremock::MockServer::start().await; - MockResponsesConfig::new(&models.uri()).write(home.path())?; + MockResponsesConfig::new(&models.uri()) + .with_root_config(&format!("log_dir = {:?}", home.path().join("log"))) + .write(home.path())?; let thread_id = codex_protocol::ThreadId::new(); - let sqlite = SqliteConfig::new_for_testing(home.path().abs()); + let sqlite = core_test_support::load_default_config_for_test(&home) + .await + .sqlite; let state = codex_state::StateRuntime::init(sqlite.clone(), "mock_provider".to_string()).await?; let review = json!({ @@ -228,6 +230,19 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures .await?; } state.close().await; + let legacy_path = home.path().join("logs_2.sqlite"); + std::fs::copy(sqlite.logs_db_path(), &legacy_path)?; + let legacy_pool = sqlite.open_read_write_pool(&legacy_path).await?; + let legacy_timestamp = chrono::Utc::now(); + sqlx::query( + "INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) + VALUES (?, 0, 'INFO', 'test', 'legacy history', ?)", + ) + .bind(legacy_timestamp.timestamp()) + .bind(thread_id.to_string()) + .execute(&legacy_pool) + .await?; + legacy_pool.close().await; // This new process can recover review context only from SQLite. let mut app_server = TestAppServer::builder() .with_codex_home(home.path()) @@ -242,7 +257,7 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures .await?; let pool = sqlite.open_read_write_pool(&sqlite.logs_db_path()).await?; // This old history exists only in SQLite, as if it had already left the memory ring. - sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) VALUES (1, 0, 'INFO', 'test', 'sqlite-only-history', ?)") + sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target, feedback_log_body, thread_id) VALUES (unixepoch(), 0, 'INFO', 'test', 'sqlite-only-history', ?)") .bind(thread_id.to_string()).execute(&pool).await?; sqlx::query("CREATE TRIGGER fail_log_insert BEFORE INSERT ON logs BEGIN SELECT RAISE(ABORT, 'synthetic-secret'); END") .execute(&pool).await?; @@ -317,8 +332,18 @@ async fn feedback_upload_recovers_persisted_evidence_and_reports_sqlite_failures if phase == Phase::WithoutLogs { assert!(!attachments.contains_key("auto-review-failures.jsonl")); assert!(!attachments.contains_key("codex-logs.log")); + assert!(!attachments.contains_key("codex-logs-legacy.log")); continue; } + assert_eq!( + attachments.get("codex-logs-legacy.log"), + (phase == Phase::Corrupt) + .then(|| format!( + "{} INFO legacy history\n", + legacy_timestamp.format("%Y-%m-%dT%H:%M:%S.000000Z") + )) + .as_ref(), + ); assert_eq!( serde_json::from_str::(&attachments["auto-review-failures.jsonl"])?, review, diff --git a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs index 2f77ce8dfb..dcebba5533 100644 --- a/codex-rs/app-server/tests/suite/v2/guardian_v2.rs +++ b/codex-rs/app-server/tests/suite/v2/guardian_v2.rs @@ -1756,7 +1756,7 @@ async fn guardian_v2_routes_scoped_tool_approvals( && !late_root_restriction { let state_db = StateRuntime::init( - codex_state::SqliteConfig::new_for_testing(codex_home.path().abs()), + load_default_config_for_test(&codex_home).await.sqlite, "mock_provider".to_owned(), ) .await?; diff --git a/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs b/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs index c97a9785f2..a3e94c74c2 100644 --- a/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs +++ b/codex-rs/app-server/tests/suite/v2/sqlite_recovery.rs @@ -14,69 +14,86 @@ use tokio::time::timeout; #[tokio::test] async fn startup_reports_shared_and_fallback_recovery() -> Result<()> { - let home = TempDir::new()?; - let sqlite = SqliteConfig::new_for_testing(home.path().canonicalize()?.as_path().abs()); - let runtime = StateRuntime::init(sqlite.clone(), "openai".to_string()).await?; - runtime.close().await; - let logs_path = sqlite.logs_db_path(); - let pool = sqlite.open_read_write_pool(&logs_path).await?; - // The schema and migrations remain readable; opening the database alone succeeds. - sqlx::raw_sql( - "CREATE TABLE sample(value INTEGER); - INSERT INTO sample VALUES (NULL); - PRAGMA writable_schema=ON; - UPDATE sqlite_schema SET sql='CREATE TABLE sample(value INTEGER NOT NULL)' WHERE name='sample'; - PRAGMA writable_schema=OFF;", - ) - .execute(&pool) - .await?; - pool.close().await; - - // Logs recover inside the pool opener; the unreadable goals header forces - // the outer startup fallback and a retry after that first recovery. - tokio::fs::write(sqlite.goals_db_path(), b"damaged goals database").await?; - - let read_timeout = Duration::from_secs(/*secs*/ 30); - let mut server = TestAppServer::builder() - .with_codex_home(sqlite.home()) - .build_initialized_with_timeout(read_timeout) - .await?; - let warning: ConfigWarningNotification = - timeout(read_timeout, server.read_notification("configWarning")).await??; - assert_eq!(warning.summary, "Codex rebuilt its local database"); - let details = warning.details.expect("recovery details"); - assert!(details.contains("Some database-only metadata may be unavailable")); - let backups = std::fs::read_dir(sqlite.home().join("db-backups"))? - .collect::>>()?; - assert_eq!(backups.len(), 2); - for backup in &backups { - assert!(details.contains(&backup.path().display().to_string())); - } - assert!(details.contains(&sqlite.logs_db_path().display().to_string())); - assert!(details.contains(&sqlite.goals_db_path().display().to_string())); - let backup_dir = backups - .iter() - .map(std::fs::DirEntry::path) - .find(|path| { - path.join(logs_path.file_name().expect("log filename")) - .exists() - }) - .expect("preserved logs database"); - let backup_pool = sqlite - .open_read_only_pool( - &backup_dir.join(logs_path.file_name().expect("log filename")), - /*busy_timeout*/ None, + for log_dir_is_configured in [false, true] { + let home = TempDir::new()?; + if log_dir_is_configured { + std::fs::write( + home.path().join("config.toml"), + format!("log_dir = {:?}\n", home.path().join("log")), + )?; + } + let sqlite = core_test_support::load_default_config_for_test(&home) + .await + .sqlite; + let runtime = StateRuntime::init(sqlite.clone(), "openai".to_string()).await?; + runtime.close().await; + let logs_path = sqlite.logs_db_path(); + let pool = sqlite.open_read_write_pool(&logs_path).await?; + // The schema and migrations remain readable; opening the database alone succeeds. + sqlx::raw_sql( + "CREATE TABLE sample(value INTEGER); + INSERT INTO sample VALUES (NULL); + PRAGMA writable_schema=ON; + UPDATE sqlite_schema SET sql='CREATE TABLE sample(value INTEGER NOT NULL)' WHERE name='sample'; + PRAGMA writable_schema=OFF;", ) + .execute(&pool) .await?; - assert_eq!( - sqlx::query_scalar::<_, Option>("SELECT value FROM sample") - .fetch_all(&backup_pool) - .await?, - vec![None] - ); - backup_pool.close().await; - server.start_thread(ThreadStartParams::default()).await?; - timeout(read_timeout, server.shutdown_gracefully()).await??; + pool.close().await; + + // Logs recover inside the pool opener; the unreadable goals header forces + // the outer startup fallback and a retry after that first recovery. + tokio::fs::write(sqlite.goals_db_path(), b"damaged goals database").await?; + + let read_timeout = Duration::from_secs(/*secs*/ 30); + let mut server = TestAppServer::builder() + .with_codex_home(sqlite.home()) + .build_initialized_with_timeout(read_timeout) + .await?; + let warning: ConfigWarningNotification = + timeout(read_timeout, server.read_notification("configWarning")).await??; + assert_eq!(warning.summary, "Codex rebuilt its local database"); + let details = warning.details.expect("recovery details"); + assert!(details.contains("Some database-only metadata may be unavailable")); + let mut backups = std::fs::read_dir(sqlite.home().join("db-backups"))? + .collect::>>()?; + let logs_home = logs_path.parent().expect("logs directory"); + if logs_home != sqlite.home() { + backups.extend( + std::fs::read_dir(logs_home.join("db-backups"))? + .collect::>>()?, + ); + } + assert_eq!(backups.len(), 2); + for backup in &backups { + assert!(details.contains(&backup.path().display().to_string())); + } + assert!(details.contains(&sqlite.logs_db_path().display().to_string())); + assert!(details.contains(&sqlite.goals_db_path().display().to_string())); + let backup_dir = backups + .iter() + .map(std::fs::DirEntry::path) + .find(|path| { + path.join(logs_path.file_name().expect("log filename")) + .exists() + }) + .expect("preserved logs database"); + let backup_pool = sqlite + .open_read_only_pool( + &backup_dir.join(logs_path.file_name().expect("log filename")), + /*busy_timeout*/ None, + ) + .await?; + assert_eq!( + sqlx::query_scalar::<_, Option>("SELECT value FROM sample") + .fetch_all(&backup_pool) + .await?, + vec![None] + ); + backup_pool.close().await; + server.start_thread(ThreadStartParams::default()).await?; + timeout(read_timeout, server.shutdown_gracefully()).await??; + } Ok(()) } diff --git a/codex-rs/cli/tests/app_server.rs b/codex-rs/cli/tests/app_server.rs index 3688a0b47b..be852c6143 100644 --- a/codex-rs/cli/tests/app_server.rs +++ b/codex-rs/cli/tests/app_server.rs @@ -132,6 +132,64 @@ fn app_server_emits_json_info_events() -> Result<()> { Ok(()) } +#[test] +fn app_server_keeps_sqlite_logs_in_overridden_sqlite_home() -> Result<()> { + let home = TempDir::new()?; + let sqlite_home = TempDir::new()?; + std::fs::write( + home.path().join("config.toml"), + "[features]\nplugins = false\n", + )?; + + codex_command(home.path())? + .env("CODEX_SQLITE_HOME", sqlite_home.path()) + .env( + "CODEX_APP_SERVER_MANAGED_CONFIG_PATH", + home.path().join("managed_config.toml"), + ) + .arg("app-server") + .write_stdin("") + .assert() + .success(); + + assert!(sqlite_home.path().join("state_5.sqlite").is_file()); + assert!(std::fs::metadata(sqlite_home.path().join("logs_2.sqlite"))?.len() > 0); + assert!(!home.path().join("logs_2.sqlite").exists()); + assert!(!home.path().join("log/logs_2.sqlite").exists()); + Ok(()) +} + +#[cfg(unix)] +#[test] +fn app_server_preserves_the_existing_sqlite_logs_symlink() -> Result<()> { + let home = TempDir::new()?; + let external_logs = TempDir::new()?; + let target = external_logs.path().join("logs_2.sqlite"); + let link = home.path().join("logs_2.sqlite"); + std::os::unix::fs::symlink(&target, &link)?; + std::fs::write( + home.path().join("config.toml"), + "[features]\nplugins = false\n", + )?; + + codex_command(home.path())? + .env_remove("CODEX_SQLITE_HOME") + .env( + "CODEX_APP_SERVER_MANAGED_CONFIG_PATH", + home.path().join("managed_config.toml"), + ) + .arg("app-server") + .write_stdin("") + .assert() + .success(); + + assert_eq!(std::fs::read_link(&link)?, target); + assert!(std::fs::metadata(&target)?.len() > 0); + assert!(home.path().join("state_5.sqlite").is_file()); + assert!(!home.path().join("log/logs_2.sqlite").exists()); + Ok(()) +} + #[test] fn print_startup_config_preserves_project_login_restrictions() -> Result<()> { let home = TempDir::new()?; diff --git a/codex-rs/cli/tests/doctor_path_safety.rs b/codex-rs/cli/tests/doctor_path_safety.rs index bbc4f9f7bb..cd5d7182ec 100644 --- a/codex-rs/cli/tests/doctor_path_safety.rs +++ b/codex-rs/cli/tests/doctor_path_safety.rs @@ -223,7 +223,7 @@ fn doctor_reports_failed_database_paths() -> Result<()> { let sqlite_home = toml::Value::String(sqlite_home.display().to_string()); std::fs::write( &config_file, - format!("sqlite_home = {sqlite_home}\n{config}"), + format!("sqlite_home = {sqlite_home}\nlog_dir = {sqlite_home}\n{config}"), )?; for (snapshot_name, failed_paths) in [ @@ -317,7 +317,7 @@ fn doctor_redacts_failed_database_paths_in_json() -> Result<()> { let sqlite_home = toml::Value::String(sqlite_home.display().to_string()); std::fs::write( &config_file, - format!("sqlite_home = {sqlite_home}\n{config}"), + format!("sqlite_home = {sqlite_home}\nlog_dir = {sqlite_home}\n{config}"), )?; std::fs::write(sqlite.logs_db_path(), "not a SQLite database")?; diff --git a/codex-rs/cli/tests/login.rs b/codex-rs/cli/tests/login.rs index 00f4af1365..476cf8613f 100644 --- a/codex-rs/cli/tests/login.rs +++ b/codex-rs/cli/tests/login.rs @@ -87,6 +87,43 @@ fn login_with_api_key_reads_stdin_and_writes_auth_json() -> Result<()> { Ok(()) } +#[test] +fn api_key_auth_commands_work_when_log_dir_is_unavailable() -> Result<()> { + let codex_home = TempDir::new()?; + let log_dir = codex_home.path().join("unavailable-logs"); + std::fs::write(&log_dir, "not a directory")?; + std::fs::write( + codex_home.path().join("config.toml"), + format!("cli_auth_credentials_store = \"file\"\nlog_dir = {log_dir:?}\n"), + )?; + + codex_command(codex_home.path())? + .args(["login", "--with-api-key"]) + .write_stdin("sk-test\n") + .assert() + .success() + .stderr(contains("Warning: failed to create login log directory")); + assert_eq!( + read_auth_json(codex_home.path())?["OPENAI_API_KEY"], + "sk-test" + ); + + codex_command(codex_home.path())? + .args(["login", "status"]) + .assert() + .success() + .stderr(contains("API key")); + + codex_command(codex_home.path())? + .arg("logout") + .assert() + .success() + .stderr(contains("Successfully logged out")); + assert!(!codex_home.path().join("auth.json").exists()); + + Ok(()) +} + #[test] fn login_status_reports_auth_storage_errors() -> Result<()> { let codex_home = TempDir::new()?; diff --git a/codex-rs/config/src/config_toml.rs b/codex-rs/config/src/config_toml.rs index e1b08bf12b..7f70fb8814 100644 --- a/codex-rs/config/src/config_toml.rs +++ b/codex-rs/config/src/config_toml.rs @@ -374,9 +374,11 @@ pub struct ConfigToml { /// Defaults to `$CODEX_SQLITE_HOME` when set. Otherwise uses `$CODEX_HOME`. pub sqlite_home: Option, - /// Directory where Codex writes log files. Setting this value explicitly - /// also enables the TUI text log in this directory. - /// Defaults to `$CODEX_HOME/log`. + /// Directory for log files. Setting this value explicitly also routes the + /// SQLite logs database here and enables the TUI text log in this directory. + /// When unset, log files use `$CODEX_HOME/log` and SQLite logs use `sqlite_home`. + /// Existing SQLite logs in `sqlite_home` are not migrated automatically. + /// A managed `sqlite_home` requirement keeps SQLite logs in that required directory. pub log_dir: Option, /// Optional URI-based file opener. If set, citations to files in the model diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 9f35ee5c90..83be6f9f06 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -7894,7 +7894,7 @@ "$ref": "#/definitions/AbsolutePathBuf" } ], - "description": "Directory where Codex writes log files. Setting this value explicitly also enables the TUI text log in this directory. Defaults to `$CODEX_HOME/log`." + "description": "Directory for log files. Setting this value explicitly also routes the SQLite logs database here and enables the TUI text log in this directory. When unset, log files use `$CODEX_HOME/log` and SQLite logs use `sqlite_home`. Existing SQLite logs in `sqlite_home` are not migrated automatically. A managed `sqlite_home` requirement keeps SQLite logs in that required directory." }, "marketplaces": { "additionalProperties": { diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index b5138f4240..02cb50cf8a 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -6130,11 +6130,86 @@ async fn sqlite_home_defaults_to_codex_home_for_workspace_write() -> std::io::Re ) .await?; - assert_eq!(config.sqlite.home(), codex_home.path()); + assert_eq!( + ( + config.sqlite.home(), + config.sqlite.logs_db_path(), + &config.log_dir + ), + ( + codex_home.path(), + codex_home.path().join("logs_2.sqlite"), + &codex_home.path().join("log"), + ), + ); + assert_eq!(config.configured_log_dir(), None); Ok(()) } +#[tokio::test] +async fn log_dir_routes_logs_without_moving_state() -> std::io::Result<()> { + let home = TempDir::new()?; + let diagnostics = home.path().join("diagnostics").abs(); + std::fs::write( + home.path().join(CONFIG_TOML_FILE), + format!("log_dir = {:?}\n", diagnostics.as_path()), + )?; + let config = ConfigBuilder::without_managed_config_for_tests() + .codex_home(home.path().to_path_buf()) + .fallback_cwd(Some(home.path().to_path_buf())) + .build() + .await?; + assert_eq!( + ( + config.sqlite.home(), + config.sqlite.logs_db_path().parent(), + config.log_dir.as_path(), + config.configured_log_dir() + ), + ( + home.path(), + Some(diagnostics.as_path()), + diagnostics.as_path(), + Some(diagnostics.as_path()) + ), + ); + Ok(()) +} + +#[tokio::test] +async fn managed_sqlite_home_preserves_configured_log_dir() -> std::io::Result<()> { + let home = TempDir::new()?; + let diagnostics = home.path().join("diagnostics"); + let state = home.path().join("managed-state"); + for (user_config, requirements) in [ + ( + format!("log_dir = {diagnostics:?}\n"), + format!("sqlite_home = {state:?}"), + ), + ( + String::new(), + format!("sqlite_home = {state:?}\nlog_dir = {diagnostics:?}"), + ), + ] { + std::fs::write(home.path().join(CONFIG_TOML_FILE), user_config)?; + let config = load_with_enterprise_requirement(&home, requirements).await?; + assert_eq!( + ( + config.sqlite.logs_db_path().parent(), + config.log_dir.as_path(), + config.configured_log_dir(), + ), + ( + Some(state.as_path()), + diagnostics.as_path(), + Some(diagnostics.as_path()), + ), + ); + } + Ok(()) +} + #[tokio::test] async fn workspace_write_includes_configured_writable_root_once_without_memories_root() -> std::io::Result<()> { @@ -13518,6 +13593,8 @@ async fn load_with_enterprise_requirement( #[tokio::test] async fn exact_requirements_apply_to_runtime_config() -> std::io::Result<()> { let codex_home = TempDir::new()?; + let unused_diagnostics = codex_home.path().join("unused-diagnostics"); + std::fs::write(&unused_diagnostics, "not a directory")?; let catalog_path = codex_home.path().join("required-models.json"); let mut catalog = bundled_models_response() .unwrap_or_else(|err| panic!("bundled models.json should parse: {err}")); @@ -13528,13 +13605,16 @@ async fn exact_requirements_apply_to_runtime_config() -> std::io::Result<()> { )?; std::fs::write( codex_home.path().join(CONFIG_TOML_FILE), - r#" + format!( + r#" +log_dir = {unused_diagnostics:?} check_for_update_on_startup = true allow_login_shell = true [feedback] enabled = true -"#, +"# + ), )?; let required_sqlite_home = codex_home.path().join("required-state"); @@ -13558,6 +13638,16 @@ enabled = false assert_eq!(config.sqlite.home(), required_sqlite_home.as_path()); assert_eq!(config.log_dir, required_log_dir); + assert_eq!( + ( + config.sqlite.logs_db_path().parent(), + config.configured_log_dir(), + ), + ( + Some(required_sqlite_home.as_path()), + Some(required_log_dir.as_path()) + ), + ); assert_eq!(config.model_catalog, Some(catalog)); assert!(!config.check_for_update_on_startup); assert!(!config.permissions.allow_login_shell); diff --git a/codex-rs/core/src/config/mod.rs b/codex-rs/core/src/config/mod.rs index 92787f11b4..1ee9f9e0ea 100644 --- a/codex-rs/core/src/config/mod.rs +++ b/codex-rs/core/src/config/mod.rs @@ -995,7 +995,9 @@ pub struct Config { /// Resolved configuration shared by all Codex SQLite databases. pub sqlite: codex_state::SqliteConfig, - /// Directory where Codex writes log files (defaults to `$CODEX_HOME/log`). + /// Directory for log files (defaults to `$CODEX_HOME/log`). + /// SQLite logs use this directory only when `log_dir` is explicitly configured + /// and no managed `sqlite_home` requirement applies. pub log_dir: PathBuf, /// Settings that govern if and what will be written to `~/.codex/history.jsonl`. @@ -1628,6 +1630,21 @@ impl Config { &self.sqlite } + /// Returns `log_dir` when explicitly set by configuration or managed requirements. + pub fn configured_log_dir(&self) -> Option<&Path> { + let configured = self + .config_layer_stack + .effective_config() + .as_table() + .is_some_and(|table| table.contains_key("log_dir")) + || self + .config_layer_stack + .requirements_toml() + .log_dir + .is_some(); + configured.then_some(self.log_dir.as_path()) + } + /// Resolves the configured, reviewer-catalog, or bundled Guardian policy. pub fn resolve_guardian_policy<'a>( &'a self, @@ -4241,6 +4258,11 @@ impl Config { .cloned() .or(sqlite_home_env) .unwrap_or_else(|| codex_home.clone()); + let mut sqlite = codex_state::SqliteConfig::from_sqlite_home(sqlite_home); + let requirements = config_layer_stack.requirements_toml(); + if cfg.log_dir.is_some() && requirements.sqlite_home.is_none() { + sqlite = sqlite.with_local_logs_home(log_dir.clone()); + } let original_permission_profile = permission_profile.clone(); apply_requirement_constrained_value( "approval_policy", @@ -4538,7 +4560,7 @@ impl Config { memories: memories_config, agent_interrupt_message_enabled, codex_home, - sqlite: codex_state::SqliteConfig::from_sqlite_home(sqlite_home), + sqlite, log_dir, config_layer_stack, application_network_policy: Default::default(), diff --git a/codex-rs/core/src/session/tests.rs b/codex-rs/core/src/session/tests.rs index 8446df3831..f6999ef92f 100644 --- a/codex-rs/core/src/session/tests.rs +++ b/codex-rs/core/src/session/tests.rs @@ -9026,7 +9026,7 @@ where .await } -async fn make_session_and_context_with_auth_config_home_and_rx( +pub(crate) async fn make_session_and_context_with_auth_config_home_and_rx( auth: CodexAuth, dynamic_tools: Vec, codex_home: &Path, diff --git a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs index 27bee5dfa2..be73c65ac3 100644 --- a/codex-rs/core/src/tools/handlers/multi_agents_tests.rs +++ b/codex-rs/core/src/tools/handlers/multi_agents_tests.rs @@ -14,6 +14,7 @@ use crate::init_state_db; use crate::local_agent_graph_store_from_state_db; use crate::session::step_context::StepContext; use crate::session::tests::make_session_and_context; +use crate::session::tests::make_session_and_context_with_auth_config_home_and_rx; use crate::session::tests::update_selected_settings_for_test; use crate::session::tests::update_turn_settings_for_test; use crate::session::turn_context::TurnContext; @@ -3943,7 +3944,16 @@ async fn multi_agent_v2_interrupt_agent_accepts_task_name_target() { #[tokio::test] async fn multi_agent_v2_interrupt_agent_accepts_unloaded_task_name_target() { - let (mut session, mut turn) = make_session_and_context().await; + let codex_home = tempfile::tempdir().expect("create codex home"); + let (session, turn, _) = make_session_and_context_with_auth_config_home_and_rx( + CodexAuth::from_api_key("Test API Key"), + Vec::new(), + codex_home.path(), + |_config| {}, + ) + .await; + let mut session = Arc::into_inner(session).expect("unshared test session"); + let mut turn = Arc::into_inner(turn).expect("unshared test turn"); let mut config = (*turn.config).clone(); config.multi_agent_v2.max_concurrent_threads_per_session = 2; config @@ -4278,7 +4288,14 @@ async fn close_agent_submits_shutdown_and_returns_previous_status() { #[tokio::test] async fn tool_handlers_cascade_close_and_resume_and_keep_explicitly_closed_subtrees_closed() { - let (_session, turn) = make_session_and_context().await; + let codex_home = tempfile::tempdir().expect("create codex home"); + let (_session, turn, _) = make_session_and_context_with_auth_config_home_and_rx( + CodexAuth::from_api_key("Test API Key"), + Vec::new(), + codex_home.path(), + |_config| {}, + ) + .await; let mut config = turn.config.as_ref().clone(); config.agent_max_depth = 3; config diff --git a/codex-rs/core/tests/suite/sqlite_state.rs b/codex-rs/core/tests/suite/sqlite_state.rs index f863f9e3d8..25ebd9989a 100644 --- a/codex-rs/core/tests/suite/sqlite_state.rs +++ b/codex-rs/core/tests/suite/sqlite_state.rs @@ -1004,6 +1004,10 @@ async fn tool_call_logs_include_thread_id() -> Result<()> { }); log_db_layer.flush().await; + let logs_path = test.config.sqlite.home().join("logs_2.sqlite"); + assert_eq!(db.sqlite().logs_db_path(), logs_path); + assert!(logs_path.is_file()); + let mut found = None; for _ in 0..80 { let query = codex_state::LogQuery { diff --git a/codex-rs/file-system/Cargo.toml b/codex-rs/file-system/Cargo.toml index 006574084f..d2ee2bb994 100644 --- a/codex-rs/file-system/Cargo.toml +++ b/codex-rs/file-system/Cargo.toml @@ -14,6 +14,7 @@ codex-utils-absolute-path = { workspace = true } codex-utils-path-uri = { workspace = true } futures = { workspace = true } serde = { workspace = true, features = ["derive"] } +tokio = { workspace = true, features = ["rt", "sync"] } [target.'cfg(target_os = "linux")'.dependencies] rustix = { workspace = true, features = ["fs"] } diff --git a/codex-rs/file-system/src/local_logs.rs b/codex-rs/file-system/src/local_logs.rs index 9f9b8d10fc..74050202f0 100644 --- a/codex-rs/file-system/src/local_logs.rs +++ b/codex-rs/file-system/src/local_logs.rs @@ -1,59 +1,86 @@ -//! Creates the configured diagnostic directory and retains a shared directory lock on Unix. +//! Defers diagnostic directory creation and the Unix cleanup lock until initialization. //! The caller selects the path. This module does not search for storage. use std::io; use std::path::Path; use std::path::PathBuf; +use std::sync::Arc; +use tokio::sync::OnceCell; -/// A diagnostic directory with a Unix cleanup lock retained by all clones. +/// A diagnostic directory whose clones share deferred setup and a Unix cleanup lock. #[derive(Clone, Debug)] #[must_use = "keep the guard until all log connections close"] pub struct LocalLogsDir { path: PathBuf, + initialized: Arc>, +} + +#[derive(Debug)] +struct LocalLogsDirLock { #[cfg(unix)] - _cleanup_lock: std::sync::Arc, + _cleanup_lock: std::fs::File, } impl LocalLogsDir { - /// Create the selected directory. Keep the guard until all log connections close. + /// Select a directory without accessing the filesystem. + pub fn new(path: PathBuf) -> Self { + Self { + path, + initialized: Arc::new(OnceCell::new()), + } + } + + /// Create the directory and acquire its cleanup lock once. Failed attempts remain retryable. + /// Keep a clone alive until all log connections close. /// On Unix, newly created directories start with mode 0700. Existing permissions /// are preserved, including on user-created directories or after the user changes /// the mode. We do not enforce or restore private permissions on later calls. - pub fn new(path: &Path) -> io::Result { - let mut builder = std::fs::DirBuilder::new(); - builder.recursive(true); - #[cfg(unix)] - { - use std::os::unix::fs::DirBuilderExt; + pub async fn initialize(&self) -> io::Result<()> { + let _ = self + .initialized + .get_or_try_init(|| async { + let path = self.path.clone(); + tokio::task::spawn_blocking(move || { + let mut builder = std::fs::DirBuilder::new(); + builder.recursive(true); + #[cfg(unix)] + { + use std::os::unix::fs::DirBuilderExt; - // Set private permissions only at creation; preserve user-chosen modes afterward. - builder.mode(0o700); - } - builder.create(path)?; - // the main reason to use the diagnostic is to redirect the logs to the tmp/ or /dev/shm (tmpfs) - // so on unix system we have to ensure that the directory is not going to be cleaned via shared lock - #[cfg(unix)] - let directory = { - use std::os::unix::fs::MetadataExt; + // Set private permissions only at creation; preserve existing permissions. + builder.mode(0o700); + } + builder.create(&path)?; + // Protect temporary log directories from cleanup while connections use them. + #[cfg(unix)] + let directory = { + use std::os::unix::fs::MetadataExt; - let directory = std::fs::File::open(path)?; - directory.try_lock_shared()?; - let locked = directory.metadata()?; - let current = std::fs::metadata(path)?; - if !locked.is_dir() || (current.dev(), current.ino()) != (locked.dev(), locked.ino()) { - return Err(io::Error::new( - io::ErrorKind::NotFound, - "log directory changed during initialization", - )); - } - std::sync::Arc::new(directory) - }; + let directory = std::fs::File::open(&path)?; + directory.try_lock_shared()?; + let locked = directory.metadata()?; + let current = std::fs::metadata(&path)?; + if !locked.is_dir() + || (current.dev(), current.ino()) != (locked.dev(), locked.ino()) + { + return Err(io::Error::new( + io::ErrorKind::NotFound, + "log directory changed during initialization", + )); + } + directory + }; - Ok(Self { - path: path.to_path_buf(), - #[cfg(unix)] - _cleanup_lock: directory, - }) + Ok::<_, io::Error>(LocalLogsDirLock { + #[cfg(unix)] + _cleanup_lock: directory, + }) + }) + .await + .map_err(io::Error::other)? + }) + .await?; + Ok(()) } pub fn path(&self) -> &Path { diff --git a/codex-rs/state/src/runtime/logs.rs b/codex-rs/state/src/runtime/logs.rs index 9b331b491a..cf35b70162 100644 --- a/codex-rs/state/src/runtime/logs.rs +++ b/codex-rs/state/src/runtime/logs.rs @@ -304,6 +304,33 @@ WHERE id IN ( pub async fn query_feedback_logs_for_threads( &self, thread_ids: &[&str], + ) -> anyhow::Result> { + Self::query_feedback_logs_from_pool(self.logs_pool.as_ref(), thread_ids).await + } + + /// Read logs from the previous directory without creating or migrating its database. + pub async fn query_legacy_feedback_logs_for_threads( + &self, + thread_ids: &[&str], + ) -> anyhow::Result> { + let Some(path) = self.sqlite.legacy_logs_db_path() else { + return Ok(Vec::new()); + }; + if thread_ids.is_empty() || !tokio::fs::try_exists(&path).await? { + return Ok(Vec::new()); + } + let pool = self + .sqlite + .open_read_only_pool(&path, /*busy_timeout*/ None) + .await?; + let result = Self::query_feedback_logs_from_pool(&pool, thread_ids).await; + pool.close().await; + result + } + + async fn query_feedback_logs_from_pool( + pool: &SqlitePool, + thread_ids: &[&str], ) -> anyhow::Result> { if thread_ids.is_empty() { return Ok(Vec::new()); @@ -376,7 +403,7 @@ WHERE cumulative_estimated_bytes <= builder.push(" ORDER BY ts DESC, ts_nanos DESC, id DESC"); let rows = builder .build_query_as::() - .fetch_all(self.logs_pool.as_ref()) + .fetch_all(pool) .await?; let mut lines = Vec::new(); diff --git a/codex-rs/state/src/runtime/logs_maintenance.rs b/codex-rs/state/src/runtime/logs_maintenance.rs index 7d7879730c..669e40738b 100644 --- a/codex-rs/state/src/runtime/logs_maintenance.rs +++ b/codex-rs/state/src/runtime/logs_maintenance.rs @@ -1,4 +1,4 @@ -//! Prune diagnostic logs in the background immediately, then on a timer. +//! Prune current and legacy diagnostic logs immediately, then on a timer. //! SQLite keeps freed pages for reuse. use super::StateRuntime; @@ -29,25 +29,40 @@ impl StateRuntime { if runtime.logs_pool.is_closed() { break; } - if let Err(err) = prune_by_age_and_size( - &runtime.logs_pool, - Utc::now().timestamp(), - LOG_DATABASE_BUDGET_BYTES, - ) - .await + // Older versions write logs in sqlite_home. Keep the same retention policy + // there until those logs expire. Do not create a missing legacy database. + let legacy_pool = match runtime.sqlite.open_legacy_logs_for_cleanup().await { + Ok(pool) => pool, + Err(err) => { + tracing::warn!("failed to open legacy diagnostic logs for cleanup: {err}"); + None + } + }; + for pool in std::iter::once(runtime.logs_pool.as_ref()).chain(legacy_pool.as_ref()) { - tracing::warn!("failed to prune diagnostic logs: {err}"); - } - if first_sweep { - first_sweep = false; - // Preserve the startup checkpoint without waiting for readers or writers. - if let Err(err) = sqlx::query("PRAGMA wal_checkpoint(PASSIVE)") - .execute(runtime.logs_pool.as_ref()) - .await + if let Err(err) = prune_by_age_and_size( + pool, + Utc::now().timestamp(), + LOG_DATABASE_BUDGET_BYTES, + ) + .await { - tracing::warn!("failed to checkpoint diagnostic logs: {err}"); + tracing::warn!("failed to prune diagnostic logs: {err}"); + } + if first_sweep { + // Preserve the startup checkpoint without waiting for readers or writers. + if let Err(err) = sqlx::query("PRAGMA wal_checkpoint(PASSIVE)") + .execute(pool) + .await + { + tracing::warn!("failed to checkpoint diagnostic logs: {err}"); + } } } + if let Some(pool) = legacy_pool { + pool.close().await; + } + first_sweep = false; } }); } diff --git a/codex-rs/state/src/runtime/logs_maintenance_tests.rs b/codex-rs/state/src/runtime/logs_maintenance_tests.rs index 1ca3939b1a..442ddc2f24 100644 --- a/codex-rs/state/src/runtime/logs_maintenance_tests.rs +++ b/codex-rs/state/src/runtime/logs_maintenance_tests.rs @@ -95,72 +95,191 @@ async fn stops_at_one_second_when_recent_rows_exceed_budget() { #[tokio::test] async fn periodic_cleanup_prunes_without_new_log_writes_or_restart() { let directory = super::super::test_support::unique_temp_dir(); + let sqlite = SqliteConfig::new_for_testing(directory.as_path().abs()); let runtime = StateRuntime::init( - SqliteConfig::new_for_testing(directory.as_path().abs()), + sqlite.clone().with_local_logs_home(directory.join("log")), "test-provider".to_string(), ) .await .unwrap(); - let now = Utc::now().timestamp(); - for timestamp in [now - 11 * DAY, now] { - sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target) VALUES (?, 0, 'INFO', 'test')") - .bind(timestamp) - .execute(runtime.logs_pool.as_ref()) - .await - .unwrap(); - } + // An older process can create the legacy database after the new runtime starts. + let legacy_pool = sqlite + .open_logs_db(&LOGS_MIGRATOR, /*telemetry_override*/ None) + .await + .unwrap(); runtime.start_periodic_logs_maintenance(Duration::from_millis(10)); - tokio::time::timeout(Duration::from_secs(5), async { - loop { - let timestamps: Vec = sqlx::query_scalar("SELECT ts FROM logs ORDER BY ts") - .fetch_all(runtime.logs_pool.as_ref()) + let now = Utc::now().timestamp(); + for round in 1..=2 { + for pool in [runtime.logs_pool.as_ref(), &legacy_pool] { + for timestamp in [now - 11 * DAY, now] { + sqlx::query( + "INSERT INTO logs (ts, ts_nanos, level, target) VALUES (?, 0, 'INFO', 'test')", + ) + .bind(timestamp) + .execute(pool) .await .unwrap(); - if timestamps == vec![now] { - break; } - tokio::time::sleep(Duration::from_millis(10)).await; } - }) - .await - .expect("periodic cleanup should remove expired logs while idle"); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let current: Vec = sqlx::query_scalar("SELECT ts FROM logs ORDER BY ts") + .fetch_all(runtime.logs_pool.as_ref()) + .await + .unwrap(); + let legacy: Vec = sqlx::query_scalar("SELECT ts FROM logs ORDER BY ts") + .fetch_all(&legacy_pool) + .await + .unwrap(); + if (current, legacy) == (vec![now; round], vec![now; round]) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("periodic cleanup should remove expired logs from both locations while idle"); + } + legacy_pool.close().await; runtime.close().await; tokio::fs::remove_dir_all(directory).await.unwrap(); } #[tokio::test] -async fn startup_cleanup_runs_without_waiting_for_the_period() { +async fn legacy_cleanup_skips_missing_and_shared_databases() -> anyhow::Result<()> { let directory = super::super::test_support::unique_temp_dir(); - tokio::fs::create_dir_all(&directory).await.unwrap(); + tokio::fs::create_dir_all(&directory).await?; let sqlite = SqliteConfig::new_for_testing(directory.as_path().abs()); + let legacy_path = sqlite.logs_db_path(); + for logs_home in [directory.clone(), directory.join("log")] { + let sqlite = sqlite.clone().with_local_logs_home(logs_home); + assert!(sqlite.open_legacy_logs_for_cleanup().await?.is_none()); + assert!(!legacy_path.exists()); + } let pool = sqlite - .open_read_write_pool(&sqlite.logs_db_path()) - .await - .unwrap(); - LOGS_MIGRATOR.run(&pool).await.unwrap(); - sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target) VALUES (0, 0, 'INFO', 'test')") - .execute(&pool) - .await - .unwrap(); + .open_logs_db(&LOGS_MIGRATOR, /*telemetry_override*/ None) + .await?; + assert!(sqlite.open_legacy_logs_for_cleanup().await?.is_none()); pool.close().await; + tokio::fs::remove_dir_all(directory).await?; + Ok(()) +} - let runtime = StateRuntime::init(sqlite, "test-provider".to_string()) - .await - .unwrap(); +#[tokio::test] +async fn legacy_writer_does_not_block_current_cleanup_and_is_retried() -> anyhow::Result<()> { + let directory = super::super::test_support::unique_temp_dir(); + tokio::fs::create_dir_all(&directory).await?; + let sqlite = SqliteConfig::new_for_testing(directory.as_path().abs()); + let legacy_pool = sqlite + .open_logs_db(&LOGS_MIGRATOR, /*telemetry_override*/ None) + .await?; + sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target) VALUES (0, 0, 'INFO', 'test')") + .execute(&legacy_pool) + .await?; + let writer = legacy_pool.begin_with("BEGIN IMMEDIATE").await?; + let runtime = StateRuntime::init( + sqlite.with_local_logs_home(directory.join("log")), + "test-provider".to_string(), + ) + .await?; + sqlx::query("INSERT INTO logs (ts, ts_nanos, level, target) VALUES (0, 0, 'INFO', 'test')") + .execute(runtime.logs_pool.as_ref()) + .await?; + runtime.start_periodic_logs_maintenance(Duration::from_millis(10)); tokio::time::timeout(Duration::from_secs(5), async { - loop { - let count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM logs") - .fetch_one(runtime.logs_pool.as_ref()) - .await - .unwrap(); - if count == 0 { - break; - } + while sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM logs") + .fetch_one(runtime.logs_pool.as_ref()) + .await + .unwrap() + != 0 + { tokio::time::sleep(Duration::from_millis(10)).await; } }) - .await - .expect("startup cleanup should run before the 30-minute interval"); + .await?; + assert_eq!( + sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM logs") + .fetch_one(&legacy_pool) + .await?, + 1 + ); + writer.rollback().await?; + tokio::time::timeout(Duration::from_secs(5), async { + while sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM logs") + .fetch_one(&legacy_pool) + .await + .unwrap() + != 0 + { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await?; + legacy_pool.close().await; runtime.close().await; - tokio::fs::remove_dir_all(directory).await.unwrap(); + tokio::fs::remove_dir_all(directory).await?; + Ok(()) +} + +#[tokio::test] +async fn startup_cleanup_runs_without_waiting_for_the_period() { + for separate_logs in [false, true] { + let directory = super::super::test_support::unique_temp_dir(); + tokio::fs::create_dir_all(&directory).await.unwrap(); + let sqlite = SqliteConfig::new_for_testing(directory.as_path().abs()); + let legacy_path = sqlite.logs_db_path(); + let legacy_pool = sqlite + .open_logs_db(&LOGS_MIGRATOR, /*telemetry_override*/ None) + .await + .unwrap(); + let sqlite = if separate_logs { + sqlite.with_local_logs_home(directory.join("log")) + } else { + sqlite + }; + let pool = sqlite + .open_logs_db(&LOGS_MIGRATOR, /*telemetry_override*/ None) + .await + .unwrap(); + for pool in [&pool, &legacy_pool] { + sqlx::query( + "INSERT INTO logs (ts, ts_nanos, level, target) VALUES (0, 0, 'INFO', 'test')", + ) + .execute(pool) + .await + .unwrap(); + } + pool.close().await; + legacy_pool.close().await; + + let runtime = StateRuntime::init(sqlite, "test-provider".to_string()) + .await + .unwrap(); + let legacy_pool = runtime + .sqlite + .open_read_only_pool(&legacy_path, /*busy_timeout*/ None) + .await + .unwrap(); + tokio::time::timeout(Duration::from_secs(5), async { + loop { + let current: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM logs") + .fetch_one(runtime.logs_pool.as_ref()) + .await + .unwrap(); + let legacy: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM logs") + .fetch_one(&legacy_pool) + .await + .unwrap(); + if (current, legacy) == (0, 0) { + break; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("startup cleanup should prune both locations before the 30-minute interval"); + legacy_pool.close().await; + runtime.close().await; + tokio::fs::remove_dir_all(directory).await.unwrap(); + } } diff --git a/codex-rs/state/src/runtime/reclamation.rs b/codex-rs/state/src/runtime/reclamation.rs index 952aead663..36201e16e9 100644 --- a/codex-rs/state/src/runtime/reclamation.rs +++ b/codex-rs/state/src/runtime/reclamation.rs @@ -1,4 +1,5 @@ //! Reclaims unused SQLite pages in opted-in databases, reducing batches after stalled passes. +//! Current and legacy log directories each retain their own maintenance lock. //! NFS on Linux skips opening a maintenance connection. use crate::RuntimeDbPath; use crate::SqliteConfig; @@ -8,8 +9,10 @@ use sqlx::Connection; use sqlx::SqliteConnection; use sqlx::sqlite::SqliteConnectOptions; use sqlx::sqlite::SqliteSynchronous; +use std::collections::BTreeMap; use std::fs::File; use std::path::Path; +use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; use std::time::Instant; @@ -42,30 +45,18 @@ impl SqliteReclamationWorker { // Declared first so completion is signaled after all worker resources drop. let _finish = finish; let mut scheduled = sqlite - .runtime_db_paths() + .reclamation_db_paths() .into_iter() - .filter_map(|db| { - if db.background_reclamation { - Some((db, Instant::now(), ReclamationState::default())) - } else { - None - } - }) + .map(|db| (db, Instant::now(), ReclamationState::default())) .collect::>(); - let mut owner = None; + let mut owners = BTreeMap::new(); let mut delay = IDLE_INTERVAL; loop { tokio::select! { _ = receiver.changed() => break, _ = tokio::time::sleep(delay) => {} } - if owner.is_none() { - owner = try_ownership(sqlite.home()).await.ok().flatten(); - } - if owner.is_none() { - continue; - } - delay = visit(&sqlite, &mut scheduled, &receiver).await; + delay = visit(&sqlite, &mut scheduled, &mut owners, &receiver).await; } }); @@ -82,6 +73,7 @@ impl SqliteReclamationWorker { async fn visit( sqlite: &SqliteConfig, scheduled: &mut [(RuntimeDbPath, Instant, ReclamationState)], + owners: &mut BTreeMap, shutdown: &watch::Receiver<()>, ) -> Duration { for (db, due, state) in &mut *scheduled { @@ -92,6 +84,18 @@ async fn visit( if *due > started { continue; } + let Some(home) = db.path.parent() else { + continue; + }; + if let std::collections::btree_map::Entry::Vacant(entry) = owners.entry(home.to_path_buf()) + { + if let Ok(Some(owner)) = try_ownership(home).await { + entry.insert(owner); + } else { + *due = started + IDLE_INTERVAL; + continue; + } + } let result = reclaim( sqlite, &db.path, @@ -214,8 +218,14 @@ async fn reclaim( options: impl Into, shutdown: &watch::Receiver<()>, ) -> anyhow::Result { - // do not perform autovacuum and cleanup on NFS as it creates a lot of load and contention on IO - if sqlite.is_nfs().await? { + // Check the selected logs directory. An administrator can also select NFS. + let logs = path == sqlite.logs_db_path(); + let is_nfs = if logs { + sqlite.logs_are_on_nfs().await? + } else { + sqlite.is_nfs().await? + }; + if is_nfs { return Ok(ReclamationPass { pages: 0, outcome: PassOutcome::Idle, diff --git a/codex-rs/state/src/runtime/reclamation_tests.rs b/codex-rs/state/src/runtime/reclamation_tests.rs index fb15e3602d..d4e0fb4de7 100644 --- a/codex-rs/state/src/runtime/reclamation_tests.rs +++ b/codex-rs/state/src/runtime/reclamation_tests.rs @@ -4,6 +4,88 @@ use crate::runtime::test_support::unique_temp_dir; use codex_utils_absolute_path::test_support::PathExt; use pretty_assertions::assert_eq; +#[tokio::test] +async fn reclaims_current_and_legacy_logs_with_independent_directory_locks() -> anyhow::Result<()> { + let directory = unique_temp_dir(); + tokio::fs::create_dir_all(&directory).await?; + let legacy = SqliteConfig::new_for_testing(directory.as_path().abs()); + let sqlite = legacy.clone().with_local_logs_home(directory.join("log")); + let mut pools = Vec::new(); + let mut sizes = Vec::new(); + for config in [&sqlite, &legacy] { + let pool = config + .open_logs_db(&runtime_logs_migrator(), /*telemetry_override*/ None) + .await?; + sqlx::query( + "WITH RECURSIVE n(i) AS (VALUES(1) UNION ALL SELECT i+1 FROM n WHERE i < 4096) \ + INSERT INTO logs(ts,ts_nanos,level,target,feedback_log_body) \ + SELECT unixepoch(),0,'INFO','fixture',printf('%032768d',i) FROM n", + ) + .execute(&pool) + .await?; + sqlx::query("DELETE FROM logs WHERE id%4 != 0") + .execute(&pool) + .await?; + sqlx::query("PRAGMA wal_checkpoint(TRUNCATE)") + .execute(&pool) + .await?; + sizes.push(tokio::fs::metadata(config.logs_db_path()).await?.len()); + pools.push(pool); + } + let rows = "SELECT id, feedback_log_body FROM logs ORDER BY id"; + let expected: Vec<(i64, String)> = sqlx::query_as(rows).fetch_all(&pools[0]).await?; + let mut scheduled = sqlite + .reclamation_db_paths() + .into_iter() + .map(|db| (db, Instant::now(), ReclamationState::default())) + .collect::>(); + let mut owners = BTreeMap::new(); + let (_shutdown, receiver) = watch::channel(()); + + // A legacy process owns its directory. Current logs must still be reclaimed. + let legacy_owner = try_ownership(legacy.home()).await?.unwrap(); + for (index, config) in [&sqlite, &legacy].into_iter().enumerate() { + tokio::time::timeout(Duration::from_secs(10), async { + loop { + for (_, due, _) in &mut scheduled { + *due = Instant::now(); + } + visit(&sqlite, &mut scheduled, &mut owners, &receiver).await; + if config.logs_are_on_nfs().await.unwrap() + || tokio::fs::metadata(config.logs_db_path()) + .await + .unwrap() + .len() + < sizes[index] + { + break; + } + } + }) + .await?; + if index == 0 { + assert_eq!( + tokio::fs::metadata(legacy.logs_db_path()).await?.len(), + sizes[1] + ); + legacy_owner.unlock()?; + } + } + for pool in pools { + assert_eq!( + sqlx::query_as::<_, (i64, String)>(rows) + .fetch_all(&pool) + .await?, + expected + ); + pool.close().await; + } + drop(owners); + drop(legacy_owner); + tokio::fs::remove_dir_all(directory).await?; + Ok(()) +} + #[tokio::test] async fn interrupted_reclamation_releases_writer_and_resumes_without_data_loss() -> anyhow::Result<()> { diff --git a/codex-rs/state/src/sqlite.rs b/codex-rs/state/src/sqlite.rs index 5ac878f35d..18b0a2b72d 100644 --- a/codex-rs/state/src/sqlite.rs +++ b/codex-rs/state/src/sqlite.rs @@ -1,6 +1,7 @@ //! Shared SQLite connection configuration. //! Writable pools preserve the journal and vacuum modes of initialized databases. //! Startup retries SQLITE_BUSY and reports other errors to callers. +//! The configured logs directory is created and locked only when opening its database. #![expect( clippy::disallowed_methods, @@ -12,6 +13,7 @@ use crate::migrations::repair_legacy_recency_migration_version; use crate::runtime::recovery::RuntimeDbInitError; use crate::telemetry; use crate::telemetry::DbKind; +use codex_file_system::LocalLogsDir; use codex_utils_absolute_path::AbsolutePathBuf; use log::LevelFilter; use sqlx::ConnectOptions; @@ -65,8 +67,13 @@ struct RuntimeDbSpec { } impl RuntimeDbSpec { - fn path(self, codex_home: &Path) -> PathBuf { - codex_home.join(self.filename) + fn path(self, sqlite: &SqliteConfig) -> PathBuf { + let home = if matches!(self.kind, DbKind::Logs) { + sqlite.logs_home() + } else { + sqlite.home() + }; + home.join(self.filename) } } @@ -156,19 +163,23 @@ pub struct RuntimeDbPath { } /// Resolved configuration shared by all Codex SQLite connections. -/// Clones share filesystem probes and quick-check attempts; a new config checks independently. +/// Clones share directory locks, filesystem probes, and quick-check attempts. #[derive(Clone, Debug)] pub struct SqliteConfig { sqlite_home: AbsolutePathBuf, + local_logs_dir: Option, is_nfs: Arc>, + logs_is_nfs: Arc>, quick_check_manager: SqliteQuickCheckManager, } -// Cached filesystem information and validation history are operational state, +// Directory locks, cached filesystem information, and validation history are operational state, // not part of configuration equality. impl PartialEq for SqliteConfig { fn eq(&self, other: &Self) -> bool { self.sqlite_home == other.sqlite_home + && self.local_logs_dir.as_ref().map(LocalLogsDir::path) + == other.local_logs_dir.as_ref().map(LocalLogsDir::path) } } @@ -178,7 +189,9 @@ impl SqliteConfig { pub fn from_sqlite_home(sqlite_home: AbsolutePathBuf) -> Self { Self { sqlite_home, + local_logs_dir: None, is_nfs: Arc::new(OnceCell::new()), + logs_is_nfs: Arc::new(OnceCell::new()), quick_check_manager: SqliteQuickCheckManager::new(), } } @@ -191,42 +204,55 @@ impl SqliteConfig { self.sqlite_home.as_path() } + /// Select the logs directory without accessing it until its database is opened. + pub fn with_local_logs_home(mut self, home: PathBuf) -> Self { + self.local_logs_dir = Some(LocalLogsDir::new(home)); + self.logs_is_nfs = Arc::new(OnceCell::new()); + self + } + + pub(crate) fn logs_home(&self) -> &Path { + self.local_logs_dir + .as_ref() + .map_or(self.home(), LocalLogsDir::path) + } + /// Detects if the sqlite_home directory is under the NFS mount (once per the lifetime) pub async fn is_nfs(&self) -> anyhow::Result { - self.is_nfs - .get_or_try_init(|| async { - let sqlite_home = self.home().to_path_buf(); - Ok( - tokio::task::spawn_blocking(move || codex_file_system::is_nfs(&sqlite_home)) - .await??, - ) - }) - .await - .copied() + probe_nfs(self.home(), &self.is_nfs).await + } + + pub(crate) async fn logs_are_on_nfs(&self) -> anyhow::Result { + probe_nfs(self.logs_home(), &self.logs_is_nfs).await } /// Return the path to the primary state database. pub fn state_db_path(&self) -> PathBuf { - STATE_DB.path(self.home()) + STATE_DB.path(self) } /// Return the path to the logs database. pub fn logs_db_path(&self) -> PathBuf { - LOGS_DB.path(self.home()) + LOGS_DB.path(self) + } + + pub(crate) fn legacy_logs_db_path(&self) -> Option { + let path = self.home().join(LOGS_DB_FILENAME); + (path != self.logs_db_path()).then_some(path) } /// Return the path to the goals database. pub fn goals_db_path(&self) -> PathBuf { - GOALS_DB.path(self.home()) + GOALS_DB.path(self) } /// Return the path to the memories database. pub fn memories_db_path(&self) -> PathBuf { - MEMORIES_DB.path(self.home()) + MEMORIES_DB.path(self) } pub(crate) fn memories_v2_db_path(&self) -> PathBuf { - MEMORIES_V2_DB.path(self.home()) + MEMORIES_V2_DB.path(self) } pub(crate) async fn open_memories_v2_db(&self) -> anyhow::Result { @@ -240,12 +266,12 @@ impl SqliteConfig { /// Return the path to the durable user-message queue database. pub fn queue_db_path(&self) -> PathBuf { - QUEUE_DB.path(self.home()) + QUEUE_DB.path(self) } /// Return the path to the paginated thread-history database. pub fn thread_history_db_path(&self) -> PathBuf { - THREAD_HISTORY_DB.path(self.home()) + THREAD_HISTORY_DB.path(self) } /// Return the paths to every database managed by the state runtime. @@ -254,12 +280,24 @@ impl SqliteConfig { .iter() .map(|spec| RuntimeDbPath { label: spec.label, - path: spec.path(self.home()), + path: spec.path(self), background_reclamation: spec.background_reclamation, }) .collect() } + pub(crate) fn reclamation_db_paths(&self) -> Vec { + self.runtime_db_paths() + .into_iter() + .filter(|db| db.background_reclamation) + .chain(self.legacy_logs_db_path().map(|path| RuntimeDbPath { + label: "legacy log DB", + path, + background_reclamation: true, + })) + .collect() + } + pub(super) async fn open_state_db( &self, migrator: &Migrator, @@ -323,7 +361,7 @@ impl SqliteConfig { migrator: &Migrator, telemetry_override: Option<&dyn DbTelemetry>, ) -> anyhow::Result { - let path = spec.path(self.home()); + let path = spec.path(self); let started = Instant::now(); let pool_result = self .open_read_write_pool_with_spec(path.as_path(), Some(spec), telemetry_override) @@ -363,6 +401,7 @@ impl SqliteConfig { } /// Open a writable Codex SQLite database, creating it if necessary. + /// Keep this configuration alive to retain the cleanup lock for local logs. pub async fn open_read_write_pool(&self, path: &Path) -> anyhow::Result { self.open_read_write_pool_with_spec( path, /*recover_spec*/ None, /*telemetry_override*/ None, @@ -376,6 +415,7 @@ impl SqliteConfig { recover_spec: Option, telemetry_override: Option<&dyn DbTelemetry>, ) -> anyhow::Result { + self.prepare_logs_directory(path).await?; let recovery = recover_spec.map_or(RecoveryMode::Unavailable, |spec| spec.recovery); let options = SqliteConnectOptions::new() .filename(path) @@ -423,11 +463,13 @@ impl SqliteConfig { } /// Open an existing Codex SQLite database without creating or modifying it. + /// Keep this configuration alive to retain the cleanup lock for local logs. pub async fn open_read_only_pool( &self, path: &Path, busy_timeout: Option, ) -> Result { + self.prepare_logs_directory(path).await.map_err(Error::Io)?; let mut options = SqliteConnectOptions::new() .filename(path) .create_if_missing(false) @@ -441,6 +483,39 @@ impl SqliteConfig { .connect_with(options) .await } + + /// Open old logs for cleanup without creating, migrating, or rebuilding the database. + pub(crate) async fn open_legacy_logs_for_cleanup(&self) -> anyhow::Result> { + let Some(path) = self.legacy_logs_db_path() else { + return Ok(None); + }; + if !tokio::fs::try_exists(&path).await? { + return Ok(None); + } + let options = SqliteConnectOptions::new() + .filename(path) + .create_if_missing(false) + .busy_timeout(Duration::ZERO) + .log_statements(LevelFilter::Off); + SqlitePoolOptions::new() + .max_connections(1) + .connect_with(options) + .await + .map(Some) + .map_err(anyhow::Error::from) + } + + async fn prepare_logs_directory(&self, path: &Path) -> std::io::Result<()> { + // Login, login status, and logout load configuration without opening SQLite. + // Create and lock the logs directory only when opening its database. + // An unavailable log_dir must not prevent those commands from loading configuration. + if let Some(directory) = &self.local_logs_dir + && path == self.logs_db_path() + { + directory.initialize().await?; + } + Ok(()) + } } async fn connect_once(options: &SqliteConnectOptions) -> Result { @@ -527,6 +602,16 @@ async fn connect_once(options: &SqliteConnectOptions) -> Result) -> anyhow::Result { + probe + .get_or_try_init(|| async { + let home = home.to_path_buf(); + Ok(tokio::task::spawn_blocking(move || codex_file_system::is_nfs(&home)).await??) + }) + .await + .copied() +} + #[cfg(test)] #[path = "sqlite_tests.rs"] mod tests; diff --git a/codex-rs/state/src/sqlite/validation_tests.rs b/codex-rs/state/src/sqlite/validation_tests.rs index f3d3853c8a..38cb918068 100644 --- a/codex-rs/state/src/sqlite/validation_tests.rs +++ b/codex-rs/state/src/sqlite/validation_tests.rs @@ -54,7 +54,7 @@ async fn runtime_opens_recover_or_report_corruption_by_database_policy() -> anyh let home = root.join(spec.label); tokio::fs::create_dir_all(&home).await?; let sqlite = SqliteConfig::new_for_testing(home.as_path().abs()); - let path = spec.path(&home); + let path = spec.path(&sqlite); let fixture = path.with_extension("fixture"); let pool = sqlx::SqlitePool::connect_with( sqlx::sqlite::SqliteConnectOptions::new() diff --git a/codex-rs/state/src/sqlite_tests.rs b/codex-rs/state/src/sqlite_tests.rs index 19bc89432f..b837bbbaac 100644 --- a/codex-rs/state/src/sqlite_tests.rs +++ b/codex-rs/state/src/sqlite_tests.rs @@ -8,6 +8,89 @@ use sqlx::sqlite::SqliteConnectOptions; use sqlx::sqlite::SqliteJournalMode; use sqlx::sqlite::SqlitePoolOptions; +#[tokio::test] +async fn configured_log_directory_is_initialized_on_use_and_retries_after_failure() +-> anyhow::Result<()> { + let sqlite_home = crate::runtime::test_support::unique_temp_dir(); + tokio::fs::create_dir_all(&sqlite_home).await?; + let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| { + let _ = std::fs::remove_dir_all(sqlite_home); + }); + let logs_home = sqlite_home.join("logs"); + tokio::fs::write(&logs_home, "not a directory").await?; + let sqlite = SqliteConfig::new_for_testing(sqlite_home.as_path().abs()); + let fallback_logs_path = sqlite.logs_db_path(); + let sqlite = sqlite.with_local_logs_home(logs_home.clone()); + + // Opening unrelated state must not require the configured logs directory. + let pool = sqlite.open_read_write_pool(&sqlite.state_db_path()).await?; + pool.close().await; + assert!( + crate::StateRuntime::init(sqlite.clone(), "openai".to_string()) + .await + .is_err() + ); + assert!(!fallback_logs_path.exists()); + + // A failed initialization must remain retryable without replacing the config. + tokio::fs::remove_file(&logs_home).await?; + let runtime = crate::StateRuntime::init(sqlite.clone(), "openai".to_string()).await?; + assert!(sqlite.logs_db_path().is_file()); + assert!(!fallback_logs_path.exists()); + runtime.close().await; + Ok(()) +} + +#[cfg(unix)] +#[tokio::test] +async fn configured_log_directory_lock_is_shared_by_reader_and_writer_config_clones() +-> anyhow::Result<()> { + let sqlite_home = crate::runtime::test_support::unique_temp_dir(); + let logs_home = sqlite_home.join("logs"); + tokio::fs::create_dir_all(&logs_home).await?; + let _cleanup = scopeguard::guard(sqlite_home.clone(), |sqlite_home| { + let _ = std::fs::remove_dir_all(sqlite_home); + }); + let initial = SqliteConfig::new_for_testing(logs_home.as_path().abs()); + let logs_path = initial.logs_db_path(); + initial + .open_read_write_pool(&logs_path) + .await? + .close() + .await; + + for read_only in [false, true] { + let sqlite = SqliteConfig::new_for_testing(sqlite_home.as_path().abs()) + .with_local_logs_home(logs_home.clone()); + let clone = sqlite.clone(); + let cleanup_lock = std::fs::File::open(&logs_home)?; + cleanup_lock.try_lock()?; + cleanup_lock.unlock()?; + + let pool = if read_only { + sqlite + .open_read_only_pool(&logs_path, /*busy_timeout*/ None) + .await? + } else { + sqlite.open_read_write_pool(&logs_path).await? + }; + assert!(matches!( + cleanup_lock.try_lock(), + Err(std::fs::TryLockError::WouldBlock) + )); + drop(sqlite); + assert!(matches!( + cleanup_lock.try_lock(), + Err(std::fs::TryLockError::WouldBlock) + )); + + pool.close().await; + drop(clone); + cleanup_lock.try_lock()?; + } + Ok(()) +} + #[cfg(target_os = "linux")] #[tokio::test] async fn sqlite_home_filesystem_probe_is_shared_by_clones_until_config_replacement()