diff --git a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst index edd0687601..397ba2f016 100644 Binary files a/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst and b/codex-rs/app-server-protocol/schema/precomputed/app-server-exports-experimental.json.zst differ diff --git a/codex-rs/app-server-protocol/src/protocol/common.rs b/codex-rs/app-server-protocol/src/protocol/common.rs index a718641f7a..005e14a3ca 100644 --- a/codex-rs/app-server-protocol/src/protocol/common.rs +++ b/codex-rs/app-server-protocol/src/protocol/common.rs @@ -681,6 +681,12 @@ client_request_definitions! { serialization: thread_id(params.thread_id), response: v2::ThreadMemoryModeSetResponse, }, + #[experimental("memory/status")] + MemoryStatus => "memory/status" { + params: v2::MemoryStatusParams, + serialization: global("memory"), + response: v2::MemoryStatusResponse, + }, #[experimental("memory/reset")] MemoryReset => "memory/reset" { params: #[ts(type = "undefined")] #[serde(skip_serializing_if = "Option::is_none")] Option<()>, diff --git a/codex-rs/app-server-protocol/src/protocol/v2/memory.rs b/codex-rs/app-server-protocol/src/protocol/v2/memory.rs new file mode 100644 index 0000000000..3e5ccc3a10 --- /dev/null +++ b/codex-rs/app-server-protocol/src/protocol/v2/memory.rs @@ -0,0 +1,23 @@ +//! Readiness for selecting v2 memory context after background consolidation. + +use crate::JsonSchema; +use crate::TS; +use serde::Deserialize; +use serde::Serialize; + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct MemoryStatusParams { + /// Required distinct consolidated threads. Defaults to 20; supported range is 1..=4096. + #[ts(optional = nullable)] + pub min_consolidated_threads: Option, +} + +#[derive(Serialize, Deserialize, Debug, Clone, PartialEq, JsonSchema, TS)] +#[serde(rename_all = "camelCase")] +#[ts(export_to = "v2/")] +pub struct MemoryStatusResponse { + pub v2_consolidated_threads: u32, + pub v2_ready: bool, +} diff --git a/codex-rs/app-server-protocol/src/protocol/v2/mod.rs b/codex-rs/app-server-protocol/src/protocol/v2/mod.rs index 9c91c85c26..736c577706 100644 --- a/codex-rs/app-server-protocol/src/protocol/v2/mod.rs +++ b/codex-rs/app-server-protocol/src/protocol/v2/mod.rs @@ -19,6 +19,7 @@ mod fs; mod hook; mod item; mod mcp; +mod memory; mod model; mod notification; mod permissions; @@ -55,6 +56,7 @@ pub use fs::*; pub use hook::*; pub use item::*; pub use mcp::*; +pub use memory::*; pub use model::*; pub use notification::*; pub use permissions::*; diff --git a/codex-rs/app-server/src/message_processor.rs b/codex-rs/app-server/src/message_processor.rs index 3557370eee..c462bb738f 100644 --- a/codex-rs/app-server/src/message_processor.rs +++ b/codex-rs/app-server/src/message_processor.rs @@ -1318,6 +1318,9 @@ impl MessageProcessor { ClientRequest::ThreadMemoryModeSet { params, .. } => { self.thread_processor.thread_memory_mode_set(params).await } + ClientRequest::MemoryStatus { params, .. } => { + self.thread_processor.memory_status(params).await + } ClientRequest::MemoryReset { .. } => self.thread_processor.memory_reset().await, ClientRequest::ThreadUnarchive { params, .. } => { self.thread_processor diff --git a/codex-rs/app-server/src/request_processors.rs b/codex-rs/app-server/src/request_processors.rs index 4b22acac21..9795665a98 100644 --- a/codex-rs/app-server/src/request_processors.rs +++ b/codex-rs/app-server/src/request_processors.rs @@ -552,6 +552,7 @@ mod initialize_processor; mod marketplace_processor; mod mcp_event_stream; mod mcp_processor; +mod memory_status; mod persisted_resume_settings; mod plugins; mod process_exec_processor; diff --git a/codex-rs/app-server/src/request_processors/memory_status.rs b/codex-rs/app-server/src/request_processors/memory_status.rs new file mode 100644 index 0000000000..b7847015dc --- /dev/null +++ b/codex-rs/app-server/src/request_processors/memory_status.rs @@ -0,0 +1,50 @@ +//! Reports v2 readiness without exposing memory contents or persistence details to clients. + +use super::*; +use codex_app_server_protocol::MemoryStatusParams; +use codex_app_server_protocol::MemoryStatusResponse; +use codex_protocol::MemoryVersion; + +impl ThreadRequestProcessor { + pub(crate) async fn memory_status( + &self, + params: MemoryStatusParams, + ) -> Result, JSONRPCErrorError> { + let minimum = params.min_consolidated_threads.unwrap_or(20); + if !(1..=4096).contains(&minimum) { + return Err(invalid_params( + "minConsolidatedThreads must be between 1 and 4096", + )); + } + let db = self + .state_db + .as_ref() + .ok_or_else(|| internal_error("sqlite state db unavailable for memory status"))?; + let store = db + .memories_for_version(MemoryVersion::V2) + .await + .map_err(|error| internal_error(format!("failed to open v2 memory state: {error}")))?; + let count = store + .max_consolidated_thread_count() + .await + .map_err(|error| internal_error(format!("failed to read memory progress: {error}")))?; + let root = self + .config + .codex_home + .join(MemoryVersion::V2.directory_name()); + let summary = tokio::fs::read_to_string(root.join("memory_summary.md")) + .await + .ok(); + let ready = count >= minimum + && summary + .as_deref() + .is_some_and(codex_memories_write::workspace::is_valid_v2_summary); + Ok(Some( + MemoryStatusResponse { + v2_consolidated_threads: count, + v2_ready: ready, + } + .into(), + )) + } +} 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 d0274bb88d..1effa23de5 100644 --- a/codex-rs/app-server/tests/suite/v2/memory_reset.rs +++ b/codex-rs/app-server/tests/suite/v2/memory_reset.rs @@ -145,3 +145,112 @@ async fn init_state_db(codex_home: &Path) -> Result> { .await?; Ok(state_db) } + +#[tokio::test] +async fn memory_status_requires_successful_v2_consolidation_and_resets() -> Result<()> { + use codex_app_server_protocol::MemoryStatusResponse; + use codex_state::Phase2JobClaimOutcome; + let home = TempDir::new()?; + MockResponsesConfig::new("http://127.0.0.1:9") + .enable_feature(Feature::Sqlite) + .write(home.path())?; + let db = init_state_db(home.path()).await?; + seed_stage1_output(&db, home.path(), MemoryVersion::V1).await?; + let source = seed_stage1_output(&db, home.path(), MemoryVersion::V2).await?; + seed_stage1_output(&db, home.path(), MemoryVersion::V2).await?; + let mut server = TestAppServer::builder() + .with_codex_home(home.path()) + .without_auto_env() + .build_initialized_with_timeout(DEFAULT_READ_TIMEOUT) + .await?; + let params = serde_json::json!({"minConsolidatedThreads": 2}); + let request = server + .send_raw_request("memory/status", Some(params.clone())) + .await?; + let status: MemoryStatusResponse = server.read_response(request).await?; + assert_eq!( + status, + MemoryStatusResponse { + v2_consolidated_threads: 0, + v2_ready: false + } + ); + + let store = db.memories_for_version(MemoryVersion::V2).await?; + let outputs = store.list_stage1_outputs_for_global(/*n*/ 20).await?; + let Phase2JobClaimOutcome::Claimed { + ownership_token, + input_watermark, + } = store + .try_claim_global_phase2_job(source, /*lease_seconds*/ 60) + .await? + else { + panic!("claim phase 2") + }; + assert!( + !store + .mark_global_phase2_job_succeeded("wrong owner", input_watermark, &outputs) + .await? + ); + assert_eq!(store.max_consolidated_thread_count().await?, 0); + assert!( + store + .mark_global_phase2_job_succeeded(&ownership_token, input_watermark, &outputs) + .await? + ); + // A completed job without a usable artifact must not activate v2. + let request = server + .send_raw_request("memory/status", Some(params.clone())) + .await?; + let status: MemoryStatusResponse = server.read_response(request).await?; + assert_eq!( + status, + MemoryStatusResponse { + v2_consolidated_threads: 2, + v2_ready: false + } + ); + let root = home.path().join("memories_v2"); + tokio::fs::create_dir_all(&root).await?; + tokio::fs::write(root.join("memory_summary.md"), "v1\n## User Profile\nTest user\n## User preferences\nTest preference\n## General Tips\nTest tip\n## What's in Memory\nTest source\n").await?; + db.delete_thread(source).await?; + let request = server + .send_raw_request("memory/status", Some(params.clone())) + .await?; + let status: MemoryStatusResponse = server.read_response(request).await?; + assert_eq!( + status, + MemoryStatusResponse { + v2_consolidated_threads: 2, + v2_ready: true + } + ); + // The default threshold remains 20 even though a caller can choose a smaller cohort. + let request = server + .send_raw_request("memory/status", Some(serde_json::json!({}))) + .await?; + let status: MemoryStatusResponse = server.read_response(request).await?; + assert_eq!( + status, + MemoryStatusResponse { + v2_consolidated_threads: 2, + v2_ready: false + } + ); + let request = server + .send_raw_request("memory/reset", /*params*/ None) + .await?; + let _: MemoryResetResponse = server.read_response(request).await?; + let request = server + .send_raw_request("memory/status", Some(params)) + .await?; + let status: MemoryStatusResponse = server.read_response(request).await?; + assert_eq!( + status, + MemoryStatusResponse { + v2_consolidated_threads: 0, + v2_ready: false + } + ); + Ok(()) +} diff --git a/codex-rs/config/src/types.rs b/codex-rs/config/src/types.rs index 15f02473e5..abd8ee058f 100644 --- a/codex-rs/config/src/types.rs +++ b/codex-rs/config/src/types.rs @@ -294,6 +294,8 @@ 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. + 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")] pub disable_on_external_context: Option, @@ -328,6 +330,7 @@ pub struct MemoriesToml { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct MemoriesConfig { pub version: MemoryVersion, + pub dual_write: bool, pub disable_on_external_context: bool, pub generate_memories: bool, pub use_memories: bool, @@ -346,6 +349,7 @@ impl Default for MemoriesConfig { fn default() -> Self { Self { version: MemoryVersion::V1, + dual_write: false, disable_on_external_context: false, generate_memories: true, use_memories: true, @@ -367,6 +371,7 @@ impl From for MemoriesConfig { let defaults = Self::default(); Self { version: toml.version.unwrap_or(defaults.version), + dual_write: toml.dual_write.unwrap_or(defaults.dual_write), disable_on_external_context: toml .disable_on_external_context .unwrap_or(defaults.disable_on_external_context), diff --git a/codex-rs/core/config.schema.json b/codex-rs/core/config.schema.json index 02688af380..024ebbc285 100644 --- a/codex-rs/core/config.schema.json +++ b/codex-rs/core/config.schema.json @@ -2081,6 +2081,10 @@ "description": "When `true`, external context sources mark the thread `memory_mode` as `\"polluted\"`.", "type": "boolean" }, + "dual_write": { + "description": "Generate both versions while the selected version supplies context.", + "type": "boolean" + }, "extract_model": { "description": "Model used for thread summarisation.", "type": "string" diff --git a/codex-rs/core/src/config/config_tests.rs b/codex-rs/core/src/config/config_tests.rs index 50bea9b56b..35e496f3cf 100644 --- a/codex-rs/core/src/config/config_tests.rs +++ b/codex-rs/core/src/config/config_tests.rs @@ -334,6 +334,7 @@ consolidation_model = "gpt-5.2" toml::from_str::(memories).expect("TOML deserialization should succeed"); assert_eq!( Some(MemoriesToml { + dual_write: None, version: None, disable_on_external_context: Some(true), generate_memories: Some(false), @@ -361,6 +362,7 @@ consolidation_model = "gpt-5.2" assert_eq!( config.memories, MemoriesConfig { + dual_write: false, version: codex_protocol::MemoryVersion::V1, disable_on_external_context: true, generate_memories: false, diff --git a/codex-rs/memories/write/src/runtime.rs b/codex-rs/memories/write/src/runtime.rs index d35c5fbc78..cbd4a06978 100644 --- a/codex-rs/memories/write/src/runtime.rs +++ b/codex-rs/memories/write/src/runtime.rs @@ -47,6 +47,7 @@ pub(crate) struct SpawnedConsolidationAgent { #[derive(Clone, Debug)] pub(crate) struct StageOneRequestContext { + version: MemoryVersion, pub(crate) model_info: ModelInfo, pub(crate) session_telemetry: SessionTelemetry, pub(crate) reasoning_effort: Option, @@ -56,15 +57,19 @@ pub(crate) struct StageOneRequestContext { impl StageOneRequestContext { pub(crate) fn start_timer(&self, name: &str) -> Option { - self.session_telemetry.start_timer(name, &[]).ok() + self.session_telemetry + .start_timer(name, &memory_metric_tags(self.version, &[])) + .ok() } pub(crate) fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) { - self.session_telemetry.counter(name, inc, tags); + self.session_telemetry + .counter(name, inc, &memory_metric_tags(self.version, tags)); } pub(crate) fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) { - self.session_telemetry.histogram(name, value, tags); + self.session_telemetry + .histogram(name, value, &memory_metric_tags(self.version, tags)); } } @@ -78,6 +83,21 @@ pub(crate) struct MemoryStartupContext { session_telemetry: SessionTelemetry, } +fn memory_metric_tags<'a>( + version: MemoryVersion, + tags: &[(&'a str, &'a str)], +) -> Vec<(&'a str, &'a str)> { + let mut tags = tags.to_vec(); + tags.push(( + "memory_version", + match version { + MemoryVersion::V1 => "v1", + MemoryVersion::V2 => "v2", + }, + )); + tags +} + fn build_session_telemetry( auth_manager: &AuthManager, thread_id: ThreadId, @@ -209,15 +229,19 @@ impl MemoryStartupContext { } pub(crate) fn counter(&self, name: &str, inc: i64, tags: &[(&str, &str)]) { - self.session_telemetry.counter(name, inc, tags); + self.session_telemetry + .counter(name, inc, &memory_metric_tags(self.version, tags)); } pub(crate) fn histogram(&self, name: &str, value: i64, tags: &[(&str, &str)]) { - self.session_telemetry.histogram(name, value, tags); + self.session_telemetry + .histogram(name, value, &memory_metric_tags(self.version, tags)); } pub(crate) fn start_timer(&self, name: &str) -> Option { - self.session_telemetry.start_timer(name, &[]).ok() + self.session_telemetry + .start_timer(name, &memory_metric_tags(self.version, &[])) + .ok() } pub(crate) async fn stage_one_request_context( @@ -237,6 +261,7 @@ impl MemoryStartupContext { .unwrap_or(model_info.default_reasoning_summary); StageOneRequestContext { + version: self.version, model_info, session_telemetry: build_session_telemetry( &self.auth_manager, diff --git a/codex-rs/memories/write/src/start.rs b/codex-rs/memories/write/src/start.rs index 8a766d86c0..ebd0b9950e 100644 --- a/codex-rs/memories/write/src/start.rs +++ b/codex-rs/memories/write/src/start.rs @@ -10,6 +10,7 @@ use codex_core::ThreadManager; use codex_core::config::Config; use codex_features::Feature; use codex_login::AuthManager; +use codex_protocol::MemoryVersion; use codex_protocol::ThreadId; use codex_protocol::models::PermissionProfile; use codex_protocol::protocol::SessionSource; @@ -36,47 +37,58 @@ pub fn start_memories_startup_task( return; } - let context = Arc::new(MemoryStartupContext::new( - thread_manager, - Arc::clone(&auth_manager), - thread_id, - thread, - config.as_ref(), - source.clone(), - )); + let versions = if config.memories.dual_write { + vec![MemoryVersion::V1, MemoryVersion::V2] + } else { + vec![config.memories.version] + }; + for version in versions { + let mut pipeline_config = config.as_ref().clone(); + pipeline_config.memories.version = version; + let config = Arc::new(pipeline_config); + let auth_manager = Arc::clone(&auth_manager); + let parent_permission_profile = parent_permission_profile.clone(); + let context = Arc::new(MemoryStartupContext::new( + Arc::clone(&thread_manager), + Arc::clone(&auth_manager), + thread_id, + Arc::clone(&thread), + config.as_ref(), + source.clone(), + )); + tokio::spawn(async move { + if context.memory_store().await.is_none() { + warn!("state db unavailable for memories startup pipeline; skipping"); + return; + } + let root = config + .codex_home + .join(config.memories.version.directory_name()); + if let Err(err) = ensure_layout(&root).await { + warn!("failed preparing memories root: {err}"); + return; + } + if let Err(err) = seed_extension_instructions(&root).await { + warn!("failed seeding memory extension instructions: {err}"); + } - tokio::spawn(async move { - if context.memory_store().await.is_none() { - warn!("state db unavailable for memories startup pipeline; skipping"); - return; - } - let root = config - .codex_home - .join(config.memories.version.directory_name()); - if let Err(err) = ensure_layout(&root).await { - warn!("failed preparing memories root: {err}"); - return; - } - if let Err(err) = seed_extension_instructions(&root).await { - warn!("failed seeding memory extension instructions: {err}"); - } + // Clean memories to make preserve DB size. This does not consume tokens so can be + // done before the quota check. + phase1::prune(context.as_ref(), &config).await; - // Clean memories to make preserve DB size. This does not consume tokens so can be - // done before the quota check. - phase1::prune(context.as_ref(), &config).await; + if !guard::rate_limits_ok(&auth_manager, &config).await { + context.counter( + MEMORY_STARTUP, + /*inc*/ 1, + &[("status", "skipped_rate_limit")], + ); + return; + } - if !guard::rate_limits_ok(&auth_manager, &config).await { - context.counter( - MEMORY_STARTUP, - /*inc*/ 1, - &[("status", "skipped_rate_limit")], - ); - return; - } - - // Run phase 1. - phase1::run(Arc::clone(&context), Arc::clone(&config)).await; - // Run phase 2. - phase2::run(context, config, parent_permission_profile).await; - }); + // Run phase 1. + phase1::run(Arc::clone(&context), Arc::clone(&config)).await; + // Run phase 2. + phase2::run(context, config, parent_permission_profile).await; + }); + } } diff --git a/codex-rs/memories/write/src/startup_dual_write_tests.rs b/codex-rs/memories/write/src/startup_dual_write_tests.rs new file mode 100644 index 0000000000..8d25d03977 --- /dev/null +++ b/codex-rs/memories/write/src/startup_dual_write_tests.rs @@ -0,0 +1,110 @@ +//! Exercises both startup pipelines through model requests and committed versioned state. + +use super::*; +use codex_protocol::MemoryVersion; +use core_test_support::responses::sse_response; +use pretty_assertions::assert_eq; +use wiremock::Mock; +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; + + 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?; + let deadline = Instant::now() + Duration::from_secs(10); + 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)] + ); + 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 + .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?; + Ok(()) +} diff --git a/codex-rs/memories/write/src/startup_tests.rs b/codex-rs/memories/write/src/startup_tests.rs index 9b44b0488d..6a93d6754a 100644 --- a/codex-rs/memories/write/src/startup_tests.rs +++ b/codex-rs/memories/write/src/startup_tests.rs @@ -59,6 +59,9 @@ use tempfile::TempDir; use tokio::time::Duration; use tokio::time::Instant; +#[path = "startup_dual_write_tests.rs"] +mod dual_write; + #[tokio::test] async fn memories_startup_creates_memory_root() -> anyhow::Result<()> { let server = start_mock_server().await; @@ -74,6 +77,33 @@ async fn memories_startup_creates_memory_root() -> anyhow::Result<()> { Ok(()) } +#[tokio::test] +async fn dual_write_prepares_both_roots_without_importing_old_notes() -> anyhow::Result<()> { + let server = start_mock_server().await; + let home = Arc::new(TempDir::new()?); + let old_notes = home.path().join("memories/extensions/ad_hoc/notes"); + tokio::fs::create_dir_all(&old_notes).await?; + tokio::fs::write(old_notes.join("old.md"), "old memory").await?; + 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?; + trigger_memories_startup(&test).await; + wait_for_dir(&home.path().join("memories/extensions/ad_hoc")).await?; + wait_for_dir(&home.path().join("memories_v2/extensions/ad_hoc")).await?; + assert_eq!( + tokio::fs::read_to_string(old_notes.join("old.md")).await?, + "old memory" + ); + assert!( + !home + .path() + .join("memories_v2/extensions/ad_hoc/notes/old.md") + .exists() + ); + shutdown_test_codex(&test).await?; + Ok(()) +} + #[cfg(unix)] #[tokio::test] async fn memories_startup_removes_symlinked_extensions_before_seeding() -> anyhow::Result<()> { diff --git a/codex-rs/memories/write/src/workspace.rs b/codex-rs/memories/write/src/workspace.rs index a796e093c4..e7081fa351 100644 --- a/codex-rs/memories/write/src/workspace.rs +++ b/codex-rs/memories/write/src/workspace.rs @@ -89,23 +89,23 @@ pub(crate) async fn validate_consolidation_artifacts_for_version( ); if version == MemoryVersion::V2 { - for heading in [ + anyhow::ensure!(is_valid_v2_summary(&summary), "invalid v2 memory summary"); + } + Ok(()) +} + +/// Checks the artifact shared by the writer and the read-only readiness endpoint. +pub fn is_valid_v2_summary(summary: &str) -> bool { + summary.lines().next() == Some("v1") + && summary.len() < 10_000 + && [ "## User Profile", "## User preferences", "## General Tips", "## What's in Memory", - ] { - anyhow::ensure!( - summary.lines().any(|line| line.trim() == heading), - "v2 memory summary missing {heading}" - ); - } - anyhow::ensure!( - summary.len() < 10_000, - "v2 memory summary must be under 10000 UTF-8 bytes" - ); - } - Ok(()) + ] + .iter() + .all(|heading| summary.lines().any(|line| line.trim() == *heading)) } pub(crate) async fn remove_memory_symlinks(root: &Path) -> std::io::Result { diff --git a/codex-rs/state/memory_migrations/0002_consolidation_progress.sql b/codex-rs/state/memory_migrations/0002_consolidation_progress.sql new file mode 100644 index 0000000000..ec84e3a127 --- /dev/null +++ b/codex-rs/state/memory_migrations/0002_consolidation_progress.sql @@ -0,0 +1,6 @@ +-- A successful consolidation is the readiness boundary for the version experiment. +CREATE TABLE consolidation_progress ( + singleton INTEGER PRIMARY KEY CHECK (singleton = 1), + max_thread_count INTEGER NOT NULL DEFAULT 0 +); +INSERT INTO consolidation_progress (singleton) VALUES (1); diff --git a/codex-rs/state/src/runtime/memories.rs b/codex-rs/state/src/runtime/memories.rs index b282db92f3..6eb810bc46 100644 --- a/codex-rs/state/src/runtime/memories.rs +++ b/codex-rs/state/src/runtime/memories.rs @@ -1,3 +1,6 @@ +#[path = "memory_readiness.rs"] +mod readiness; + use super::threads::ThreadFilterOptions; use super::threads::push_thread_filters; use super::*; @@ -1295,6 +1298,12 @@ WHERE thread_id = ? AND source_updated_at = ? .await?; } + sqlx::query( + "UPDATE consolidation_progress SET max_thread_count = MAX(max_thread_count, ?)", + ) + .bind(i64::try_from(selected_outputs.len())?) + .execute(&mut *tx) + .await?; tx.commit().await?; Ok(true) } @@ -1420,6 +1429,9 @@ WHERE kind = ? AND job_key = ? pub(super) async fn clear_memory_data_in_pool(pool: &SqlitePool) -> anyhow::Result<()> { let mut tx = pool.begin().await?; + sqlx::query("UPDATE consolidation_progress SET max_thread_count = 0") + .execute(&mut *tx) + .await?; sqlx::query( r#" diff --git a/codex-rs/state/src/runtime/memory_readiness.rs b/codex-rs/state/src/runtime/memory_readiness.rs new file mode 100644 index 0000000000..7a9f1db7a3 --- /dev/null +++ b/codex-rs/state/src/runtime/memory_readiness.rs @@ -0,0 +1,16 @@ +//! Consolidation progress used to select a warmed memory pipeline. + +use super::MemoryStore; + +impl MemoryStore { + /// Largest number of distinct source threads included in a successful consolidation. + /// Kept across ordinary pruning, and cleared by an explicit memory reset. + pub async fn max_consolidated_thread_count(&self) -> anyhow::Result { + let count: i64 = sqlx::query_scalar( + "SELECT max_thread_count FROM consolidation_progress WHERE singleton = 1", + ) + .fetch_one(self.pool.as_ref()) + .await?; + Ok(u32::try_from(count)?) + } +}