From b6244f776ded1bde4beea3c6bd0c43d9e40186d4 Mon Sep 17 00:00:00 2001 From: jif-oai Date: Wed, 15 Apr 2026 10:38:11 +0100 Subject: [PATCH] feat: cleaning of memories extension (#17844) --- codex-rs/core/src/memories/README.md | 9 +- codex-rs/core/src/memories/extensions.rs | 251 ++++++++++++++++++++ codex-rs/core/src/memories/mod.rs | 2 + codex-rs/core/src/memories/phase2.rs | 32 ++- codex-rs/core/src/memories/prompts.rs | 42 +++- codex-rs/core/src/memories/prompts_tests.rs | 32 +++ codex-rs/core/src/memories/tests.rs | 29 +++ codex-rs/core/tests/suite/memories.rs | 154 ++++++++++++ 8 files changed, 535 insertions(+), 16 deletions(-) create mode 100644 codex-rs/core/src/memories/extensions.rs diff --git a/codex-rs/core/src/memories/README.md b/codex-rs/core/src/memories/README.md index eecaeaea2..a1d365435 100644 --- a/codex-rs/core/src/memories/README.md +++ b/codex-rs/core/src/memories/README.md @@ -84,7 +84,11 @@ What it does: - `raw_memories.md` (merged raw memories, latest first) - `rollout_summaries/` (one summary file per retained rollout) - prunes stale rollout summaries that are no longer retained -- if there are no inputs, marks the job successful and exits +- finds old resource files from memory extensions under + `memories_extensions//resources/` for extension directories that + have an `instructions.md`, using the memory module retention window +- if there are no Phase 1 inputs or old extension resources, marks the job + successful and exits If there is input, it then: @@ -92,10 +96,13 @@ If there is input, it then: - builds the Phase 2 prompt with a diff of the current Phase 1 input selection versus the last successful Phase 2 selection (`added`, `retained`, `removed`) +- includes old extension resource paths in the prompt diff - runs it with no approvals, no network, and local write access only - disables collab for that agent (to prevent recursive delegation) - watches the agent status and heartbeats the global job lease while it runs - marks the phase-2 job success/failure in the state DB when the agent finishes +- prunes old extension resource files after the consolidation agent completes + and the successful Phase 2 job is recorded Selection diff behavior: diff --git a/codex-rs/core/src/memories/extensions.rs b/codex-rs/core/src/memories/extensions.rs new file mode 100644 index 000000000..91c96ebff --- /dev/null +++ b/codex-rs/core/src/memories/extensions.rs @@ -0,0 +1,251 @@ +use crate::memories::memory_extensions_root; +use chrono::DateTime; +use chrono::Duration; +use chrono::NaiveDateTime; +use chrono::Utc; +use std::path::Path; +use std::path::PathBuf; +use tracing::warn; + +const FILENAME_TS_FORMAT: &str = "%Y-%m-%dT%H-%M-%S"; +pub(super) const EXTENSION_RESOURCE_RETENTION_DAYS: i64 = 7; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct RemovedExtensionResource { + pub(super) extension: String, + pub(super) resource_path: String, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub(super) struct PendingExtensionResourceRemoval { + pub(super) removed: RemovedExtensionResource, + path: PathBuf, +} + +pub(super) async fn find_old_extension_resources( + memory_root: &Path, +) -> Vec { + find_old_extension_resources_with_now(memory_root, Utc::now()).await +} + +async fn find_old_extension_resources_with_now( + memory_root: &Path, + now: DateTime, +) -> Vec { + let mut pending = Vec::new(); + let cutoff = now - Duration::days(EXTENSION_RESOURCE_RETENTION_DAYS); + let extensions_root = memory_extensions_root(memory_root); + let mut extensions = match tokio::fs::read_dir(&extensions_root).await { + Ok(extensions) => extensions, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => return pending, + Err(err) => { + warn!( + "failed reading memory extensions root {}: {err}", + extensions_root.display() + ); + return pending; + } + }; + + while let Ok(Some(extension_entry)) = extensions.next_entry().await { + let extension_path = extension_entry.path(); + let Ok(file_type) = extension_entry.file_type().await else { + continue; + }; + if !file_type.is_dir() { + continue; + } + let Some(extension) = extension_path + .file_name() + .and_then(|name| name.to_str()) + .map(ToOwned::to_owned) + else { + continue; + }; + if !tokio::fs::try_exists(extension_path.join("instructions.md")) + .await + .unwrap_or(false) + { + continue; + } + + let resources_path = extension_path.join("resources"); + let mut resources = match tokio::fs::read_dir(&resources_path).await { + Ok(resources) => resources, + Err(err) if err.kind() == std::io::ErrorKind::NotFound => continue, + Err(err) => { + warn!( + "failed reading memory extension resources {}: {err}", + resources_path.display() + ); + continue; + } + }; + + while let Ok(Some(resource_entry)) = resources.next_entry().await { + let resource_file_path = resource_entry.path(); + let Ok(file_type) = resource_entry.file_type().await else { + continue; + }; + if !file_type.is_file() { + continue; + } + let Some(file_name) = resource_file_path + .file_name() + .and_then(|name| name.to_str()) + else { + continue; + }; + if !file_name.ends_with(".md") { + continue; + } + let Some(resource_timestamp) = resource_timestamp(file_name) else { + continue; + }; + if resource_timestamp > cutoff { + continue; + } + + pending.push(PendingExtensionResourceRemoval { + removed: RemovedExtensionResource { + extension: extension.clone(), + resource_path: format!("resources/{file_name}"), + }, + path: resource_file_path, + }); + } + } + + pending.sort_by(|left, right| { + left.removed + .extension + .cmp(&right.removed.extension) + .then_with(|| left.removed.resource_path.cmp(&right.removed.resource_path)) + }); + pending +} + +pub(super) async fn remove_extension_resources(resources: &[PendingExtensionResourceRemoval]) { + for resource in resources { + if let Err(err) = tokio::fs::remove_file(&resource.path).await + && err.kind() != std::io::ErrorKind::NotFound + { + warn!( + "failed pruning old memory extension resource {}: {err}", + resource.path.display() + ); + } + } +} + +fn resource_timestamp(file_name: &str) -> Option> { + let timestamp = file_name.get(..19)?; + let naive = NaiveDateTime::parse_from_str(timestamp, FILENAME_TS_FORMAT).ok()?; + Some(DateTime::from_naive_utc_and_offset(naive, Utc)) +} + +#[cfg(test)] +mod tests { + use super::*; + use pretty_assertions::assert_eq; + use tempfile::TempDir; + + #[tokio::test] + async fn finds_only_old_resources_from_extensions_with_instructions() { + let codex_home = TempDir::new().expect("create temp codex home"); + let memory_root = codex_home.path().join("memories"); + let extensions_root = memory_extensions_root(&memory_root); + let telepathy_resources = extensions_root.join("telepathy/resources"); + tokio::fs::create_dir_all(&telepathy_resources) + .await + .expect("create telepathy resources"); + tokio::fs::write( + extensions_root.join("telepathy/instructions.md"), + "instructions", + ) + .await + .expect("write telepathy instructions"); + + let now = DateTime::from_naive_utc_and_offset( + NaiveDateTime::parse_from_str("2026-04-14T12-00-00", FILENAME_TS_FORMAT) + .expect("parse now"), + Utc, + ); + let old_file = telepathy_resources.join("2026-04-06T11-59-59-abcd-10min-old.md"); + let exact_cutoff_file = + telepathy_resources.join("2026-04-07T12-00-00-abcd-10min-cutoff.md"); + let recent_file = telepathy_resources.join("2026-04-08T12-00-00-abcd-10min-recent.md"); + let invalid_file = telepathy_resources.join("not-a-timestamp.md"); + for file in [&old_file, &exact_cutoff_file, &recent_file, &invalid_file] { + tokio::fs::write(file, "resource") + .await + .expect("write telepathy resource"); + } + + let ignored_resources = extensions_root.join("ignored/resources"); + tokio::fs::create_dir_all(&ignored_resources) + .await + .expect("create ignored resources"); + let ignored_old_file = ignored_resources.join("2026-04-06T11-59-59-abcd-10min-old.md"); + tokio::fs::write(&ignored_old_file, "ignored") + .await + .expect("write ignored resource"); + + let pending = find_old_extension_resources_with_now(&memory_root, now).await; + + assert_eq!( + pending + .iter() + .map(|resource| resource.removed.clone()) + .collect::>(), + vec![ + RemovedExtensionResource { + extension: "telepathy".to_string(), + resource_path: "resources/2026-04-06T11-59-59-abcd-10min-old.md".to_string(), + }, + RemovedExtensionResource { + extension: "telepathy".to_string(), + resource_path: "resources/2026-04-07T12-00-00-abcd-10min-cutoff.md".to_string(), + }, + ] + ); + assert!( + tokio::fs::try_exists(&old_file) + .await + .expect("check old file before remove") + ); + assert!( + tokio::fs::try_exists(&exact_cutoff_file) + .await + .expect("check cutoff file before remove") + ); + + remove_extension_resources(&pending).await; + + assert!( + !tokio::fs::try_exists(&old_file) + .await + .expect("check old file") + ); + assert!( + !tokio::fs::try_exists(&exact_cutoff_file) + .await + .expect("check cutoff file") + ); + assert!( + tokio::fs::try_exists(&recent_file) + .await + .expect("check recent file") + ); + assert!( + tokio::fs::try_exists(&invalid_file) + .await + .expect("check invalid file") + ); + assert!( + tokio::fs::try_exists(&ignored_old_file) + .await + .expect("check ignored old file") + ); + } +} diff --git a/codex-rs/core/src/memories/mod.rs b/codex-rs/core/src/memories/mod.rs index fb033a32c..29c9d375f 100644 --- a/codex-rs/core/src/memories/mod.rs +++ b/codex-rs/core/src/memories/mod.rs @@ -30,6 +30,8 @@ mod artifacts { pub(super) const RAW_MEMORIES_FILENAME: &str = "raw_memories.md"; } +mod extensions; + /// Phase 1 (startup extraction). mod phase_one { /// Default model used for phase 1. diff --git a/codex-rs/core/src/memories/phase2.rs b/codex-rs/core/src/memories/phase2.rs index 2abc435ca..dd194edfb 100644 --- a/codex-rs/core/src/memories/phase2.rs +++ b/codex-rs/core/src/memories/phase2.rs @@ -3,6 +3,9 @@ use crate::agent::status::is_final as is_final_agent_status; use crate::codex::Session; use crate::codex::emit_subagent_session_started; use crate::config::Config; +use crate::memories::extensions::PendingExtensionResourceRemoval; +use crate::memories::extensions::find_old_extension_resources; +use crate::memories::extensions::remove_extension_resources; use crate::memories::memory_root; use crate::memories::metrics; use crate::memories::phase_two; @@ -112,7 +115,12 @@ pub(super) async fn run(session: &Arc, config: Arc) { job::failed(session, db, &claim, "failed_rebuild_raw_memories").await; return; } - if raw_memories.is_empty() { + let pending_extension_resource_removals = find_old_extension_resources(&root).await; + let removed_extension_resources = pending_extension_resource_removals + .iter() + .map(|resource| resource.removed.clone()) + .collect::>(); + if raw_memories.is_empty() && pending_extension_resource_removals.is_empty() { // We check only after sync of the file system. job::succeed( session, @@ -127,7 +135,7 @@ pub(super) async fn run(session: &Arc, config: Arc) { } // 5. Spawn the agent - let prompt = agent::get_prompt(config, &selection); + let prompt = agent::get_prompt(config, &selection, &removed_extension_resources); let source = SessionSource::SubAgent(SubAgentSource::MemoryConsolidation); let thread_id = match session .services @@ -170,6 +178,7 @@ pub(super) async fn run(session: &Arc, config: Arc) { claim, new_watermark, raw_memories.clone(), + pending_extension_resource_removals, thread_id, phase_two_e2e_timer, ); @@ -268,15 +277,15 @@ mod job { completion_watermark: i64, selected_outputs: &[codex_state::Stage1Output], reason: &'static str, - ) { + ) -> bool { session.services.session_telemetry.counter( metrics::MEMORY_PHASE_TWO_JOBS, /*inc*/ 1, &[("status", reason)], ); - let _ = db - .mark_global_phase2_job_succeeded(&claim.token, completion_watermark, selected_outputs) - .await; + db.mark_global_phase2_job_succeeded(&claim.token, completion_watermark, selected_outputs) + .await + .unwrap_or(false) } } @@ -328,9 +337,10 @@ mod agent { pub(super) fn get_prompt( config: Arc, selection: &codex_state::Phase2InputSelection, + removed_extension_resources: &[crate::memories::extensions::RemovedExtensionResource], ) -> Vec { let root = memory_root(&config.codex_home); - let prompt = build_consolidation_prompt(&root, selection); + let prompt = build_consolidation_prompt(&root, selection, removed_extension_resources); vec![UserInput::Text { text: prompt, text_elements: vec![], @@ -343,6 +353,7 @@ mod agent { claim: Claim, new_watermark: i64, selected_outputs: Vec, + pending_extension_resource_removals: Vec, thread_id: ThreadId, phase_two_e2e_timer: Option, ) { @@ -379,7 +390,7 @@ mod agent { if let Some(token_usage) = agent_control.get_total_token_usage(thread_id).await { emit_token_usage_metrics(&session, &token_usage); } - job::succeed( + if job::succeed( &session, &db, &claim, @@ -387,7 +398,10 @@ mod agent { &selected_outputs, "succeeded", ) - .await; + .await + { + remove_extension_resources(&pending_extension_resource_removals).await; + } } else { job::failed(&session, &db, &claim, "failed_agent").await; } diff --git a/codex-rs/core/src/memories/prompts.rs b/codex-rs/core/src/memories/prompts.rs index b6e18b405..fc85917c9 100644 --- a/codex-rs/core/src/memories/prompts.rs +++ b/codex-rs/core/src/memories/prompts.rs @@ -1,3 +1,5 @@ +use crate::memories::extensions::EXTENSION_RESOURCE_RETENTION_DAYS; +use crate::memories::extensions::RemovedExtensionResource; use crate::memories::memory_extensions_root; use crate::memories::memory_root; use crate::memories::phase_one; @@ -10,6 +12,7 @@ use codex_utils_absolute_path::AbsolutePathBuf; use codex_utils_output_truncation::TruncationPolicy; use codex_utils_output_truncation::truncate_text; use codex_utils_template::Template; +use std::fmt::Write as _; use std::path::Path; use std::sync::LazyLock; use tokio::fs; @@ -62,8 +65,9 @@ Memory extensions (under {{ memory_extensions_root }}/): source. If the user has any memory extensions, you MUST read the instructions for each extension to -determine how to use the memory source. If it has no extension folders, continue with the standard -memory inputs only. +determine how to use the memory source. If the Phase 2 diff lists removed memory extension +resources, use that extension-specific deletion diff to remove stale memories derived only from +those resources. If it has no extension folders, continue with the standard memory inputs only. "#; const MEMORY_EXTENSIONS_PRIMARY_INPUTS: &str = r#" @@ -73,12 +77,16 @@ Under `{{ memory_extensions_root }}/`: - `/instructions.md` - If extension folders exist, read each instructions.md first and follow it when interpreting that extension's memory source. + +If the Phase 2 diff lists removed memory extension resources, use that extension-specific deletion +diff to remove stale memories derived only from those resources. "#; /// Builds the consolidation subagent prompt for a specific memory root. pub(super) fn build_consolidation_prompt( memory_root: &Path, selection: &Phase2InputSelection, + removed_extension_resources: &[RemovedExtensionResource], ) -> String { let memory_extensions_root = memory_extensions_root(memory_root); let memory_extensions_exist = memory_extensions_root.is_dir(); @@ -100,7 +108,8 @@ pub(super) fn build_consolidation_prompt( } else { String::new() }; - let phase2_input_selection = render_phase2_input_selection(selection); + let phase2_input_selection = + render_phase2_input_selection(selection, removed_extension_resources); CONSOLIDATION_PROMPT_TEMPLATE .render([ ("memory_root", memory_root.as_str()), @@ -131,7 +140,10 @@ fn render_memory_extensions_block(template: &Template, memory_extensions_root: & }) } -fn render_phase2_input_selection(selection: &Phase2InputSelection) -> String { +fn render_phase2_input_selection( + selection: &Phase2InputSelection, + removed_extension_resources: &[RemovedExtensionResource], +) -> String { let retained = selection.retained_thread_ids.len(); let added = selection.selected.len().saturating_sub(retained); let selected = if selection.selected.is_empty() { @@ -160,11 +172,29 @@ fn render_phase2_input_selection(selection: &Phase2InputSelection) -> String { .join("\n") }; - format!( + let mut rendered = format!( "- selected inputs this run: {}\n- newly added since the last successful Phase 2 run: {added}\n- retained from the last successful Phase 2 run: {retained}\n- removed from the last successful Phase 2 run: {}\n\nCurrent selected Phase 1 inputs:\n{selected}\n\nRemoved from the last successful Phase 2 selection:\n{removed}\n", selection.selected.len(), selection.removed.len(), - ) + ); + + if !removed_extension_resources.is_empty() { + rendered.push_str("\nMemory extension resources removed by retention pruning:\n"); + let _ = writeln!( + rendered, + "- retention window: {EXTENSION_RESOURCE_RETENTION_DAYS} days" + ); + let mut current_extension = ""; + for removed_resource in removed_extension_resources { + if removed_resource.extension != current_extension { + current_extension = &removed_resource.extension; + let _ = writeln!(rendered, "- extension: {current_extension}"); + } + let _ = writeln!(rendered, " - {}", removed_resource.resource_path); + } + } + + rendered } fn render_selected_input_line(item: &Stage1Output, retained: bool) -> String { diff --git a/codex-rs/core/src/memories/prompts_tests.rs b/codex-rs/core/src/memories/prompts_tests.rs index 325d5e923..959bb6911 100644 --- a/codex-rs/core/src/memories/prompts_tests.rs +++ b/codex-rs/core/src/memories/prompts_tests.rs @@ -1,5 +1,7 @@ use super::*; +use crate::memories::extensions::RemovedExtensionResource; use codex_models_manager::model_info::model_info_from_slug; +use codex_state::Phase2InputSelection; use core_test_support::PathExt; use pretty_assertions::assert_eq; use tempfile::tempdir; @@ -54,6 +56,36 @@ fn build_stage_one_input_message_uses_default_limit_when_model_context_window_mi assert!(message.contains(&expected_truncated)); } +#[test] +fn build_consolidation_prompt_includes_removed_extension_resources() { + let temp = tempdir().unwrap(); + let memory_root = temp.path().join("memories"); + std::fs::create_dir_all(temp.path().join("memories_extensions")).unwrap(); + let removed_extension_resources = vec![ + RemovedExtensionResource { + extension: "telepathy".to_string(), + resource_path: "resources/2026-04-06T11-59-59-abcd-10min-old.md".to_string(), + }, + RemovedExtensionResource { + extension: "telepathy".to_string(), + resource_path: "resources/2026-04-07T12-00-00-abcd-10min-cutoff.md".to_string(), + }, + ]; + + let prompt = build_consolidation_prompt( + &memory_root, + &Phase2InputSelection::default(), + &removed_extension_resources, + ); + + assert!(prompt.contains("Memory extension resources removed by retention pruning:")); + assert!(prompt.contains("- retention window: 7 days")); + assert!(prompt.contains("- extension: telepathy")); + assert!(prompt.contains(" - resources/2026-04-06T11-59-59-abcd-10min-old.md")); + assert!(prompt.contains(" - resources/2026-04-07T12-00-00-abcd-10min-cutoff.md")); + assert!(prompt.contains("extension-specific deletion diff")); +} + #[tokio::test] async fn build_memory_tool_developer_instructions_renders_embedded_template() { let temp = tempdir().unwrap(); diff --git a/codex-rs/core/src/memories/tests.rs b/codex-rs/core/src/memories/tests.rs index a0ed29dbe..71809d382 100644 --- a/codex-rs/core/src/memories/tests.rs +++ b/codex-rs/core/src/memories/tests.rs @@ -424,6 +424,7 @@ mod phase2 { use crate::memories::phase2; use crate::memories::raw_memories_file; use crate::memories::rollout_summaries_dir; + use chrono::Duration as ChronoDuration; use chrono::Utc; use codex_config::Constrained; use codex_login::CodexAuth; @@ -960,6 +961,28 @@ mod phase2 { "stage-1 success should enqueue global consolidation" ); + let telepathy_resources = config + .codex_home + .join("memories_extensions/telepathy/resources"); + tokio::fs::create_dir_all(&telepathy_resources) + .await + .expect("create telepathy resources"); + tokio::fs::write( + config + .codex_home + .join("memories_extensions/telepathy/instructions.md"), + "instructions", + ) + .await + .expect("write telepathy instructions"); + let old_file = telepathy_resources.join(format!( + "{}-abcd-10min-old.md", + (Utc::now() - ChronoDuration::days(8)).format("%Y-%m-%dT%H-%M-%S") + )); + tokio::fs::write(&old_file, "old resource") + .await + .expect("write old extension resource"); + phase2::run(&session, Arc::clone(&config)).await; let retry_claim = state_db @@ -971,5 +994,11 @@ mod phase2 { Phase2JobClaimOutcome::SkippedNotDirty, "spawn failures should leave the job in retry backoff instead of running" ); + assert!( + tokio::fs::try_exists(&old_file) + .await + .expect("check old extension resource"), + "spawn failures should not prune extension resources before retry" + ); } } diff --git a/codex-rs/core/tests/suite/memories.rs b/codex-rs/core/tests/suite/memories.rs index 98e618643..60685592f 100644 --- a/codex-rs/core/tests/suite/memories.rs +++ b/codex-rs/core/tests/suite/memories.rs @@ -159,6 +159,145 @@ async fn memories_startup_phase2_tracks_added_and_removed_inputs_across_runs() - Ok(()) } +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn memories_startup_phase2_prunes_old_extension_resources_and_reports_them() -> Result<()> { + let server = start_mock_server().await; + let home = Arc::new(TempDir::new()?); + let db = init_state_db(&home).await?; + let now = Utc::now(); + let thread_id = seed_stage1_output( + db.as_ref(), + home.path(), + now - ChronoDuration::hours(1), + "raw memory", + "rollout summary", + "rollout", + ) + .await?; + + let telepathy_resources = home.path().join("memories_extensions/telepathy/resources"); + tokio::fs::create_dir_all(&telepathy_resources).await?; + tokio::fs::write( + home.path() + .join("memories_extensions/telepathy/instructions.md"), + "instructions", + ) + .await?; + let old_file_name = format!( + "{}-abcd-10min-old.md", + (now - ChronoDuration::days(8)).format("%Y-%m-%dT%H-%M-%S") + ); + let old_file = telepathy_resources.join(&old_file_name); + tokio::fs::write(&old_file, "old resource").await?; + let recent_file = telepathy_resources.join(format!( + "{}-abcd-10min-recent.md", + (now - ChronoDuration::days(6)).format("%Y-%m-%dT%H-%M-%S") + )); + tokio::fs::write(&recent_file, "recent resource").await?; + + let phase2 = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-phase2"), + ev_assistant_message("msg-phase2", "phase2 complete"), + ev_completed("resp-phase2"), + ]), + ) + .await; + + let codex = build_test_codex(&server, home.clone()).await?; + let request = wait_for_single_request(&phase2).await; + let prompt = phase2_prompt_text(&request); + + assert!( + prompt.contains("Memory extension resources removed by retention pruning:"), + "expected extension resource prune report in prompt: {prompt}" + ); + assert!( + prompt.contains("- retention window: 7 days"), + "expected retention window in prompt: {prompt}" + ); + assert!( + prompt.contains("- extension: telepathy"), + "expected extension name in prompt: {prompt}" + ); + assert!( + prompt.contains(&format!(" - resources/{old_file_name}")), + "expected old resource in prompt: {prompt}" + ); + + wait_for_phase2_success(db.as_ref(), thread_id).await?; + wait_for_file_removed(&old_file).await?; + assert!( + !tokio::fs::try_exists(&old_file).await?, + "old extension resource should be pruned" + ); + assert!( + tokio::fs::try_exists(&recent_file).await?, + "recent extension resource should be retained" + ); + + shutdown_test_codex(&codex).await?; + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn memories_startup_phase2_processes_old_extension_resources_without_stage1_input() +-> Result<()> { + let server = start_mock_server().await; + let home = Arc::new(TempDir::new()?); + let db = init_state_db(&home).await?; + db.enqueue_global_consolidation(/*input_watermark*/ 1) + .await?; + + let now = Utc::now(); + let telepathy_resources = home.path().join("memories_extensions/telepathy/resources"); + tokio::fs::create_dir_all(&telepathy_resources).await?; + tokio::fs::write( + home.path() + .join("memories_extensions/telepathy/instructions.md"), + "instructions", + ) + .await?; + let old_file_name = format!( + "{}-abcd-10min-old.md", + (now - ChronoDuration::days(8)).format("%Y-%m-%dT%H-%M-%S") + ); + let old_file = telepathy_resources.join(&old_file_name); + tokio::fs::write(&old_file, "old resource").await?; + + let phase2 = mount_sse_once( + &server, + sse(vec![ + ev_response_created("resp-phase2-empty"), + ev_assistant_message("msg-phase2-empty", "phase2 complete"), + ev_completed("resp-phase2-empty"), + ]), + ) + .await; + + let codex = build_test_codex(&server, home.clone()).await?; + let request = wait_for_single_request(&phase2).await; + let prompt = phase2_prompt_text(&request); + + assert!( + prompt.contains("- selected inputs this run: 0"), + "expected no selected raw inputs in prompt: {prompt}" + ); + assert!( + prompt.contains("Memory extension resources removed by retention pruning:"), + "expected extension resource prune report in prompt: {prompt}" + ); + assert!( + prompt.contains(&format!(" - resources/{old_file_name}")), + "expected old resource in prompt: {prompt}" + ); + wait_for_file_removed(&old_file).await?; + + shutdown_test_codex(&codex).await?; + Ok(()) +} + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn web_search_pollution_moves_selected_thread_into_removed_phase2_inputs() -> Result<()> { let server = start_mock_server().await; @@ -386,6 +525,21 @@ async fn wait_for_single_request(mock: &ResponseMock) -> ResponsesRequest { wait_for_request(mock, /*expected_count*/ 1).await.remove(0) } +async fn wait_for_file_removed(path: &Path) -> Result<()> { + let deadline = Instant::now() + Duration::from_secs(10); + loop { + if !tokio::fs::try_exists(path).await? { + return Ok(()); + } + assert!( + Instant::now() < deadline, + "timed out waiting for {} to be removed", + path.display() + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } +} + async fn wait_for_request(mock: &ResponseMock, expected_count: usize) -> Vec { let deadline = Instant::now() + Duration::from_secs(10); loop {