From ba98a6afe6459621b0bbfe3c948a7db4639da607 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 2 Oct 2026 11:13:11 +0900 Subject: [PATCH] fix: harden subject memory consolidation --- crates/manifest/src/profile.rs | 49 ++++ crates/worker/src/feature/builtin/memory.rs | 66 +++++ .../src/feature/builtin/memory_lifecycle.rs | 34 ++- crates/workspace-server/src/hosts.rs | 3 +- crates/workspace-server/src/server.rs | 240 +++++++++++++++--- .../subjektiv-memory-consolidation.dcdl | 32 ++- 6 files changed, 387 insertions(+), 37 deletions(-) diff --git a/crates/manifest/src/profile.rs b/crates/manifest/src/profile.rs index c504ce154..78d91b1ff 100644 --- a/crates/manifest/src/profile.rs +++ b/crates/manifest/src/profile.rs @@ -1325,6 +1325,55 @@ mod tests { } } + #[test] + fn subjektiv_memory_consolidation_profile_is_lifecycle_and_capability_isolated() { + let tmp = TempDir::new().unwrap(); + let resolved = ProfileResolver::new() + .with_workspace_base(tmp.path()) + .resolve( + &ProfileSelector::source_named( + ProfileRegistrySource::Builtin, + "subjektiv-memory-consolidation", + ), + ProfileResolveOptions::with_worker_name("subject-consolidator"), + ) + .unwrap(); + let manifest = resolved.manifest; + let feature = &manifest.feature; + + assert!(feature.memory.profile.enabled); + assert!(feature.memory.profile.staging_tools); + assert!(!feature.memory.profile.resident.inject_summary); + assert!(!feature.memory.profile.extraction.enabled); + assert!(!feature.memory.profile.consolidation.request_enabled); + assert!(!feature.subjektiv.profile.enabled); + assert!(!feature.task.enabled); + assert!(!feature.web.enabled); + assert!(!feature.image.enabled); + assert!(!feature.sub_worker.enabled); + assert!(!feature.flow.enabled); + assert!(!feature.worker.enabled); + assert!(!feature.worker.direct_spawn); + assert!(!feature.workspace_worker_discovery.enabled); + assert!(!feature.objective.enabled); + assert!(!feature.manage_workdir.enabled); + assert!(!feature.ticket.enabled); + assert!(!feature.ticket.authoring); + assert!(!feature.ticket.thread); + assert!(!feature.ticket.intake); + assert!(!feature.ticket.workflow); + assert!(!feature.merge_request.show); + assert!(!feature.merge_request.open); + assert!(!feature.merge_request.review); + assert!(!feature.merge_request.readiness_check); + assert!(!feature.merge_request.complete); + assert!(!feature.orchestration.enabled); + assert_eq!( + manifest.web.as_ref().and_then(|web| web.enabled), + Some(false) + ); + } + #[test] fn profile_rejects_workspace_memory_snapshot_authority_fields() { let tmp = TempDir::new().unwrap(); diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index f3c707ff4..cafe6545d 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -781,6 +781,39 @@ mod tests { )) } + fn repository_error_client(status: u16, code: &str) -> Arc { + use std::io::{Read, Write}; + use std::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let body = serde_json::json!({ + "error": "request rejected", + "message": format!("{code}: retry after reread"), + "diagnostics": [{ + "code": code, + "severity": "error", + "message": format!("{code}: retry after reread"), + }], + }) + .to_string(); + std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut request = [0_u8; 4096]; + let _ = stream.read(&mut request).unwrap(); + let response = format!( + "HTTP/1.1 {status} rejected\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + body.len(), + body + ); + stream.write_all(response.as_bytes()).unwrap(); + }); + Arc::new(crate::worker::TestWorkspaceHttpClient::new( + "workspace", + format!("http://{addr}"), + )) + } + fn tool_names(definitions: Vec) -> Vec { let mut names = definitions .into_iter() @@ -991,6 +1024,39 @@ mod tests { assert!(names.contains(&"MemoryStagingClose".to_string())); } + #[tokio::test] + async fn subjektiv_candidate_conflicts_survive_workspace_http_transport() { + let input = serde_json::json!({ + "request_id": "decision-request-1", + "candidate_id": "candidate-1", + "reason": "candidate is already covered", + "decision": { + "kind": "close", + "action": "already_covered", + "affected_memory": [], + }, + }) + .to_string(); + for (status, code) in [ + (409, "candidate_decision_conflict"), + (403, "subject_scope_mismatch"), + ] { + let tool = SubjectConsolidationTool { + client: repository_error_client(status, code), + operation: SubjectConsolidationOperation::DecideCandidate, + }; + let output = tool + .execute(&input, ToolExecutionContext::direct()) + .await + .expect("typed conflicts are model-visible retry output"); + let content: serde_json::Value = + serde_json::from_str(output.content.as_deref().unwrap()).unwrap(); + assert_eq!(content["status"], "error"); + assert_eq!(content["error"]["code"], code); + assert!(output.summary.contains("requires reread")); + } + } + #[test] fn subjektiv_consolidation_has_atomic_decision_without_host_scope_fields() { let tools = workspace_http_subjektiv_consolidation_tools(test_client()); diff --git a/crates/worker/src/feature/builtin/memory_lifecycle.rs b/crates/worker/src/feature/builtin/memory_lifecycle.rs index da1bb9fe6..3bdfc9759 100644 --- a/crates/worker/src/feature/builtin/memory_lifecycle.rs +++ b/crates/worker/src/feature/builtin/memory_lifecycle.rs @@ -79,6 +79,15 @@ struct MemoryLifecycleTask { event_tx: Option>, } +fn memory_lifecycle_requested( + lifecycle_enabled: bool, + config: &manifest::ResolvedMemoryFeatureConfig, +) -> bool { + lifecycle_enabled + && config.profile.enabled + && (config.profile.extraction.enabled || config.profile.consolidation.request_enabled) +} + impl MemoryLifecycleFeature { #[allow(clippy::too_many_arguments)] pub(crate) fn from_resolved_config( @@ -93,10 +102,7 @@ impl MemoryLifecycleFeature { workspace_context: WorkerWorkspaceContext, event_tx: Option>, ) -> std::io::Result> { - if !lifecycle_enabled - || !config.profile.enabled - || (!config.profile.extraction.enabled && !config.profile.consolidation.request_enabled) - { + if !memory_lifecycle_requested(lifecycle_enabled, &config) { return Ok(None); } config @@ -1365,6 +1371,26 @@ permission = "write" config } + #[test] + fn resolved_subjektiv_consolidation_profile_does_not_install_global_memory_lifecycle() { + let workspace = tempfile::tempdir().unwrap(); + let resolved = manifest::ProfileResolver::new() + .with_workspace_base(workspace.path()) + .resolve( + &manifest::ProfileSelector::source_named( + manifest::ProfileRegistrySource::Builtin, + "subjektiv-memory-consolidation", + ), + manifest::ProfileResolveOptions::with_worker_name("subject-consolidator"), + ) + .unwrap(); + + assert!(!memory_lifecycle_requested( + true, + &resolved.manifest.feature.memory + )); + } + fn test_task( capture: CommittedSessionCapture, client: Box, diff --git a/crates/workspace-server/src/hosts.rs b/crates/workspace-server/src/hosts.rs index aa903a20f..c197071cc 100644 --- a/crates/workspace-server/src/hosts.rs +++ b/crates/workspace-server/src/hosts.rs @@ -5370,6 +5370,7 @@ fn embedded_profile_label(profile: &ProfileSelector) -> Option { const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation"; pub(crate) const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation"; +pub(crate) const SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX: &str = "subjektiv-consolidation:"; const WORKSPACE_ORCHESTRATOR_PROFILE: &str = "orchestrator"; pub(crate) const WORKSPACE_ORCHESTRATOR_SINGLETON_KEY: &str = "workspace-orchestrator"; @@ -5377,7 +5378,7 @@ pub(crate) fn is_reserved_internal_worker_singleton_key(key: &str) -> bool { matches!( key, MEMORY_CONSOLIDATION_SINGLETON_KEY | WORKSPACE_ORCHESTRATOR_SINGLETON_KEY - ) + ) || key.starts_with(SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX) } struct WorkerDisplayMetadata { diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index dd8e703fd..f07ff40ff 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -16544,7 +16544,9 @@ async fn scoped_memory_backend_operation( } const SUBJEKTIV_SINGLETON_PREFIX: &str = "subjektiv:"; -const SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX: &str = "subjektiv-consolidation:"; +const SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX: &str = + crate::hosts::SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX; +const SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE: &str = "subjektiv-memory-consolidation"; fn open_subjektiv_store(api: &WorkspaceApi) -> ApiResult { crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration) @@ -16591,6 +16593,38 @@ fn subjektiv_singleton_key(subject_id: &str) -> ApiResult { Ok(key) } +fn is_dedicated_subjektiv_consolidation_worker( + worker: &InternalWorkerSummary, + workspace_id: &str, + singleton_key: &str, +) -> bool { + let profile_matches = worker.profile.as_deref().is_some_and(|profile| { + profile.strip_prefix("builtin:").unwrap_or(profile) + == SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE + }); + let profile_tag = format!("profile:{SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE}"); + worker.workspace.workspace_id.as_deref() == Some(workspace_id) + && worker.singleton_key.as_deref() == Some(singleton_key) + && singleton_key.starts_with(SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX) + && profile_matches + && worker.tags.iter().any(|tag| tag == &profile_tag) +} + +fn require_dedicated_subjektiv_consolidation_worker( + worker: &InternalWorkerSummary, + workspace_id: &str, + singleton_key: &str, +) -> ApiResult<()> { + if is_dedicated_subjektiv_consolidation_worker(worker, workspace_id, singleton_key) { + Ok(()) + } else { + Err(Error::WorkspacePermissionDenied( + "subject Memory consolidation requires a Backend-managed dedicated Worker".to_string(), + ) + .into()) + } +} + fn subjektiv_subject_scope( api: &WorkspaceApi, workspace_id: &str, @@ -16616,6 +16650,14 @@ fn subjektiv_subject_scope( ) })?; let (subject_id, authority) = subjektiv_scope_from_singleton_key(&lease.key)?; + if authority == SubjektivWorkerAuthority::Consolidation { + let mut summary = api + .runtime + .worker(&worker) + .map_err(|error| error.into_error())?; + summary.singleton_key = Some(lease.key.clone()); + require_dedicated_subjektiv_consolidation_worker(&summary, workspace_id, &lease.key)?; + } Ok((subject_id.to_string(), worker, authority)) } @@ -18187,7 +18229,6 @@ fn subjektiv_store_error(error: crate::subjektiv::SubjektivError) -> Error { } const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation"; -const SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE: &str = "subjektiv-memory-consolidation"; const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation"; async fn scoped_subjektiv_memory_consolidation( @@ -18324,6 +18365,11 @@ fn try_reuse_subject_consolidation_worker( .worker(&owner.worker) .map_err(|error| error.into_error())?; worker.singleton_key = Some(owner.key); + require_dedicated_subjektiv_consolidation_worker( + &worker, + &api.config.workspace_id, + singleton_key, + )?; if worker.state != "idle" { return Ok(Some(MemoryConsolidationOutput { status: "skipped_existing_not_idle".into(), @@ -29621,25 +29667,32 @@ impl From for ApiError { severity: HostDiagnosticSeverity::Error, message: sanitize_backend_error(&ticket_error.to_string()), }], - Error::RepositoryConflict(message) => subjektiv_conflict_code(message) - .map(|code| { - vec![RuntimeDiagnostic { - code: code.to_string(), - severity: HostDiagnosticSeverity::Error, - message: sanitize_backend_error(message), - }] - }) - .unwrap_or_default(), + Error::RepositoryConflict(message) | Error::WorkspacePermissionDenied(message) => { + subjektiv_diagnostic_code(message) + .map(|code| { + vec![RuntimeDiagnostic { + code: code.to_string(), + severity: HostDiagnosticSeverity::Error, + message: sanitize_backend_error(message), + }] + }) + .unwrap_or_default() + } _ => Vec::new(), }; Self { error, diagnostics } } } -fn subjektiv_conflict_code(message: &str) -> Option<&'static str> { - ["revision_conflict", "stale_cursor"] - .into_iter() - .find(|code| message.starts_with(code)) +fn subjektiv_diagnostic_code(message: &str) -> Option<&'static str> { + [ + "revision_conflict", + "stale_cursor", + "candidate_decision_conflict", + "subject_scope_mismatch", + ] + .into_iter() + .find(|code| message.starts_with(code)) } impl ApiError { @@ -43413,31 +43466,42 @@ mod tests { let dir = tempfile::tempdir().unwrap(); init_clean_git_workspace(dir.path()); let api = test_api(dir.path()).await; - let request = |requested_worker_name: &str, - intent: server_api::RuntimeWorkerSpawnIntent, - profile: &str| server_api::RuntimeWorkerSpawnRequest { - requested_worker_name: Some(requested_worker_name.to_string()), - intent, - singleton_key: Some(crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY.to_string()), - acceptance: server_api::RuntimeWorkerSpawnAcceptanceRequirement::RunAccepted { - expected_segments: 0, - }, - profile: server_api::RuntimeProfileSelector::Builtin(profile.to_string()), - ticket_assignment: None, - initial_submit: Vec::new(), - workdir_attachment_requests: Vec::new(), - }; + let request = + |requested_worker_name: &str, + intent: server_api::RuntimeWorkerSpawnIntent, + profile: &str, + singleton_key: &str| server_api::RuntimeWorkerSpawnRequest { + requested_worker_name: Some(requested_worker_name.to_string()), + intent, + singleton_key: Some(singleton_key.to_string()), + acceptance: server_api::RuntimeWorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 0, + }, + profile: server_api::RuntimeProfileSelector::Builtin(profile.to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachment_requests: Vec::new(), + }; + let subject_consolidation_key = "subjektiv-consolidation:subject-forged"; for candidate in [ request( "Generic key squatter", server_api::RuntimeWorkerSpawnIntent::WorkspaceCompanion, "builtin:companion", + crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY, ), request( crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY, server_api::RuntimeWorkerSpawnIntent::WorkspaceOrchestrator, "builtin:orchestrator", + crate::hosts::WORKSPACE_ORCHESTRATOR_SINGLETON_KEY, + ), + request( + SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE, + server_api::RuntimeWorkerSpawnIntent::WorkspaceOrchestrator, + "builtin:subjektiv-memory-consolidation", + subject_consolidation_key, ), ] { let error = scoped_create_runtime_worker( @@ -43461,6 +43525,12 @@ mod tests { .unwrap() .is_none() ); + assert!( + api.store + .current_worker_singleton_owner(TEST_WORKSPACE_ID, subject_consolidation_key) + .unwrap() + .is_none() + ); assert!(find_workspace_orchestrator(&api).is_none()); } @@ -51157,6 +51227,118 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3); serde_json::from_slice(&bytes).unwrap() } + #[tokio::test] + async fn subjektiv_consolidation_scope_rejects_non_dedicated_singleton_owner() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + let spawn = |profile: &str, key: &str| WorkerSpawnRequest { + requested_worker_name: Some(profile.to_string()), + singleton_key: Some(key.to_string()), + intent: WorkerSpawnIntent::WorkspaceOrchestrator, + acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 0, + }, + profile: ProfileSelector::Builtin(profile.to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachment_requests: Vec::new(), + resolved_workdir_attachment_requests: Vec::new(), + resolved_workdir_attachments: Vec::new(), + resolved_config_bundle: None, + resolved_worker_observation_enabled: false, + resolved_worker_observation_grants: Vec::new(), + resolved_workspace_api: None, + resolved_memory_settings: None, + resolved_control_operation: None, + }; + + let dedicated_key = "subjektiv-consolidation:dedicated-subject"; + let dedicated = api + .spawn_workspace_worker( + EMBEDDED_WORKER_RUNTIME_ID, + spawn(SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE, dedicated_key), + ) + .unwrap() + .worker + .unwrap(); + assert!(is_dedicated_subjektiv_consolidation_worker( + &dedicated, + TEST_WORKSPACE_ID, + dedicated_key + )); + let dedicated_context = server_api::ServerRequestContext { + actor: None, + worker_source: None, + runtime_source: Some(server_api::ServerRuntimeSource { + runtime_id: dedicated.worker.runtime_id.clone(), + worker_id: Some(dedicated.worker.worker_id.clone()), + }), + origin: None, + transport_headers: Vec::new(), + }; + assert_eq!( + subjektiv_subject_scope(&api, TEST_WORKSPACE_ID, &dedicated_context) + .unwrap() + .2, + SubjektivWorkerAuthority::Consolidation + ); + + let forged_key = "subjektiv-consolidation:forged-subject"; + let forged = api + .spawn_workspace_worker(EMBEDDED_WORKER_RUNTIME_ID, spawn("companion", forged_key)) + .unwrap() + .worker + .unwrap(); + let forged_context = server_api::ServerRequestContext { + actor: None, + worker_source: None, + runtime_source: Some(server_api::ServerRuntimeSource { + runtime_id: forged.worker.runtime_id.clone(), + worker_id: Some(forged.worker.worker_id.clone()), + }), + origin: None, + transport_headers: Vec::new(), + }; + let error = subjektiv_subject_scope(&api, TEST_WORKSPACE_ID, &forged_context).unwrap_err(); + assert_eq!(error.into_response().status(), StatusCode::FORBIDDEN); + assert!( + try_reuse_subject_consolidation_worker( + &api, + forged_key, + "process pending subject Memory", + 1, + 1, + ) + .is_err() + ); + } + + #[test] + fn subjektiv_http_errors_emit_typed_retry_diagnostics() { + for (error, code, status) in [ + ( + Error::RepositoryConflict( + "candidate_decision_conflict: candidate already resolved".to_string(), + ), + "candidate_decision_conflict", + StatusCode::CONFLICT, + ), + ( + Error::WorkspacePermissionDenied( + "subject_scope_mismatch: candidate belongs to another subject".to_string(), + ), + "subject_scope_mismatch", + StatusCode::FORBIDDEN, + ), + ] { + assert_eq!(api_error_status(&error), status); + let response = ApiError::from(error).into_repository_api_error(); + assert_eq!(response.diagnostics.len(), 1); + assert_eq!(response.diagnostics[0].code, code); + } + } + #[test] fn subjektiv_singleton_scope_is_explicit_and_opaque() { assert_eq!( diff --git a/resources/profiles/subjektiv-memory-consolidation.dcdl b/resources/profiles/subjektiv-memory-consolidation.dcdl index 676610f7e..f7fbba83e 100644 --- a/resources/profiles/subjektiv-memory-consolidation.dcdl +++ b/resources/profiles/subjektiv-memory-consolidation.dcdl @@ -4,14 +4,40 @@ import "./base.dcdl" // { scope = "workspace_read"; model = { ref = "codex-oauth/gpt-5.6-luna"; }; engine = { reasoning = "medium"; }; + web = { enabled = false; }; feature = { task = { enabled = false; }; - memory = { enabled = true; staging_tools = true; }; + memory = { + enabled = true; + staging_tools = true; + resident = { inject_summary = false; }; + extraction = { enabled = false; }; + consolidation = { request_enabled = false; }; + }; + subjektiv = { enabled = false; extraction = { enabled = false; }; }; web = { enabled = false; }; + image = { enabled = false; }; sub_worker = { enabled = false; }; - worker = { enabled = false; }; + flow = { enabled = false; }; + worker = { enabled = false; direct_spawn = false; }; + workspace_worker_discovery = { enabled = false; }; objective = { enabled = false; }; - ticket = { enabled = false; thread = false; }; + manage_workdir = { enabled = false; }; + ticket = { + enabled = false; + authoring = false; + thread = false; + intake = false; + workflow = false; + }; + merge_request = { + show = false; + open = false; + review = false; + readiness_check = false; + complete = false; + }; + orchestration = { enabled = false; }; }; }