fix: harden Memory restore and scheduling

crates/manifest/src/lib.rs

@@ -956,6 +956,34 @@ impl WorkerManifest {
     }
 }
 
+#[derive(Debug, Default, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+struct LegacyMemoryFeatureConfig {
+    enabled: bool,
+    staging: bool,
+}
+
+#[derive(Debug, Default, Deserialize)]
+#[serde(default, deny_unknown_fields)]
+struct LegacyMemoryConfig {
+    #[serde(rename = "workspace_root")]
+    _workspace_root: Option<PathBuf>,
+    #[serde(rename = "query_result_limit")]
+    _query_result_limit: Option<usize>,
+    #[serde(rename = "query_excerpt_lines")]
+    _query_excerpt_lines: Option<usize>,
+    inject_summary: Option<bool>,
+    workspace_id: Option<String>,
+    settings_revision: Option<u64>,
+    language: Option<String>,
+    extract_model: Option<ModelManifest>,
+    extract_threshold: Option<u64>,
+    extract_worker_max_turns: Option<u32>,
+    consolidation_model: Option<ModelManifest>,
+    consolidation_threshold_files: Option<usize>,
+    consolidation_threshold_bytes: Option<u64>,
+}
+
 const RESOLVED_MANIFEST_SNAPSHOT_SCHEMA_VERSION: u64 = 2;
 
 /// Serialize a resolved Worker Manifest for durable Worker-specific storage.
@@ -1004,12 +1032,37 @@ pub fn read_persisted_worker_manifest_snapshot(
                 "resolved Worker manifest snapshot is missing manifest",
             ))
         })?;
-        return serde_json::from_value(manifest);
+        if manifest
+            .as_object()
+            .is_some_and(|manifest| manifest.contains_key("memory"))
+        {
+            return Err(serde_json::Error::io(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "current resolved Worker manifest contains removed top-level memory authority",
+            )));
+        }
+        return validate_persisted_worker_manifest(serde_json::from_value(manifest)?);
     }
 
     migrate_legacy_resolved_manifest_snapshot(snapshot)
 }
 
+fn validate_persisted_worker_manifest(
+    manifest: WorkerManifest,
+) -> Result<WorkerManifest, serde_json::Error> {
+    manifest
+        .feature
+        .memory
+        .validate_execution()
+        .map_err(|message| {
+            serde_json::Error::io(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                message,
+            ))
+        })?;
+    Ok(manifest)
+}
+
 fn migrate_legacy_resolved_manifest_snapshot(
     mut snapshot: serde_json::Value,
 ) -> Result<WorkerManifest, serde_json::Error> {
@@ -1030,44 +1083,21 @@ fn migrate_legacy_resolved_manifest_snapshot(
                 "legacy resolved Worker manifest feature must be an object",
             ))
         })?;
-    let legacy_feature_memory = feature
+    let legacy_feature_memory: LegacyMemoryFeatureConfig = serde_json::from_value(
+        feature
             .remove("memory")
-        .unwrap_or_else(|| serde_json::json!({}));
+            .unwrap_or_else(|| serde_json::json!({})),
-    let legacy_feature_memory = legacy_feature_memory.as_object().ok_or_else(|| {
+    )?;
-        serde_json::Error::io(std::io::Error::new(
+    let enabled = legacy_feature_memory.enabled;
-            std::io::ErrorKind::InvalidData,
+    let staging_tools = legacy_feature_memory.staging;
-            "legacy resolved Worker manifest feature.memory must be an object",
-        ))
-    })?;
-    if legacy_feature_memory
-        .keys()
-        .any(|key| !matches!(key.as_str(), "enabled" | "staging"))
-    {
-        return Err(serde_json::Error::io(std::io::Error::new(
-            std::io::ErrorKind::InvalidData,
-            "legacy resolved Worker manifest mixes old and new Memory configuration",
-        )));
-    }
-    let enabled = legacy_feature_memory
-        .get("enabled")
-        .and_then(serde_json::Value::as_bool)
-        .unwrap_or(false);
-    let staging_tools = legacy_feature_memory
-        .get("staging")
-        .and_then(serde_json::Value::as_bool)
-        .unwrap_or(false);
 
