diff --git a/codex-rs/app-server/src/codex_message_processor.rs b/codex-rs/app-server/src/codex_message_processor.rs index 3cf48699b..513837c7c 100644 --- a/codex-rs/app-server/src/codex_message_processor.rs +++ b/codex-rs/app-server/src/codex_message_processor.rs @@ -3069,41 +3069,44 @@ impl CodexMessageProcessor { } }; - if let Err(err) = state_db.reset_memory_data_for_fresh_start().await { + if let Err(err) = state_db.clear_memory_data().await { self.send_internal_error( request_id, - format!("failed to reset memory rows in state db: {err}"), + format!("failed to clear memory rows in state db: {err}"), ) .await; return; } let memory_root = self.config.codex_home.join("memories"); + let memory_extensions_root = self.config.codex_home.join("memories_extensions"); let clear_memory_root_result: std::io::Result<()> = async { - match tokio::fs::symlink_metadata(&memory_root).await { - Ok(metadata) if metadata.file_type().is_symlink() => { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - format!( - "refusing to clear symlinked memory root {}", - memory_root.display() - ), - )); + for directory in [memory_root.as_path(), memory_extensions_root.as_path()] { + match tokio::fs::symlink_metadata(directory).await { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!( + "refusing to clear symlinked memory root {}", + directory.display() + ), + )); + } + Ok(_) => {} + Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} + Err(err) => return Err(err), } - Ok(_) => {} - Err(err) if err.kind() == std::io::ErrorKind::NotFound => {} - Err(err) => return Err(err), - } - tokio::fs::create_dir_all(&memory_root).await?; - let mut entries = tokio::fs::read_dir(&memory_root).await?; - while let Some(entry) = entries.next_entry().await? { - let path = entry.path(); - let file_type = entry.file_type().await?; - if file_type.is_dir() { - tokio::fs::remove_dir_all(path).await?; - } else { - tokio::fs::remove_file(path).await?; + tokio::fs::create_dir_all(directory).await?; + let mut entries = tokio::fs::read_dir(directory).await?; + while let Some(entry) = entries.next_entry().await? { + let path = entry.path(); + let file_type = entry.file_type().await?; + if file_type.is_dir() { + tokio::fs::remove_dir_all(path).await?; + } else { + tokio::fs::remove_file(path).await?; + } } } diff --git a/codex-rs/app-server/tests/suite/v2/memory_reset.rs b/codex-rs/app-server/tests/suite/v2/memory_reset.rs index 954a4fc6b..3c7ae3867 100644 --- a/codex-rs/app-server/tests/suite/v2/memory_reset.rs +++ b/codex-rs/app-server/tests/suite/v2/memory_reset.rs @@ -20,7 +20,7 @@ use uuid::Uuid; const DEFAULT_READ_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); #[tokio::test] -async fn memory_reset_clears_memory_files_and_state_db_rows() -> Result<()> { +async fn memory_reset_clears_memory_files_and_rows_preserves_threads() -> Result<()> { let codex_home = TempDir::new()?; create_config_toml(codex_home.path())?; let state_db = init_state_db(codex_home.path()).await?; @@ -53,7 +53,7 @@ async fn memory_reset_clears_memory_files_and_state_db_rows() -> Result<()> { assert_eq!(stage1_outputs, Vec::new()); assert_eq!( state_db.get_thread_memory_mode(thread_id).await?.as_deref(), - Some("disabled") + Some("enabled") ); let mut remaining_entries = tokio::fs::read_dir(&memory_root).await?; diff --git a/codex-rs/cli/src/main.rs b/codex-rs/cli/src/main.rs index 5ad86a80f..4e962cf09 100644 --- a/codex-rs/cli/src/main.rs +++ b/codex-rs/cli/src/main.rs @@ -1269,7 +1269,7 @@ async fn run_debug_clear_memories_command( let state_db = StateRuntime::init(config.sqlite_home.clone(), config.model_provider_id.clone()) .await?; - state_db.reset_memory_data_for_fresh_start().await?; + state_db.clear_memory_data().await?; cleared_state_db = true; } diff --git a/codex-rs/cli/tests/debug_clear_memories.rs b/codex-rs/cli/tests/debug_clear_memories.rs index c68172ba3..164e4bd3e 100644 --- a/codex-rs/cli/tests/debug_clear_memories.rs +++ b/codex-rs/cli/tests/debug_clear_memories.rs @@ -125,12 +125,6 @@ INSERT INTO jobs ( .fetch_one(&pool) .await?; assert_eq!(memory_jobs_count, 0); - - let memory_mode: String = sqlx::query_scalar("SELECT memory_mode FROM threads WHERE id = ?") - .bind(thread_id) - .fetch_one(&pool) - .await?; - assert_eq!(memory_mode, "disabled"); assert!(!memory_root.exists()); Ok(()) diff --git a/codex-rs/state/src/runtime/memories.rs b/codex-rs/state/src/runtime/memories.rs index 50f2f0cd8..4817d5db6 100644 --- a/codex-rs/state/src/runtime/memories.rs +++ b/codex-rs/state/src/runtime/memories.rs @@ -30,21 +30,6 @@ impl StateRuntime { /// stage-1 (`memory_stage1`) and phase-2 (`memory_consolidate_global`) /// memory pipelines. pub async fn clear_memory_data(&self) -> anyhow::Result<()> { - self.clear_memory_data_inner(/*disable_existing_threads*/ false) - .await - } - - /// Resets persisted memory state for a clean-slate local start. - /// - /// In addition to clearing persisted stage-1 outputs and memory pipeline - /// jobs, this disables memory generation for all existing threads so - /// historical rollouts are not immediately picked up again. - pub async fn reset_memory_data_for_fresh_start(&self) -> anyhow::Result<()> { - self.clear_memory_data_inner(/*disable_existing_threads*/ true) - .await - } - - async fn clear_memory_data_inner(&self, disable_existing_threads: bool) -> anyhow::Result<()> { let mut tx = self.pool.begin().await?; sqlx::query( @@ -66,18 +51,6 @@ WHERE kind = ? OR kind = ? .execute(&mut *tx) .await?; - if disable_existing_threads { - sqlx::query( - r#" -UPDATE threads -SET memory_mode = 'disabled' -WHERE memory_mode = 'enabled' - "#, - ) - .execute(&mut *tx) - .await?; - } - tx.commit().await?; Ok(()) } @@ -1823,7 +1796,7 @@ mod tests { } #[tokio::test] - async fn reset_memory_data_for_fresh_start_clears_rows_and_disables_threads() { + async fn clear_memory_data_clears_rows_and_preserves_thread_memory_modes() { let codex_home = unique_temp_dir(); let runtime = StateRuntime::init(codex_home.clone(), "test-provider".to_string()) .await @@ -1893,9 +1866,9 @@ mod tests { .expect("disable existing thread"); runtime - .reset_memory_data_for_fresh_start() + .clear_memory_data() .await - .expect("reset memory data"); + .expect("clear memory data"); let stage1_outputs_count: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM stage1_outputs") .fetch_one(runtime.pool.as_ref()) @@ -1918,7 +1891,7 @@ mod tests { .fetch_one(runtime.pool.as_ref()) .await .expect("read enabled thread memory mode"); - assert_eq!(enabled_memory_mode, "disabled"); + assert_eq!(enabled_memory_mode, "enabled"); let disabled_memory_mode: String = sqlx::query_scalar("SELECT memory_mode FROM threads WHERE id = ?")