make it better

This commit is contained in:
jif-oai
2026-04-20 17:37:02 +01:00
parent 738b915021
commit 3cc268c947
6 changed files with 316 additions and 62 deletions
+4 -2
View File
@@ -123,8 +123,10 @@ Selection diff behavior:
- rows that were previously selected but still exist outside the current top-N
selection are surfaced as `removed`
- before the agent starts, local `rollout_summaries/` and `raw_memories.md`
keep the union of the current selection and the previous successful
selection, so removed-thread evidence stays available during forgetting
reflect the current selection so removed-thread summaries show up as
workspace deletions during forgetting; when removals are pending, existing
consolidated artifacts such as `MEMORY.md`, `memory_summary.md`, and
`skills/` are preserved for the agent's targeted cleanup
Watermark behavior:
+8 -1
View File
@@ -2,7 +2,14 @@ use std::path::Path;
pub async fn clear_memory_roots_contents(codex_home: &Path) -> std::io::Result<()> {
let memory_root = codex_home.join("memories");
clear_memory_root_contents(&memory_root).await
clear_memory_root_contents(&memory_root).await?;
let legacy_extensions_root = codex_home.join("memories_extensions");
match tokio::fs::symlink_metadata(&legacy_extensions_root).await {
Ok(_) => clear_memory_root_contents(&legacy_extensions_root).await,
Err(err) if err.kind() == std::io::ErrorKind::NotFound => Ok(()),
Err(err) => Err(err),
}
}
pub(crate) async fn clear_memory_root_contents(memory_root: &Path) -> std::io::Result<()> {
+13 -3
View File
@@ -7,6 +7,7 @@ use crate::memories::memory_root;
use crate::memories::metrics;
use crate::memories::phase_two;
use crate::memories::prompts::build_consolidation_prompt;
use crate::memories::storage::EmptyArtifactCleanup;
use crate::memories::storage::rebuild_raw_memories_file_from_memories;
use crate::memories::storage::sync_rollout_summaries_from_memories;
use crate::memories::workspace::commit_all;
@@ -102,14 +103,23 @@ pub(super) async fn run(session: &Arc<Session>, config: Arc<Config>) {
};
let raw_memories = selection.selected.to_vec();
let artifact_memories = selection.selected.clone();
let empty_artifact_cleanup = if selection.removed.is_empty() {
EmptyArtifactCleanup::RemoveConsolidated
} else {
EmptyArtifactCleanup::PreserveConsolidated
};
let new_watermark = get_watermark(claim.watermark, &raw_memories);
// 4. Update the file system by syncing the raw memories with the one extracted from DB at
// step 3
// [`rollout_summaries/`]
if let Err(err) =
sync_rollout_summaries_from_memories(&root, &artifact_memories, artifact_memories.len())
.await
if let Err(err) = sync_rollout_summaries_from_memories(
&root,
&artifact_memories,
artifact_memories.len(),
empty_artifact_cleanup,
)
.await
{
tracing::error!("failed syncing local memory artifacts for global consolidation: {err}");
job::failed(session, db, &claim, "failed_sync_artifacts").await;
+8 -1
View File
@@ -9,6 +9,12 @@ use crate::memories::ensure_layout;
use crate::memories::raw_memories_file;
use crate::memories::rollout_summaries_dir;
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(super) enum EmptyArtifactCleanup {
RemoveConsolidated,
PreserveConsolidated,
}
/// Rebuild `raw_memories.md` from DB-backed stage-1 outputs.
pub(super) async fn rebuild_raw_memories_file_from_memories(
root: &Path,
@@ -24,6 +30,7 @@ pub(super) async fn sync_rollout_summaries_from_memories(
root: &Path,
memories: &[Stage1Output],
max_raw_memories_for_consolidation: usize,
empty_artifact_cleanup: EmptyArtifactCleanup,
) -> std::io::Result<()> {
ensure_layout(root).await?;
@@ -38,7 +45,7 @@ pub(super) async fn sync_rollout_summaries_from_memories(
write_rollout_summary_for_thread(root, memory).await?;
}
if retained.is_empty() {
if retained.is_empty() && empty_artifact_cleanup == EmptyArtifactCleanup::RemoveConsolidated {
for file_name in ["MEMORY.md", "memory_summary.md"] {
let path = root.join(file_name);
if let Err(err) = tokio::fs::remove_file(path).await
+239 -18
View File
@@ -1,4 +1,6 @@
use super::control::clear_memory_root_contents;
use super::control::clear_memory_roots_contents;
use super::storage::EmptyArtifactCleanup;
use super::storage::rebuild_raw_memories_file_from_memories;
use super::storage::sync_rollout_summaries_from_memories;
use crate::memories::ensure_layout;
@@ -102,6 +104,59 @@ async fn clear_memory_root_contents_preserves_root_directory() {
);
}
#[tokio::test]
async fn clear_memory_roots_contents_clears_legacy_extensions_root() {
let dir = tempdir().expect("tempdir");
let memory_root = dir.path().join("memories");
let new_extensions_root = memory_root.join("extensions");
let legacy_extensions_root = dir.path().join("memories_extensions");
tokio::fs::create_dir_all(&new_extensions_root)
.await
.expect("create new extensions dir");
tokio::fs::create_dir_all(&legacy_extensions_root)
.await
.expect("create legacy extensions dir");
tokio::fs::write(
new_extensions_root.join("stale.md"),
"new stale extension\n",
)
.await
.expect("write new extension artifact");
tokio::fs::write(
legacy_extensions_root.join("stale.md"),
"legacy stale extension\n",
)
.await
.expect("write legacy extension artifact");
clear_memory_roots_contents(dir.path())
.await
.expect("clear memory roots contents");
let mut memory_entries = tokio::fs::read_dir(&memory_root)
.await
.expect("read memory root after clear");
assert!(
memory_entries
.next_entry()
.await
.expect("read memory root entry")
.is_none(),
"memory root should be empty after clearing contents"
);
let mut legacy_entries = tokio::fs::read_dir(&legacy_extensions_root)
.await
.expect("read legacy extensions root after clear");
assert!(
legacy_entries
.next_entry()
.await
.expect("read legacy extensions root entry")
.is_none(),
"legacy extensions root should be empty after clearing contents"
);
}
#[cfg(unix)]
#[tokio::test]
async fn clear_memory_root_contents_rejects_symlinked_root() {
@@ -163,6 +218,7 @@ async fn sync_rollout_summaries_and_raw_memories_file_keeps_latest_memories_only
&root,
&memories,
DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
EmptyArtifactCleanup::RemoveConsolidated,
)
.await
.expect("sync rollout summaries");
@@ -236,6 +292,70 @@ async fn sync_rollout_summaries_and_raw_memories_file_keeps_latest_memories_only
assert!(rollout_path_pos < file_pos);
}
#[tokio::test]
async fn sync_rollout_summaries_preserves_consolidated_artifacts_for_removed_inputs() {
let dir = tempdir().expect("tempdir");
let root = dir.path().join("memory");
let summaries_dir = rollout_summaries_dir(&root);
let skills_dir = root.join("skills/demo");
tokio::fs::create_dir_all(&summaries_dir)
.await
.expect("create rollout summaries dir");
tokio::fs::create_dir_all(&skills_dir)
.await
.expect("create skills dir");
let stale_summary_path = summaries_dir.join("stale.md");
let memory_path = root.join("MEMORY.md");
let memory_summary_path = root.join("memory_summary.md");
let skill_path = skills_dir.join("SKILL.md");
tokio::fs::write(&stale_summary_path, "stale rollout summary\n")
.await
.expect("write stale rollout summary");
tokio::fs::write(&memory_path, "consolidated memory\n")
.await
.expect("write memory");
tokio::fs::write(&memory_summary_path, "summary\n")
.await
.expect("write memory summary");
tokio::fs::write(&skill_path, "skill\n")
.await
.expect("write skill");
sync_rollout_summaries_from_memories(
&root,
&[],
DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
EmptyArtifactCleanup::PreserveConsolidated,
)
.await
.expect("sync rollout summaries");
assert!(
!tokio::fs::try_exists(&stale_summary_path)
.await
.expect("check stale summary path"),
"removed-only sync should still prune stale rollout summaries"
);
assert!(
tokio::fs::try_exists(&memory_path)
.await
.expect("check memory path"),
"removed-only sync should preserve MEMORY.md for surgical cleanup"
);
assert!(
tokio::fs::try_exists(&memory_summary_path)
.await
.expect("check memory summary path"),
"removed-only sync should preserve memory_summary.md for surgical cleanup"
);
assert!(
tokio::fs::try_exists(&skill_path)
.await
.expect("check skill path"),
"removed-only sync should preserve skills for surgical cleanup"
);
}
#[tokio::test]
async fn sync_rollout_summaries_uses_timestamp_hash_and_sanitized_slug_filename() {
let dir = tempdir().expect("tempdir");
@@ -269,6 +389,7 @@ async fn sync_rollout_summaries_uses_timestamp_hash_and_sanitized_slug_filename(
&root,
&memories,
DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
EmptyArtifactCleanup::RemoveConsolidated,
)
.await
.expect("sync rollout summaries");
@@ -372,6 +493,7 @@ task_outcome: success
&root,
&memories,
DEFAULT_MEMORIES_MAX_RAW_MEMORIES_FOR_CONSOLIDATION,
EmptyArtifactCleanup::RemoveConsolidated,
)
.await
.expect("sync rollout summaries");
@@ -422,6 +544,7 @@ mod phase2 {
use crate::memories::phase2;
use crate::memories::raw_memories_file;
use crate::memories::rollout_summaries_dir;
use crate::memories::storage::EmptyArtifactCleanup;
use crate::memories::storage::rebuild_raw_memories_file_from_memories;
use crate::memories::storage::sync_rollout_summaries_from_memories;
use crate::memories::workspace::prepare_git_repo;
@@ -871,6 +994,112 @@ mod phase2 {
harness.shutdown_threads().await;
}
#[tokio::test]
async fn dispatch_with_removed_only_inputs_preserves_consolidated_artifacts() {
let harness = DispatchHarness::new().await;
let source_updated_at = Utc::now().timestamp();
let thread_id = harness.seed_stage1_output(source_updated_at).await;
let root = memory_root(&harness.config.codex_home);
let selection = harness
.state_db
.get_phase2_input_selection(/*n*/ 1, /*max_unused_days*/ 30)
.await
.expect("load phase2 input selection");
let selected = selection.selected.clone();
sync_rollout_summaries_from_memories(
&root,
&selected,
selected.len(),
EmptyArtifactCleanup::RemoveConsolidated,
)
.await
.expect("sync selected rollout summaries");
rebuild_raw_memories_file_from_memories(&root, &selected, selected.len())
.await
.expect("sync selected raw memories");
let memory_index_path = root.join("MEMORY.md");
let memory_summary_path = root.join("memory_summary.md");
let skill_path = root.join("skills/demo/SKILL.md");
tokio::fs::write(&memory_index_path, "consolidated memory index\n")
.await
.expect("write memory index");
tokio::fs::write(&memory_summary_path, "consolidated memory summary\n")
.await
.expect("write memory summary");
tokio::fs::create_dir_all(skill_path.parent().expect("skill parent"))
.await
.expect("create skill dir");
tokio::fs::write(&skill_path, "consolidated skill\n")
.await
.expect("write skill");
prepare_git_repo(&root)
.await
.expect("commit current memory workspace as baseline");
let claim = harness
.state_db
.try_claim_global_phase2_job(ThreadId::new(), /*lease_seconds*/ 3_600)
.await
.expect("claim global phase2 job");
let Phase2JobClaimOutcome::Claimed {
ownership_token, ..
} = claim
else {
panic!("unexpected phase2 claim outcome: {claim:?}");
};
assert!(
harness
.state_db
.mark_global_phase2_job_succeeded(&ownership_token, source_updated_at, &selected,)
.await
.expect("mark phase2 succeeded"),
"phase2 success should update selected baseline"
);
assert!(
harness
.state_db
.mark_thread_memory_mode_polluted(thread_id)
.await
.expect("mark thread polluted"),
"polluted selected thread should enqueue phase2 forgetting"
);
phase2::run(&harness.session, Arc::clone(&harness.config)).await;
pretty_assertions::assert_eq!(harness.user_input_ops_count(), 1);
assert!(
tokio::fs::try_exists(&memory_index_path)
.await
.expect("check memory index existence"),
"removed-only phase2 should preserve MEMORY.md before agent cleanup"
);
assert!(
tokio::fs::try_exists(&memory_summary_path)
.await
.expect("check memory summary existence"),
"removed-only phase2 should preserve memory_summary.md before agent cleanup"
);
assert!(
tokio::fs::try_exists(&skill_path)
.await
.expect("check skill existence"),
"removed-only phase2 should preserve skills before agent cleanup"
);
let workspace_diff = tokio::fs::read_to_string(root.join("phase2_workspace_diff.md"))
.await
.expect("read workspace diff");
assert!(
workspace_diff.contains("- D rollout_summaries/"),
"removed-only phase2 should surface deleted rollout summaries: {workspace_diff}"
);
assert!(
!workspace_diff.contains("- D MEMORY.md"),
"removed-only phase2 should not delete consolidated memory wholesale: {workspace_diff}"
);
harness.shutdown_threads().await;
}
#[tokio::test]
async fn dispatch_with_clean_workspace_preserves_selected_phase2_baseline() {
let harness = DispatchHarness::new().await;
@@ -883,9 +1112,14 @@ mod phase2 {
.expect("load phase2 input selection");
let selected = selection.selected.clone();
sync_rollout_summaries_from_memories(&root, &selected, selected.len())
.await
.expect("sync selected rollout summaries");
sync_rollout_summaries_from_memories(
&root,
&selected,
selected.len(),
EmptyArtifactCleanup::RemoveConsolidated,
)
.await
.expect("sync selected rollout summaries");
rebuild_raw_memories_file_from_memories(&root, &selected, selected.len())
.await
.expect("sync selected raw memories");
@@ -1081,21 +1315,8 @@ mod phase2 {
assert!(
!tokio::fs::try_exists(&old_file)
.await
.expect("check legacy old extension resource"),
"phase2 should migrate extension resources into the memory workspace before spawn"
);
let migrated_old_file = config.codex_home.join(format!(
"memories/extensions/chronicle/resources/{}",
old_file
.file_name()
.expect("old file name")
.to_string_lossy()
));
assert!(
!tokio::fs::try_exists(&migrated_old_file)
.await
.expect("check migrated old extension resource"),
"old extension resources are pruned before spawn so retry sees the git deletion"
.expect("check old extension resource"),
"phase2 should prune old extension resources before spawn"
);
}
}
+44 -37
View File
@@ -55,14 +55,15 @@ async fn memories_startup_phase2_tracks_added_and_removed_inputs_across_runs() -
let first = build_test_codex(&server, home.clone()).await?;
let first_request = wait_for_single_request(&first_phase2).await;
let first_prompt = phase2_prompt_text(&first_request);
let _first_prompt = phase2_prompt_text(&first_request);
let first_workspace_diff = read_workspace_diff(&home).await?;
assert!(
first_prompt.contains("- A raw_memories.md"),
"expected raw memories to be added in first prompt: {first_prompt}"
first_workspace_diff.contains("- A raw_memories.md"),
"expected raw memories to be added in first workspace diff: {first_workspace_diff}"
);
assert!(
first_prompt.contains("rollout_a.md"),
"expected rollout A summary to be added: {first_prompt}"
first_workspace_diff.contains("rollout_a.md"),
"expected rollout A summary to be added: {first_workspace_diff}"
);
wait_for_phase2_success(db.as_ref(), thread_a).await?;
@@ -99,18 +100,19 @@ async fn memories_startup_phase2_tracks_added_and_removed_inputs_across_runs() -
let second = build_test_codex(&server, home.clone()).await?;
let second_request = wait_for_single_request(&second_phase2).await;
let second_prompt = phase2_prompt_text(&second_request);
let _second_prompt = phase2_prompt_text(&second_request);
let second_workspace_diff = read_workspace_diff(&home).await?;
assert!(
second_prompt.contains("- M raw_memories.md"),
"expected raw memories to be modified in second prompt: {second_prompt}"
second_workspace_diff.contains("- M raw_memories.md"),
"expected raw memories to be modified in second workspace diff: {second_workspace_diff}"
);
assert!(
second_prompt.contains("rollout_b.md"),
"expected rollout B summary to be added: {second_prompt}"
second_workspace_diff.contains("rollout_b.md"),
"expected rollout B summary to be added: {second_workspace_diff}"
);
assert!(
second_prompt.contains("- D rollout_summaries/"),
"expected rollout A summary to be deleted: {second_prompt}"
second_workspace_diff.contains("- D rollout_summaries/"),
"expected rollout A summary to be deleted: {second_workspace_diff}"
);
wait_for_phase2_success(db.as_ref(), thread_b).await?;
@@ -155,11 +157,11 @@ async fn memories_startup_phase2_prunes_old_extension_resources_and_reports_them
)
.await?;
let chronicle_resources = home.path().join("memories_extensions/chronicle/resources");
let chronicle_resources = home.path().join("memories/extensions/chronicle/resources");
tokio::fs::create_dir_all(&chronicle_resources).await?;
tokio::fs::write(
home.path()
.join("memories_extensions/chronicle/instructions.md"),
.join("memories/extensions/chronicle/instructions.md"),
"instructions",
)
.await?;
@@ -167,8 +169,8 @@ async fn memories_startup_phase2_prunes_old_extension_resources_and_reports_them
"{}-abcd-10min-old.md",
(now - ChronoDuration::days(8)).format("%Y-%m-%dT%H-%M-%S")
);
let legacy_old_file = chronicle_resources.join(&old_file_name);
tokio::fs::write(&legacy_old_file, "old resource").await?;
let old_file = chronicle_resources.join(&old_file_name);
tokio::fs::write(&old_file, "old resource").await?;
let recent_file = chronicle_resources.join(format!(
"{}-abcd-10min-recent.md",
(now - ChronoDuration::days(6)).format("%Y-%m-%dT%H-%M-%S")
@@ -187,13 +189,14 @@ async fn memories_startup_phase2_prunes_old_extension_resources_and_reports_them
let codex = build_test_codex(&server, home.clone()).await?;
let request = wait_for_single_request(&phase2).await;
let prompt = phase2_prompt_text(&request);
let _prompt = phase2_prompt_text(&request);
let workspace_diff = read_workspace_diff(&home).await?;
assert!(
prompt.contains(&format!(
workspace_diff.contains(&format!(
"- D extensions/chronicle/resources/{old_file_name}"
)),
"expected old resource deletion in prompt: {prompt}"
"expected old resource deletion in workspace diff: {workspace_diff}"
);
wait_for_phase2_success(db.as_ref(), thread_id).await?;
@@ -229,11 +232,11 @@ async fn memories_startup_phase2_processes_old_extension_resources_without_stage
.await?;
let now = Utc::now();
let chronicle_resources = home.path().join("memories_extensions/chronicle/resources");
let chronicle_resources = home.path().join("memories/extensions/chronicle/resources");
tokio::fs::create_dir_all(&chronicle_resources).await?;
tokio::fs::write(
home.path()
.join("memories_extensions/chronicle/instructions.md"),
.join("memories/extensions/chronicle/instructions.md"),
"instructions",
)
.await?;
@@ -241,8 +244,8 @@ async fn memories_startup_phase2_processes_old_extension_resources_without_stage
"{}-abcd-10min-old.md",
(now - ChronoDuration::days(8)).format("%Y-%m-%dT%H-%M-%S")
);
let legacy_old_file = chronicle_resources.join(&old_file_name);
tokio::fs::write(&legacy_old_file, "old resource").await?;
let old_file = chronicle_resources.join(&old_file_name);
tokio::fs::write(&old_file, "old resource").await?;
let phase2 = mount_sse_once(
&server,
@@ -256,13 +259,14 @@ async fn memories_startup_phase2_processes_old_extension_resources_without_stage
let codex = build_test_codex(&server, home.clone()).await?;
let request = wait_for_single_request(&phase2).await;
let prompt = phase2_prompt_text(&request);
let _prompt = phase2_prompt_text(&request);
let workspace_diff = read_workspace_diff(&home).await?;
assert!(
prompt.contains(&format!(
workspace_diff.contains(&format!(
"- D extensions/chronicle/resources/{old_file_name}"
)),
"expected old resource deletion in prompt: {prompt}"
"expected old resource deletion in workspace diff: {workspace_diff}"
);
let old_file = home.path().join(format!(
"memories/extensions/chronicle/resources/{old_file_name}"
@@ -370,10 +374,11 @@ async fn web_search_pollution_moves_selected_thread_into_removed_phase2_inputs()
let first_phase2_request = wait_for_request(&responses, /*expected_count*/ 1)
.await
.remove(0);
let first_phase2_prompt = phase2_prompt_text(&first_phase2_request);
let _first_phase2_prompt = phase2_prompt_text(&first_phase2_request);
let first_workspace_diff = read_workspace_diff(&home).await?;
assert!(
first_phase2_prompt.contains("- A raw_memories.md"),
"expected raw memories to be added before pollution: {first_phase2_prompt}"
first_workspace_diff.contains("- A raw_memories.md"),
"expected raw memories to be added before pollution: {first_workspace_diff}"
);
wait_for_phase2_success(db.as_ref(), thread_id).await?;
@@ -440,16 +445,14 @@ async fn web_search_pollution_moves_selected_thread_into_removed_phase2_inputs()
.await;
let cleanup = build_test_codex(&server, home.clone()).await?;
let removed_request = wait_for_single_request(&removed_phase2).await;
let removed_prompt = phase2_prompt_text(&removed_request);
let _removed_prompt = phase2_prompt_text(&removed_request);
let workspace_diff = read_workspace_diff(&home).await?;
assert!(
removed_prompt.contains("- D rollout_summaries/"),
"expected polluted thread rollout summary to be deleted: {removed_prompt}"
workspace_diff.contains("- D rollout_summaries/"),
"expected polluted thread rollout summary to be deleted: {workspace_diff}"
);
let workspace_diff =
tokio::fs::read_to_string(home.path().join("memories/phase2_workspace_diff.md")).await?;
assert!(
workspace_diff.contains("Status: deleted"),
workspace_diff.contains("deleted file mode 100644"),
"expected deleted file section in workspace diff: {workspace_diff}"
);
assert!(
@@ -527,6 +530,10 @@ async fn wait_for_single_request(mock: &ResponseMock) -> ResponsesRequest {
wait_for_request(mock, /*expected_count*/ 1).await.remove(0)
}
async fn read_workspace_diff(home: &TempDir) -> Result<String> {
Ok(tokio::fs::read_to_string(home.path().join("memories/phase2_workspace_diff.md")).await?)
}
async fn wait_for_file_removed(path: &Path) -> Result<()> {
let deadline = Instant::now() + Duration::from_secs(10);
loop {
@@ -562,7 +569,7 @@ fn phase2_prompt_text(request: &ResponsesRequest) -> String {
request
.message_input_texts("user")
.into_iter()
.find(|text| text.contains("Memory workspace changes:"))
.find(|text| text.contains("Memory workspace diff:"))
.expect("phase2 prompt text")
}