-    let legacy_memory = legacy_memory.unwrap_or_else(|| serde_json::json!({}));
+    let legacy_memory: LegacyMemoryConfig =
-    let legacy_memory = legacy_memory.as_object().ok_or_else(|| {
+        serde_json::from_value(legacy_memory.unwrap_or_else(|| serde_json::json!({})))?;
-        serde_json::Error::io(std::io::Error::new(
+    let mut workspace_settings = match (
-            std::io::ErrorKind::InvalidData,
+        legacy_memory.workspace_id,
-            "legacy resolved Worker manifest memory must be an object",
+        legacy_memory.settings_revision,
-        ))
+        legacy_memory.language,
-    })?;
+    ) {
-    let workspace_id = legacy_memory.get("workspace_id").cloned();
-    let settings_revision = legacy_memory.get("settings_revision").cloned();
-    let language = legacy_memory.get("language").cloned();
-    let workspace_settings = match (workspace_id, settings_revision, language) {
         (Some(workspace_id), Some(settings_revision), Some(language)) => Some(serde_json::json!({
             "workspace_id": workspace_id,
             "settings_revision": settings_revision,
@@ -1081,27 +1111,20 @@ fn migrate_legacy_resolved_manifest_snapshot(
             )));
         }
     };
-    let extraction_threshold = legacy_memory
+    if !enabled {
-        .get("extract_threshold")
+        workspace_settings = None;
-        .cloned()
+    }
-        .unwrap_or(serde_json::Value::Null);
+    let extraction_enabled = legacy_memory.extract_threshold.is_some();
-    let extraction_enabled = !extraction_threshold.is_null();
+    if legacy_memory.consolidation_model.is_some() {
-    if legacy_memory
-        .get("consolidation_model")
-        .is_some_and(|model| !model.is_null())
-    {
         return Err(serde_json::Error::io(std::io::Error::new(
             std::io::ErrorKind::InvalidData,
             "legacy resolved Worker manifest uses a Worker-owned consolidation model that cannot be migrated to Backend authority",
         )));
     }
-    let threshold_files = legacy_memory
+    let consolidation_enabled = match (
-        .get("consolidation_threshold_files")
+        legacy_memory.consolidation_threshold_files,
-        .and_then(serde_json::Value::as_u64);
+        legacy_memory.consolidation_threshold_bytes,
-    let threshold_bytes = legacy_memory
+    ) {
-        .get("consolidation_threshold_bytes")
-        .and_then(serde_json::Value::as_u64);
-    let consolidation_enabled = match (threshold_files, threshold_bytes) {
         (None, None) => false,
         (Some(5), Some(50_000)) => true,
         _ => {
@@ -1116,19 +1139,13 @@ fn migrate_legacy_resolved_manifest_snapshot(
             "enabled": enabled,
             "staging_tools": staging_tools,
             "resident": {
-                "inject_summary": legacy_memory
+                "inject_summary": legacy_memory.inject_summary.unwrap_or(true),
-                    .get("inject_summary")
-                    .and_then(serde_json::Value::as_bool)
-                    .unwrap_or(true),
             },
             "extraction": {
                 "enabled": extraction_enabled,
-                "model": legacy_memory.get("extract_model").cloned().unwrap_or(serde_json::Value::Null),
+                "model": serde_json::to_value(legacy_memory.extract_model)?,
-                "threshold": extraction_threshold,
+                "threshold": legacy_memory.extract_threshold,
-                "worker_max_turns": legacy_memory
+                "worker_max_turns": legacy_memory.extract_worker_max_turns,
-                    .get("extract_worker_max_turns")
-                    .cloned()
-                    .unwrap_or(serde_json::Value::Null),
             },
             "consolidation": {
                 "request_enabled": consolidation_enabled,
@@ -1142,7 +1159,7 @@ fn migrate_legacy_resolved_manifest_snapshot(
             .insert("workspace_settings".to_string(), workspace_settings);
     }
     feature.insert("memory".to_string(), resolved);
-    serde_json::from_value(snapshot)
+    validate_persisted_worker_manifest(serde_json::from_value(snapshot)?)
 }
 
 #[cfg(test)]
@@ -1566,6 +1583,18 @@ model_id = "claude-sonnet-4-20250514"
         let current = write_persisted_worker_manifest_snapshot(&migrated).unwrap();
         assert_eq!(current["schema_version"], 2);
         assert!(current["manifest"].get("memory").is_none());
+
+        let mut disabled =
+            serde_json::to_value(WorkerManifest::from_toml(MINIMAL_REQUIRED).unwrap()).unwrap();
+        disabled["feature"]["memory"] = serde_json::json!({ "enabled": false });
+        disabled["memory"] = serde_json::json!({
+            "workspace_id": "workspace-1",
+            "settings_revision": 9,
+            "language": "Français",
+        });
+        let disabled = read_persisted_worker_manifest_snapshot(disabled).unwrap();
+        assert!(!disabled.feature.memory.profile.enabled);
+        assert!(disabled.feature.memory.workspace_settings.is_none());
     }
 
     #[test]
@@ -1589,6 +1618,25 @@ model_id = "claude-sonnet-4-20250514"
         });
         assert!(read_persisted_worker_manifest_snapshot(custom_policy).is_err());
 
+        let current = WorkerManifest::from_toml(MINIMAL_REQUIRED).unwrap();
+        let mut current = write_persisted_worker_manifest_snapshot(&current).unwrap();
+        current["manifest"]["memory"] = serde_json::json!({
+            "workspace_id": "workspace-1",
+            "settings_revision": 1,
+            "language": "English",
+        });
+        assert!(read_persisted_worker_manifest_snapshot(current).is_err());
+
+        let mut missing_settings = WorkerManifest::from_toml(MINIMAL_REQUIRED).unwrap();
+        missing_settings.feature.memory.profile.enabled = true;
+        let missing_settings = write_persisted_worker_manifest_snapshot(&missing_settings).unwrap();
+        assert!(read_persisted_worker_manifest_snapshot(missing_settings).is_err());
+
+        let mut malformed_legacy = manifest.clone();
+        malformed_legacy["feature"]["memory"] = serde_json::json!({ "enabled": "yes" });
+        malformed_legacy["memory"] = serde_json::json!({ "unknown": true });
+        assert!(read_persisted_worker_manifest_snapshot(malformed_legacy).is_err());
+
         assert!(
             read_persisted_worker_manifest_snapshot(serde_json::json!({
                 "schema_version": 3,

crates/worker/src/feature/builtin/memory.rs

@@ -620,6 +620,48 @@ mod tests {
         assert_eq!(plan.resident_summary.as_deref(), Some("# Durable Memory"));
     }
 
+    #[tokio::test]
+    async fn memory_prompt_contribution_rereads_resident_summary_for_each_install() {
+        let prompts = crate::prompt::catalog::PromptCatalog::builtins_only().unwrap();
+        let mut config = manifest::ResolvedMemoryFeatureConfig::default();
+        config.profile.enabled = true;
+        config
+            .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot {
+                workspace_id: "workspace".to_string(),
+                settings_revision: 1,
+                language: "English".to_string(),
+            })
+            .unwrap();
+
+        let first = MemoryFeatureInstallPlan::prepare_resolved(
+            config.clone(),
+            resident_client("first resident summary"),
+            prompts.clone(),
+            None,
+        )
+        .await
+        .unwrap()
+        .unwrap();
+        let restored = MemoryFeatureInstallPlan::prepare_resolved(
+            config,
+            resident_client("updated resident summary"),
+            prompts,
+            None,
+        )
+        .await
+        .unwrap()
+        .unwrap();
+
+        assert_eq!(
+            first.resident_summary.as_deref(),
+            Some("first resident summary")
+        );
+        assert_eq!(
+            restored.resident_summary.as_deref(),
+            Some("updated resident summary")
+        );
+    }
+
     #[test]
     fn memory_feature_owns_normal_and_staging_tool_surfaces() {
         let normal = MemoryToolsFeature::new(test_client(), false);

crates/worker/src/feature/builtin/memory_lifecycle.rs

@@ -77,7 +77,10 @@ impl MemoryLifecycleFeature {
         workspace_context: WorkerWorkspaceContext,
         event_tx: Option<broadcast::Sender<Event>>,
     ) -> std::io::Result<Option<Self>> {
-        if !lifecycle_enabled || !config.profile.enabled || !config.profile.extraction.enabled {
+        if !lifecycle_enabled
+            || !config.profile.enabled
+            || (!config.profile.extraction.enabled && !config.profile.consolidation.request_enabled)
+        {
             return Ok(None);
         }
         config
@@ -521,9 +524,12 @@ impl FeatureBackgroundTask for MemoryLifecycleTask {
         context: BackgroundTaskContext,
         cancellation: BackgroundTaskCancellation,
     ) -> Result<(), HookError> {
-        let extraction = self
+        let extraction = if self.config.profile.extraction.enabled {
-            .run_extraction(context.clone(), cancellation.clone())
+            self.run_extraction(context.clone(), cancellation.clone())
-            .await;
+                .await
+        } else {
+            Ok(())
+        };
         if !cancellation.is_cancelled() {
             context.generation_fence.ensure_current()?;
             if self.config.profile.consolidation.request_enabled {
@@ -1355,6 +1361,38 @@ permission = "write"
         );
     }
 
+    #[tokio::test]
+    async fn lifecycle_task_requests_consolidation_when_extraction_is_disabled() {
+        let client = ScriptClient::new(Vec::new());
+        let extension_writes = Arc::new(Mutex::new(Vec::new()));
+        let (event_tx, _) = broadcast::channel(16);
+        let workspace_client = Arc::new(RecordingWorkspaceClient::default());
+        let mut interrupted = capture(2, 250);
+        interrupted.run_exit = CommittedRunExit::Interrupted;
+        let mut task = test_task(
+            interrupted,
+            Box::new(client),
+            extension_writes.clone(),
+            event_tx,
+            workspace_client.clone(),
+        );
+        task.config.profile.extraction.enabled = false;
+        task.config.profile.consolidation.request_enabled = true;
+        run_background_task(task).await;
+
+        assert!(extension_writes.lock().unwrap().is_empty());
+        let requests = workspace_client.requests.lock().unwrap();
+        assert!(
+            requests.iter().any(|request| {
+                request
+                    .body
+                    .as_deref()
+                    .is_some_and(|body| body == "{\"force\":false}")
+            }),
+            "recorded requests: {requests:?}"
+        );
+    }
+
     fn internal_result(
         lifecycle: WorkerRunResult,
     ) -> Result<InternalWorkerResult, InternalWorkerError> {