From 71ab2ea77bc7fca41bb641e8a131bcba7b304a21 Mon Sep 17 00:00:00 2001 From: jif Date: Mon, 19 Oct 2026 11:27:11 +0000 Subject: [PATCH] Stop memory dual-write when v2 is selected (#55262) ## Why Dual-write prepares v2 memories while v1 is selected. Once v2 is selected, running the v1 pipeline is no longer needed. ## What changed Run both memory pipelines only when `memories.dual_write` is enabled and `memories.version` selects v1. Run only v2 when it is selected, even if dual-write remains enabled. Update the configuration documentation and schema to describe this behavior. ## Testing Extend the startup test to cover both selected versions, checking model request counts, versioned extraction and consolidation state, and the absence of v1 rollout summaries when v2 is selected. GitOrigin-RevId: 1eb85bd67d7b32883b4ff086514a2acfb6b18f3a --- codex-rs/config/src/types.rs | 2 +- codex-rs/core/config.schema.json | 2 +- codex-rs/memories/write/src/start.rs | 3 +- .../write/src/startup_dual_write_tests.rs | 205 ++++++++++-------- 4 files changed, 113 insertions(+), 99 deletions(-) diff --git a/codex-rs/config/src/types.rs b/codex-rs/config/src/types.rs index 8846fb3d95..5c0ebe0e3c 100644 --- a/codex-rs/config/src/types.rs +++ b/codex-rs/config/src/types.rs @@ -322,7 +322,7 @@ pub use codex_protocol::MemoryVersion; pub struct MemoriesToml { /// Selects the memory pipeline; v1 remains the default. pub version: Option, - /// Generate both versions while the selected version supplies context. + /// Generate both versions while v1 is selected; v2 runs alone after cutover. pub dual_write: Option, /// When `true`, external context sources mark the thread `memory_mode` as `"polluted"`. #[serde(alias = "no_memories_if_mcp_or_web_search")] diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 6ea1ea8e1b..e91aa7a3d5 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -2645,7 +2645,7 @@ "type": "boolean" }, "dual_write": { - "description": "Generate both versions while the selected version supplies context.", + "description": "Generate both versions while v1 is selected; v2 runs alone after cutover.", "type": "boolean" }, "extract_model": { diff --git a/codex-rs/memories/write/src/start.rs b/codex-rs/memories/write/src/start.rs index ebd0b9950e..102881cf6f 100644 --- a/codex-rs/memories/write/src/start.rs +++ b/codex-rs/memories/write/src/start.rs @@ -37,7 +37,8 @@ pub fn start_memories_startup_task( return; } - let versions = if config.memories.dual_write { + // Dual-write warms v2 while v1 is selected; it is no longer needed after cutover. + let versions = if config.memories.dual_write && config.memories.version == MemoryVersion::V1 { vec![MemoryVersion::V1, MemoryVersion::V2] } else { vec![config.memories.version] diff --git a/codex-rs/memories/write/src/startup_dual_write_tests.rs b/codex-rs/memories/write/src/startup_dual_write_tests.rs index 430fca52c3..9b62c3bc64 100644 --- a/codex-rs/memories/write/src/startup_dual_write_tests.rs +++ b/codex-rs/memories/write/src/startup_dual_write_tests.rs @@ -1,4 +1,4 @@ -//! Exercises both startup pipelines through model requests and committed versioned state. +//! Exercises dual-write and v2 cutover through model requests and committed versioned state. use super::*; use codex_protocol::MemoryVersion; @@ -9,104 +9,117 @@ use wiremock::matchers::method; use wiremock::matchers::path_regex; #[tokio::test] -async fn dual_write_extracts_and_consolidates_into_independent_stores() -> anyhow::Result<()> { - let server = start_mock_server().await; - let home = Arc::new(TempDir::new()?); - let mut memories = startup_test_memories_config(); - memories.dual_write = true; - let test = build_test_codex_with_memories_config(&server, Arc::clone(&home), memories).await?; - let db = test.codex.state_db().expect("memory state"); - let source = seed_stage1_candidate( - &db, - home.path(), - chrono::Utc::now() - chrono::Duration::hours(2), - "dual-write", - ) - .await?; - let v1_root = home.path().join("memories"); - let v2_root = home.path().join("memories_v2"); - seed_required_memory_artifacts(&v1_root).await?; - tokio::fs::create_dir_all(&v2_root).await?; - tokio::fs::write( - v2_root.join("memory_summary.md"), - "v1\n\n## User Profile\nV2 user\n\n## User preferences\nV2 preference\n\n## General Tips\nV2 tip\n\n## What's in Memory\nV2 source\n", - ) - .await?; - // Dispatch by the output contract rather than request order: both pipelines - // run concurrently, and each phase two can start before the other phase one. - Mock::given(method("POST")) - .and(path_regex(".*/responses$")) - .respond_with(|request: &wiremock::Request| { - let body: serde_json::Value = request.body_json().expect("response request"); - let properties = &body["text"]["format"]["schema"]["properties"]; - let output = if properties.get("raw_memory").is_some() { - r#"{"raw_memory":"legacy fact","rollout_summary":"legacy rollout","rollout_slug":"legacy"}"# - } else if properties.get("rollout_summary").is_some() { - r#"{"rollout_summary":"v2 rollout","rollout_slug":"v2"}"# - } else { - "consolidation complete" - }; - sse_response(sse(vec![ - ev_response_created("response"), - ev_assistant_message("message", output), - ev_completed("response"), - ])) - }) - .mount(&server) - .await; +async fn dual_write_runs_both_versions_until_v2_is_selected() -> anyhow::Result<()> { + for (selected_version, expected_pipelines) in [(MemoryVersion::V1, 2), (MemoryVersion::V2, 1)] { + let server = start_mock_server().await; + let home = Arc::new(TempDir::new()?); + let mut memories = startup_test_memories_config(); + memories.dual_write = true; + memories.version = selected_version; + let test = + build_test_codex_with_memories_config(&server, Arc::clone(&home), memories).await?; + let db = test.codex.state_db().expect("memory state"); + let source = seed_stage1_candidate( + &db, + home.path(), + chrono::Utc::now() - chrono::Duration::hours(2), + "dual-write", + ) + .await?; + let v1_root = home.path().join("memories"); + let v2_root = home.path().join("memories_v2"); + seed_required_memory_artifacts(&v1_root).await?; + tokio::fs::create_dir_all(&v2_root).await?; + tokio::fs::write( + v2_root.join("memory_summary.md"), + "v1\n\n## User Profile\nV2 user\n\n## User preferences\nV2 preference\n\n## General Tips\nV2 tip\n\n## What's in Memory\nV2 source\n", + ) + .await?; + // Dispatch by the output contract rather than request order: both pipelines + // run concurrently, and each phase two can start before the other phase one. + Mock::given(method("POST")) + .and(path_regex(".*/responses$")) + .respond_with(|request: &wiremock::Request| { + let body: serde_json::Value = request.body_json().expect("response request"); + let properties = &body["text"]["format"]["schema"]["properties"]; + let output = if properties.get("raw_memory").is_some() { + r#"{"raw_memory":"legacy fact","rollout_summary":"legacy rollout","rollout_slug":"legacy"}"# + } else if properties.get("rollout_summary").is_some() { + r#"{"rollout_summary":"v2 rollout","rollout_slug":"v2"}"# + } else { + "consolidation complete" + }; + sse_response(sse(vec![ + ev_response_created("response"), + ev_assistant_message("message", output), + ev_completed("response"), + ])) + }) + .mount(&server) + .await; - trigger_memories_startup(&test).await; - for (version, root, expected_raw, expected_summary) in [ - (MemoryVersion::V1, &v1_root, "legacy fact", "legacy rollout"), - (MemoryVersion::V2, &v2_root, "", "v2 rollout"), - ] { - let store = db.memories_for_version(version).await?; - // Both pipelines initialize Git workspaces and run consolidation agents. - // Allow for process startup under contention on Windows CI runners. - let deadline = Instant::now() + Duration::from_secs(60); - while store.max_consolidated_thread_count().await? == 0 { - anyhow::ensure!( - Instant::now() < deadline, - "pipeline {version:?} did not consolidate" + trigger_memories_startup(&test).await; + for (version, root, expected_raw, expected_summary) in [ + (MemoryVersion::V2, &v2_root, "", "v2 rollout"), + (MemoryVersion::V1, &v1_root, "legacy fact", "legacy rollout"), + ] { + let store = db.memories_for_version(version).await?; + if selected_version == MemoryVersion::V2 && version == MemoryVersion::V1 { + assert_eq!( + store.list_stage1_outputs_for_global(/*n*/ 10).await?, + vec![] + ); + assert_eq!(store.max_consolidated_thread_count().await?, 0); + assert!(!root.join("rollout_summaries").exists()); + continue; + } + // Both pipelines initialize Git workspaces and run consolidation agents. + // Allow for process startup under contention on Windows CI runners. + let deadline = Instant::now() + Duration::from_secs(60); + while store.max_consolidated_thread_count().await? == 0 { + anyhow::ensure!( + Instant::now() < deadline, + "pipeline {version:?} did not consolidate" + ); + tokio::time::sleep(Duration::from_millis(50)).await; + } + let outputs = store.list_stage1_outputs_for_global(/*n*/ 10).await?; + assert_eq!( + outputs + .iter() + .map(|output| ( + output.thread_id, + output.raw_memory.as_str(), + output.rollout_summary.as_str() + )) + .collect::>(), + vec![(source, expected_raw, expected_summary)] ); - tokio::time::sleep(Duration::from_millis(50)).await; + let summaries = read_rollout_summary_bodies(root).await?; + assert_eq!(summaries.len(), 1); + assert!(summaries[0].contains(expected_summary)); } - let outputs = store.list_stage1_outputs_for_global(/*n*/ 10).await?; - assert_eq!( - outputs - .iter() - .map(|output| ( - output.thread_id, - output.raw_memory.as_str(), - output.rollout_summary.as_str() - )) - .collect::>(), - vec![(source, expected_raw, expected_summary)] - ); - let summaries = read_rollout_summary_bodies(root).await?; - assert_eq!(summaries.len(), 1); - assert!(summaries[0].contains(expected_summary)); - } - let requests = server.received_requests().await.expect("recorded requests"); - let bodies = requests - .iter() - .filter(|request| request.url.path().ends_with("/responses")) - .map(|request| { - request - .body_json::() - .expect("request JSON") - }) - .collect::>(); - assert_eq!(bodies.len(), 4); - assert_eq!( - bodies + let requests = server.received_requests().await.expect("recorded requests"); + let bodies = requests .iter() - .filter(|body| body["text"]["format"]["schema"].is_null()) - .count(), - 2 - ); - assert!(!v2_root.join("MEMORY.md").exists()); - assert!(!v2_root.join("raw_memories.md").exists()); - shutdown_test_codex(&test).await?; + .filter(|request| request.url.path().ends_with("/responses")) + .map(|request| { + request + .body_json::() + .expect("request JSON") + }) + .collect::>(); + assert_eq!(bodies.len(), expected_pipelines * 2); + assert_eq!( + bodies + .iter() + .filter(|body| body["text"]["format"]["schema"].is_null()) + .count(), + expected_pipelines + ); + assert!(!v2_root.join("MEMORY.md").exists()); + assert!(!v2_root.join("raw_memories.md").exists()); + shutdown_test_codex(&test).await?; + } Ok(()) }