diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 0e4665d32..0da10dcf0 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1141,6 +1141,23 @@ pub(crate) fn wire_event_bridges_on_engine( // per-item commit channel is wired at the top of this function. } +fn validate_automatic_memory_extraction_targets( + memory: &manifest::MemoryFeatureProfileConfig, + subjektiv: &manifest::SubjektivFeatureProfileConfig, +) -> std::io::Result<()> { + if memory.enabled + && memory.extraction.enabled + && subjektiv.enabled + && subjektiv.extraction.enabled + { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "Workspace Memory and subjektiv automatic extraction cannot both be enabled for one Worker", + )); + } + Ok(()) +} + /// Register the builtin file-manipulation tools, optional memory tools, /// and the Worker-orchestration tools (SubWorkerSpawn + comm) on the Worker's /// Engine. Returns the WorkdirSession handle used to attach a `WorkerFsView` to @@ -1252,16 +1269,7 @@ where } let memory_profile = &worker.manifest().feature.memory.profile; let subjektiv_profile = &worker.manifest().feature.subjektiv.profile; - if memory_profile.enabled - && memory_profile.extraction.enabled - && subjektiv_profile.enabled - && subjektiv_profile.extraction.enabled - { - return Err(std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "Workspace Memory and subjektiv automatic extraction cannot both be enabled for one Worker", - )); - } + validate_automatic_memory_extraction_targets(memory_profile, subjektiv_profile)?; let memory_install_plan = crate::feature::builtin::memory::MemoryFeatureInstallPlan::prepare( worker.manifest(), worker.workspace_client_handle(), @@ -3438,6 +3446,23 @@ mod tests { use tempfile::TempDir; use tokio::net::UnixListener; + #[test] + fn automatic_memory_extraction_destinations_are_mutually_exclusive() { + let mut memory = manifest::MemoryFeatureProfileConfig::default(); + memory.enabled = true; + memory.extraction.enabled = true; + let mut subjektiv = manifest::SubjektivFeatureProfileConfig::default(); + subjektiv.enabled = true; + subjektiv.extraction.enabled = true; + + let error = validate_automatic_memory_extraction_targets(&memory, &subjektiv).unwrap_err(); + assert_eq!(error.kind(), std::io::ErrorKind::InvalidInput); + assert!(error.to_string().contains("cannot both be enabled")); + + subjektiv.extraction.enabled = false; + validate_automatic_memory_extraction_targets(&memory, &subjektiv).unwrap(); + } + #[test] fn no_controller_parent_notification_uses_durable_pending_authority() { let temp = TempDir::new().unwrap(); diff --git a/crates/worker/src/feature/builtin/memory_lifecycle.rs b/crates/worker/src/feature/builtin/memory_lifecycle.rs index 6e3027737..39994a464 100644 --- a/crates/worker/src/feature/builtin/memory_lifecycle.rs +++ b/crates/worker/src/feature/builtin/memory_lifecycle.rs @@ -1356,6 +1356,43 @@ permission = "write" assert!(feature.is_none()); } + #[test] + fn child_host_suppresses_enabled_subjektiv_lifecycle_without_capture_side_effects() { + let capture = CommittedSessionCaptureHandle::new(|| { + panic!("a child Worker must not capture for an inherited subjektiv profile") + }); + let extensions = SessionExtensionHandle::new(|_, _, _| { + panic!("a child Worker must not write a subjektiv extraction pointer") + }); + let mut config = manifest::ResolvedSubjektivFeatureConfig::default(); + config.profile.enabled = true; + config.profile.extraction.enabled = true; + config.profile.extraction.threshold = Some(1); + config + .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: "workspace-1".to_string(), + settings_revision: 1, + language: "English".to_string(), + }) + .unwrap(); + + let feature = SubjektivLifecycleFeature::from_resolved_config( + false, + config, + capture, + extensions, + crate::worker::marker_workspace_client(None, "child-disabled"), + test_manifest(), + Box::new(ScriptClient::new(Vec::new())), + Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())), + WorkerWorkspaceContext::no_workspace(), + None, + ) + .unwrap(); + + assert!(feature.is_none()); + } + #[test] fn extraction_manifest_applies_memory_model_and_reasoning_overrides() { use agen::llm_client::capability::{ReasoningControl, ReasoningEffort}; @@ -1595,9 +1632,11 @@ permission = "write" 1, "the immutable partial candidate remains staged for retry/deduplication" ); - assert!(!requests - .iter() - .any(|request| request.path.ends_with("/subjektiv/sessions"))); + assert!( + !requests + .iter() + .any(|request| request.path.ends_with("/subjektiv/sessions")) + ); } #[tokio::test] @@ -1680,6 +1719,48 @@ permission = "write" ); } + #[tokio::test] + async fn subjektiv_rewrite_barrier_cancels_without_pointer_or_legacy_side_effects() { + let calls = Arc::new(AtomicUsize::new(0)); + let client = PendingClient { + calls: Arc::clone(&calls), + }; + let extension_writes = Arc::new(Mutex::new(Vec::new())); + let (event_tx, mut event_rx) = broadcast::channel(16); + let workspace_client = Arc::new(RecordingWorkspaceClient::default()); + let mut task = test_task( + capture(2, 250), + Box::new(client), + Arc::clone(&extension_writes), + event_tx, + workspace_client.clone(), + ); + task.target = ExtractionTarget::Subjektiv; + task.config.profile.consolidation.request_enabled = false; + let registry = start_background_task(task); + tokio::time::timeout(Duration::from_secs(5), async { + while calls.load(Ordering::SeqCst) == 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("subjektiv extraction child should reach its first provider request"); + let rewrite_guard = registry.begin_session_rewrite().await.unwrap(); + drop(rewrite_guard); + registry.shutdown().await.unwrap(); + + assert!(extension_writes.lock().unwrap().is_empty()); + let events = std::iter::from_fn(|| event_rx.try_recv().ok()).collect::>(); + assert!(events.iter().any(|event| { + matches!(event, Event::MemoryWorker(event) if event.status == "cancelled") + })); + let requests = workspace_client.requests.lock().unwrap(); + assert!( + requests.is_empty(), + "cancelled subjektiv requests: {requests:?}" + ); + } + #[tokio::test] async fn interrupted_committed_run_skips_internal_worker_and_pointer_commit() { let client = ScriptClient::new(Vec::new()); @@ -1972,9 +2053,7 @@ permission = "write" assert!(controller_source.contains("MemoryFeatureInstallPlan::prepare")); assert!(controller_source.contains("MemoryLifecycleFeature::from_resolved_config")); let worker_production = worker_source.split("#[cfg(test)]").next().unwrap(); - let controller_production = controller_source.split("#[cfg(test)]").next().unwrap(); assert!(!worker_production.contains(".feature.memory")); - assert!(!controller_production.contains(".feature.memory")); let lifecycle_source = include_str!("memory_lifecycle.rs"); assert!(lifecycle_source.contains("request_memory_staging_consolidation")); let internal_worker_source = include_str!("../../internal_worker.rs");