From 046e606a38b1a48958cb82c2c5d7b08e4ce3f709 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 2 Oct 2026 10:46:12 +0900 Subject: [PATCH 1/2] feat: connect subject memory consolidation --- crates/manifest/src/builtin_profile.rs | 7 + crates/manifest/src/profile.rs | 21 +- crates/server-api/src/lib.rs | 182 +++++ crates/worker/src/feature/builtin/memory.rs | 281 ++++++- .../src/feature/builtin/memory_lifecycle.rs | 56 +- crates/worker/src/prompt/catalog.rs | 14 + crates/workspace-server/src/server.rs | 571 ++++++++++++- crates/workspace-server/src/subjektiv.rs | 759 ++++++++++++++++-- openapi/server-api.json | 2 +- .../subjektiv-memory-consolidation.dcdl | 17 + resources/prompts/catalog.dcdl | 2 + .../subjektiv_memory_consolidation_system.md | 41 + .../src/lib/generated/companion-api.ts | 2 +- .../src/lib/generated/repository-api.ts | 2 +- .../src/lib/generated/runtime-api.ts | 2 +- web/workspace/src/lib/generated/ticket-api.ts | 2 +- .../src/lib/generated/worker-launch-api.ts | 2 +- 17 files changed, 1857 insertions(+), 106 deletions(-) create mode 100644 resources/profiles/subjektiv-memory-consolidation.dcdl create mode 100644 resources/prompts/internal/subjektiv_memory_consolidation_system.md diff --git a/crates/manifest/src/builtin_profile.rs b/crates/manifest/src/builtin_profile.rs index b2d291787..e19b44956 100644 --- a/crates/manifest/src/builtin_profile.rs +++ b/crates/manifest/src/builtin_profile.rs @@ -108,6 +108,13 @@ pub const BUILTIN_PROFILE_RESOURCES: &[BuiltinProfileResource] = &[ description: "Internal Memory consolidation service.", imports: BASE_IMPORT, }, + BuiltinProfileResource { + selector: Some("builtin:subjektiv-memory-consolidation"), + path: "profiles/subjektiv-memory-consolidation.dcdl", + source: include_str!("../../../resources/profiles/subjektiv-memory-consolidation.dcdl"), + description: "Delegated subject Memory consolidation service.", + imports: BASE_IMPORT, + }, ]; #[derive(Debug, Clone, PartialEq, Eq)] diff --git a/crates/manifest/src/profile.rs b/crates/manifest/src/profile.rs index 2e59e18ff..c504ce154 100644 --- a/crates/manifest/src/profile.rs +++ b/crates/manifest/src/profile.rs @@ -1265,6 +1265,7 @@ mod tests { "coder", "reviewer", "memory-consolidation", + "subjektiv-memory-consolidation", ] { let role = resolve(name); assert_eq!( @@ -1311,15 +1312,17 @@ mod tests { ); } - let consolidation = resolve("memory-consolidation"); - assert_eq!( - consolidation.manifest.model.ref_.as_deref(), - Some("codex-oauth/gpt-5.6-luna") - ); - assert_eq!( - consolidation.manifest.engine.reasoning, - Some(ReasoningControl::Effort(ReasoningEffort::Medium)) - ); + for name in ["memory-consolidation", "subjektiv-memory-consolidation"] { + let consolidation = resolve(name); + assert_eq!( + consolidation.manifest.model.ref_.as_deref(), + Some("codex-oauth/gpt-5.6-luna") + ); + assert_eq!( + consolidation.manifest.engine.reasoning, + Some(ReasoningControl::Effort(ReasoningEffort::Medium)) + ); + } } #[test] diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index d3aea23ba..089d7c65c 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -1190,6 +1190,19 @@ pub trait ServerApi { #[path] workspace_id: String, #[body] request: SubjektivMemoryBackendRequest, ) -> Result; + #[post( + "/api/w/{workspace_id}/subjektiv/consolidation", + status = 200, + error_status = 400, + additional_error_statuses = [401, 403, 404, 409, 500], + openapi = false, + )] + async fn subjektiv_memory_consolidation( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[body] request: MemoryConsolidateStagingRequest, + ) -> Result; #[get( "/api/w/{workspace_id}/skills", @@ -4650,6 +4663,168 @@ pub struct SubjektivMemoryStageExplicitRequest { pub proposal: Option, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryCandidateListRequest { + #[serde(default, skip_serializing_if = "Option::is_none")] + #[schemars(range(min = 1, max = 100))] + pub limit: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryCandidateReadRequest { + pub candidate_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryRevisionProposalMetadata { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub expected_revision: u64, + pub intent: SubjektivMemoryRevisionIntent, + pub change_reason: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryCandidate { + pub candidate_id: String, + pub kind: memory::extract::CandidateKind, + pub claim: String, + pub why_useful: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub staleness: Option, + pub source: memory::schema::SourceRef, + #[serde(default)] + pub evidence: Vec, + #[serde(default)] + pub source_refs: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub revision_proposal: Option, + pub created_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryCandidateSummary { + pub candidate_id: String, + pub kind: memory::extract::CandidateKind, + pub claim: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub revision_proposal: Option, + pub created_at: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryCandidateListResponse { + pub items: Vec, + pub has_more: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum SubjektivMemoryApplyTarget { + Create, + Revise { + memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + expected_revision: u64, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryDraft { + pub kind: memory::extract::CandidateKind, + pub state: SubjektivMemoryState, + pub claim: String, + pub body_md: String, + pub why_useful: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub staleness: Option, + #[serde(default)] + pub derived_from: Vec, + pub change_reason: String, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum SubjektivMemoryCandidateCloseAction { + Discarded, + Invalid, + Duplicate, + AlreadyCovered, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum SubjektivMemoryCandidateDecision { + Apply { + target: SubjektivMemoryApplyTarget, + memory: SubjektivMemoryDraft, + }, + Close { + action: SubjektivMemoryCandidateCloseAction, + #[serde(default)] + affected_memory: Vec, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryCandidateDecisionRequest { + pub request_id: String, + pub candidate_id: String, + pub reason: String, + pub decision: SubjektivMemoryCandidateDecision, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum SubjektivMemoryAffectedOperation { + Create, + Revise, + Reference, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryAffectedRef { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub revision: u64, + pub operation: SubjektivMemoryAffectedOperation, +} + +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum SubjektivMemoryCandidateResolutionAction { + Applied, + Discarded, + Invalid, + Duplicate, + AlreadyCovered, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivMemoryCandidateDecisionResponse { + pub request_id: String, + pub candidate_id: String, + pub action: SubjektivMemoryCandidateResolutionAction, + pub reason: String, + #[serde(default)] + pub affected_memory: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memory: Option, + #[schemars(range(min = 0, max = 9_007_199_254_740_991_u64))] + pub store_revision: u64, + pub surface_dirty: bool, +} + #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(tag = "operation", content = "input", rename_all = "snake_case")] pub enum SubjektivMemoryBackendOperation { @@ -4659,6 +4834,9 @@ pub enum SubjektivMemoryBackendOperation { ValidateProposal(SubjektivMemoryValidateProposalRequest), ReceiptStatus(SubjektivMemoryReceiptStatusRequest), StageExplicit(SubjektivMemoryStageExplicitRequest), + ListCandidates(SubjektivMemoryCandidateListRequest), + ReadCandidate(SubjektivMemoryCandidateReadRequest), + DecideCandidate(SubjektivMemoryCandidateDecisionRequest), } #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] @@ -4824,6 +5002,9 @@ pub enum SubjektivMemoryBackendResponse { ProposalValidated(SubjektivMemoryProposalValidationResponse), ReceiptStatus(SubjektivMemoryReceiptStatusResponse), Staged(SubjektivMemoryStageExplicitResponse), + Candidates(SubjektivMemoryCandidateListResponse), + Candidate(SubjektivMemoryCandidate), + CandidateDecided(SubjektivMemoryCandidateDecisionResponse), } pub const WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES: usize = 128; @@ -11994,6 +12175,7 @@ mod openapi_artifact_tests { "subjektiv_stage_candidate", "subjektiv_record_session", "subjektiv_memory_backend", + "subjektiv_memory_consolidation", "workspace_worker_discovery", "workspace_worker_remove", ]; diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index 3198b5ea6..f3c707ff4 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -84,7 +84,14 @@ impl dyn WorkspaceClient + '_ { &self, operation: MemoryConsolidateStagingOperation, ) -> Result { - execute_memory_consolidation(self, operation).await + execute_memory_consolidation(self, operation, false).await + } + + pub async fn request_subjektiv_memory_staging_consolidation( + &self, + operation: MemoryConsolidateStagingOperation, + ) -> Result { + execute_memory_consolidation(self, operation, true).await } } @@ -125,6 +132,7 @@ async fn execute_memory_backend( async fn execute_memory_consolidation( client: &dyn WorkspaceClient, operation: MemoryConsolidateStagingOperation, + subjektiv: bool, ) -> Result { let workspace_id = client @@ -137,7 +145,14 @@ async fn execute_memory_consolidation( })?; let response = client.execute(WorkspaceRequest::json( WorkspaceRequestMethod::Post, - format!("/api/w/{workspace_id}/memory/consolidation"), + format!( + "/api/w/{workspace_id}/{}", + if subjektiv { + "subjektiv/consolidation" + } else { + "memory/consolidation" + } + ), serde_json::to_string(&server_api::MemoryConsolidateStagingRequest { force: operation.force, })?, @@ -202,6 +217,205 @@ pub fn workspace_http_memory_tools(client: Arc) -> Vec, +) -> Vec { + use SubjectConsolidationOperation as Operation; + [ + ( + "SubjektivMemoryQuery", + "Search current Memory revisions for the delegated subject.", + schema_for::(), + Operation::Query, + ), + ( + "SubjektivMemoryRead", + "Read one current or historical Memory revision for the delegated subject.", + schema_for::(), + Operation::Read, + ), + ( + "SubjektivMemoryListRevisions", + "List immutable revisions of one Memory for the delegated subject.", + schema_for::(), + Operation::ListRevisions, + ), + ( + "MemoryStagingList", + "List pending staging candidates for the delegated subject.", + schema_for::(), + Operation::ListCandidates, + ), + ( + "MemoryStagingRead", + "Read one pending candidate, including typed revision proposal metadata and provenance.", + schema_for::(), + Operation::ReadCandidate, + ), + ( + "MemoryApplyCandidate", + "Atomically apply or reject one subject candidate. A request_id is idempotent; application revalidates exact revision proposal metadata and returns concrete affected revision refs.", + schema_for::(), + Operation::DecideCandidate, + ), + ] + .into_iter() + .map(|(name, description, schema, operation)| { + let client = Arc::clone(&client); + Arc::new(move || { + ( + ToolMeta::new(name) + .description(description) + .input_schema(schema.clone()), + Arc::new(SubjectConsolidationTool { + client: Arc::clone(&client), + operation, + }) as Arc, + ) + }) as ToolDefinition + }) + .collect() +} + +#[derive(Clone, Copy)] +enum SubjectConsolidationOperation { + Query, + Read, + ListRevisions, + ListCandidates, + ReadCandidate, + DecideCandidate, +} + +struct SubjectConsolidationTool { + client: Arc, + operation: SubjectConsolidationOperation, +} + +#[async_trait] +impl Tool for SubjectConsolidationTool { + async fn execute( + &self, + input_json: &str, + _ctx: ToolExecutionContext, + ) -> Result { + use SubjectConsolidationOperation as Operation; + let operation = match self.operation { + Operation::Query => { + server_api::SubjektivMemoryBackendOperation::Query(parse_input(input_json)?) + } + Operation::Read => { + server_api::SubjektivMemoryBackendOperation::Read(parse_input(input_json)?) + } + Operation::ListRevisions => { + server_api::SubjektivMemoryBackendOperation::ListRevisions(parse_input(input_json)?) + } + Operation::ListCandidates => { + server_api::SubjektivMemoryBackendOperation::ListCandidates(parse_input( + input_json, + )?) + } + Operation::ReadCandidate => { + server_api::SubjektivMemoryBackendOperation::ReadCandidate(parse_input(input_json)?) + } + Operation::DecideCandidate => { + server_api::SubjektivMemoryBackendOperation::DecideCandidate(parse_input( + input_json, + )?) + } + }; + let workspace_id = self.client.workspace_id().ok_or_else(|| { + ToolError::ExecutionFailed( + "subjektiv consolidation requires Workspace authority".into(), + ) + })?; + let response = self + .client + .execute(WorkspaceRequest::json( + WorkspaceRequestMethod::Post, + format!("/api/w/{workspace_id}/subjektiv/memory"), + serde_json::to_string(&server_api::SubjektivMemoryBackendRequest { operation }) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?, + )) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + if !response.is_success() { + let parsed = + serde_json::from_str::(&response.body).ok(); + if let Some((code, message)) = parsed.as_ref().and_then(|error| { + error + .diagnostics + .iter() + .find(|diagnostic| { + matches!( + diagnostic.code.as_str(), + "revision_conflict" + | "candidate_decision_conflict" + | "subject_scope_mismatch" + ) + }) + .map(|diagnostic| (diagnostic.code.clone(), error.message.clone())) + }) { + return Ok(ToolOutput { + summary: format!("Candidate decision requires reread: {code}."), + content: Some( + serde_json::json!({ + "status": "error", + "error": { "code": code, "message": message } + }) + .to_string(), + ), + attachments: Vec::new(), + }); + } + let detail = parsed.map(|error| error.message).unwrap_or(response.body); + return Err(if matches!(response.status, 400 | 404 | 409 | 422) { + ToolError::InvalidArgument(detail) + } else { + ToolError::ExecutionFailed(detail) + }); + } + let response: server_api::SubjektivMemoryBackendResponse = + serde_json::from_str(&response.body) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; + let summary = match &response { + server_api::SubjektivMemoryBackendResponse::Query(value) => { + format!("Found {} subject Memory item(s).", value.items.len()) + } + server_api::SubjektivMemoryBackendResponse::Read(value) => { + format!( + "Read Memory {} revision {}.", + value.memory_id, value.revision + ) + } + server_api::SubjektivMemoryBackendResponse::ListRevisions(value) => { + format!("Listed {} revision(s).", value.items.len()) + } + server_api::SubjektivMemoryBackendResponse::Candidates(value) => { + format!("Listed {} pending candidate(s).", value.items.len()) + } + server_api::SubjektivMemoryBackendResponse::Candidate(value) => { + format!("Read pending candidate {}.", value.candidate_id) + } + server_api::SubjektivMemoryBackendResponse::CandidateDecided(value) => { + format!("Candidate {} decision committed.", value.candidate_id) + } + _ => { + return Err(ToolError::ExecutionFailed( + "unexpected subjektiv consolidation response".into(), + )); + } + }; + Ok(ToolOutput { + summary, + content: Some( + serde_json::to_string_pretty(&response) + .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?, + ), + attachments: Vec::new(), + }) + } +} + pub fn workspace_http_memory_consolidation_tools( client: Arc, ) -> Vec { @@ -407,16 +621,21 @@ impl MemoryFeatureInstallPlan { prompts: Arc, profile: Option, ) -> std::io::Result> { - let memory_consolidation_worker = profile.as_ref().is_some_and(|snapshot| { - matches!( - &snapshot.source, + let profile_name = profile + .as_ref() + .and_then(|snapshot| match &snapshot.source { manifest::ProfileSource::Registry { source: manifest::ProfileRegistrySource::Builtin, name, .. - } if name == "memory-consolidation" - ) - }); + } => Some(name.as_str()), + _ => None, + }); + let memory_consolidation_worker = matches!( + profile_name, + Some("memory-consolidation" | "subjektiv-memory-consolidation") + ); + let subjektiv_consolidation_worker = profile_name == Some("subjektiv-memory-consolidation"); config .validate_execution() .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?; @@ -450,16 +669,24 @@ impl MemoryFeatureInstallPlan { let system_prompt_override = if memory_consolidation_worker { let language = settings.language; Some( - prompts - .memory_consolidation_system(&language) - .map_err(|error| std::io::Error::other(error.to_string()))?, + if subjektiv_consolidation_worker { + prompts.subjektiv_memory_consolidation_system(&language) + } else { + prompts.memory_consolidation_system(&language) + } + .map_err(|error| std::io::Error::other(error.to_string()))?, ) } else { None }; + let module = if subjektiv_consolidation_worker { + MemoryToolsFeature::from_tools(workspace_http_subjektiv_consolidation_tools(client)) + } else { + MemoryToolsFeature::new(client, config.profile.staging_tools) + }; Ok(Some(Self { - module: MemoryToolsFeature::new(client, config.profile.staging_tools), + module, resident_summary_source, system_prompt_override, resolved_config: config, @@ -481,6 +708,10 @@ impl MemoryToolsFeature { }; Self { tools } } + + fn from_tools(tools: Vec) -> Self { + Self { tools } + } } impl FeatureModule for MemoryToolsFeature { @@ -759,4 +990,30 @@ mod tests { assert!(names.contains(&"MemoryStagingRead".to_string())); assert!(names.contains(&"MemoryStagingClose".to_string())); } + + #[test] + fn subjektiv_consolidation_has_atomic_decision_without_host_scope_fields() { + let tools = workspace_http_subjektiv_consolidation_tools(test_client()); + let names = tool_names(tools.clone()); + assert_eq!( + names, + vec![ + "MemoryApplyCandidate", + "MemoryStagingList", + "MemoryStagingRead", + "SubjektivMemoryListRevisions", + "SubjektivMemoryQuery", + "SubjektivMemoryRead", + ] + ); + let decision = tool_meta(tools, "MemoryApplyCandidate").to_string(); + assert!(decision.contains("request_id")); + assert!(decision.contains("expected_revision")); + assert!(decision.contains("already_covered")); + assert!(!decision.contains("subject_id")); + assert!(!decision.contains("workspace_id")); + assert!(!decision.contains("runtime_id")); + assert!(!decision.contains("worker_id")); + assert!(!decision.contains("source_candidate_ids")); + } } diff --git a/crates/worker/src/feature/builtin/memory_lifecycle.rs b/crates/worker/src/feature/builtin/memory_lifecycle.rs index 7fb6863d9..da1bb9fe6 100644 --- a/crates/worker/src/feature/builtin/memory_lifecycle.rs +++ b/crates/worker/src/feature/builtin/memory_lifecycle.rs @@ -164,7 +164,7 @@ impl SubjektivLifecycleFeature { workspace_context: WorkerWorkspaceContext, event_tx: Option>, ) -> std::io::Result> { - if !lifecycle_enabled || !config.profile.enabled || !config.profile.extraction.enabled { + if !lifecycle_enabled || !config.profile.enabled { return Ok(None); } config @@ -712,8 +712,16 @@ impl FeatureBackgroundTask for MemoryLifecycleTask { }; if !cancellation.is_cancelled() { context.generation_fence.ensure_current()?; - if self.config.profile.consolidation.request_enabled { - self.request_consolidation().await; + match self.target { + ExtractionTarget::WorkspaceMemory + if self.config.profile.consolidation.request_enabled => + { + self.request_consolidation(false).await; + } + ExtractionTarget::Subjektiv => { + self.request_consolidation(true).await; + } + ExtractionTarget::WorkspaceMemory => {} } } extraction @@ -750,20 +758,24 @@ impl MemoryLifecycleTask { Ok(()) } - async fn request_consolidation(&self) { + async fn request_consolidation(&self, subjektiv: bool) { let audit = WorkerAuditBase::new( memory::audit::AuditWorker::MemoryConsolidation, memory::audit::AuditTrigger::StagingBacklog, Some(model_audit_from_manifest(&self.manifest.model)), ) .with_memory_settings(&self.config); - match self - .workspace_client - .request_memory_staging_consolidation( - memory::backend::MemoryConsolidateStagingOperation { force: false }, - ) - .await - { + let operation = memory::backend::MemoryConsolidateStagingOperation { force: false }; + let request = if subjektiv { + self.workspace_client + .request_subjektiv_memory_staging_consolidation(operation) + .await + } else { + self.workspace_client + .request_memory_staging_consolidation(operation) + .await + }; + match request { Ok(output) => { tracing::debug!( status = output.status.as_str(), @@ -1079,6 +1091,7 @@ mod tests { ) -> Result { let is_subjektiv_stage = request.path.ends_with("/subjektiv/staging"); let is_subjektiv_session = request.path.ends_with("/subjektiv/sessions"); + let is_subjektiv_consolidation = request.path.ends_with("/subjektiv/consolidation"); let is_stage_candidate = request .body .as_deref() @@ -1088,6 +1101,18 @@ mod tests { .as_deref() .is_some_and(|body| body.contains("append_audit")); self.requests.lock().unwrap().push(request); + if is_subjektiv_consolidation { + return Ok(crate::worker::WorkspaceResponse { + status: 200, + body: serde_json::to_string(&server_api::MemoryConsolidationResponse { + status: "started".into(), + summary: "subject consolidation requested".into(), + candidate_count: 1, + total_bytes: 1, + }) + .unwrap(), + }); + } if is_subjektiv_session { return Ok(crate::worker::WorkspaceResponse { status: 200, @@ -1635,8 +1660,15 @@ permission = "write" request .body .as_deref() - .is_none_or(|body| !body.contains("append_audit") && body != "{\"force\":false}") + .is_none_or(|body| !body.contains("append_audit")) })); + assert_eq!( + requests + .iter() + .filter(|request| request.path.ends_with("/subjektiv/consolidation")) + .count(), + 1 + ); } #[tokio::test] diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index bba265bdd..13b32e874 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -94,6 +94,7 @@ pub enum WorkerPrompt { CompactSystem, MemoryExtractSystem, MemoryConsolidationSystem, + SubjektivMemoryConsolidationSystem, FlowVerifierSystem, BackendJobSystem, NotifyWrapper, @@ -112,6 +113,9 @@ impl WorkerPrompt { Self::CompactSystem => "internal.compact_system", Self::MemoryExtractSystem => "internal.memory_extract_system", Self::MemoryConsolidationSystem => "internal.memory_consolidation_system", + Self::SubjektivMemoryConsolidationSystem => { + "internal.subjektiv_memory_consolidation_system" + } Self::FlowVerifierSystem => "internal.flow_verifier_system", Self::BackendJobSystem => "internal.backend_job_system", Self::NotifyWrapper => "internal.notify_wrapper", @@ -131,6 +135,7 @@ impl WorkerPrompt { WorkerPrompt::CompactSystem, WorkerPrompt::MemoryExtractSystem, WorkerPrompt::MemoryConsolidationSystem, + WorkerPrompt::SubjektivMemoryConsolidationSystem, WorkerPrompt::FlowVerifierSystem, WorkerPrompt::BackendJobSystem, WorkerPrompt::NotifyWrapper, @@ -449,6 +454,15 @@ impl PromptCatalog { single("language", language), ) } + pub fn subjektiv_memory_consolidation_system( + &self, + language: &str, + ) -> Result { + self.render( + WorkerPrompt::SubjektivMemoryConsolidationSystem, + single("language", language), + ) + } pub fn flow_verifier_system(&self) -> Result { self.render(WorkerPrompt::FlowVerifierSystem, Value::UNDEFINED) } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index ba610e53c..dd8e703fd 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -7166,6 +7166,31 @@ impl server_api::ServerApi for ServerApiContractService { .map_err(ApiError::into_repository_api_error) } + async fn subjektiv_memory_consolidation( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + request: server_api::MemoryConsolidateStagingRequest, + ) -> std::result::Result + { + scoped_subjektiv_memory_consolidation( + State(self.workspace_api()?.clone()), + AxumPath(ScopedWorkspacePath { workspace_id }), + context, + Json(MemoryConsolidateStagingOperation { + force: request.force, + }), + ) + .await + .map(|Json(response)| server_api::MemoryConsolidationResponse { + status: response.status, + summary: response.summary, + candidate_count: response.candidate_count, + total_bytes: response.total_bytes, + }) + .map_err(ApiError::into_repository_api_error) + } + async fn skill_list( &self, workspace_id: String, @@ -16519,25 +16544,42 @@ async fn scoped_memory_backend_operation( } const SUBJEKTIV_SINGLETON_PREFIX: &str = "subjektiv:"; +const SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX: &str = "subjektiv-consolidation:"; fn open_subjektiv_store(api: &WorkspaceApi) -> ApiResult { crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration) .map_err(|error| ApiError::from(Error::Store(error.to_string()))) } -fn subjektiv_subject_id_from_singleton_key(key: &str) -> ApiResult<&str> { - key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) - .filter(|subject_id| { - !subject_id.is_empty() - && !subject_id.chars().any(char::is_control) - && subject_id.trim() == *subject_id - }) - .ok_or_else(|| { - Error::WorkspacePermissionDenied( - "current Worker is not connected to a subjektiv subject".to_string(), - ) - .into() - }) +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum SubjektivWorkerAuthority { + Subject, + Consolidation, +} + +fn subjektiv_scope_from_singleton_key(key: &str) -> ApiResult<(&str, SubjektivWorkerAuthority)> { + let (subject_id, authority) = if let Some(subject_id) = + key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) + { + (subject_id, SubjektivWorkerAuthority::Subject) + } else if let Some(subject_id) = key.strip_prefix(SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX) { + (subject_id, SubjektivWorkerAuthority::Consolidation) + } else { + return Err(Error::WorkspacePermissionDenied( + "current Worker is not connected to a subjektiv subject".to_string(), + ) + .into()); + }; + if subject_id.is_empty() + || subject_id.chars().any(char::is_control) + || subject_id.trim() != subject_id + { + return Err(Error::WorkspacePermissionDenied( + "current Worker is not connected to a subjektiv subject".to_string(), + ) + .into()); + } + Ok((subject_id, authority)) } fn subjektiv_singleton_key(subject_id: &str) -> ApiResult { @@ -16553,7 +16595,7 @@ fn subjektiv_subject_scope( api: &WorkspaceApi, workspace_id: &str, context: &server_api::ServerRequestContext, -) -> ApiResult<(String, RuntimeWorkerRef)> { +) -> ApiResult<(String, RuntimeWorkerRef, SubjektivWorkerAuthority)> { let source = context.runtime_source.as_ref().ok_or_else(|| { Error::WorkspacePermissionDenied( "subjektiv operations require authenticated Runtime-owned source authority".to_string(), @@ -16573,8 +16615,23 @@ fn subjektiv_subject_scope( "current Worker has no keyed singleton ownership".to_string(), ) })?; - let subject_id = subjektiv_subject_id_from_singleton_key(&lease.key)?.to_string(); - Ok((subject_id, worker)) + let (subject_id, authority) = subjektiv_scope_from_singleton_key(&lease.key)?; + Ok((subject_id.to_string(), worker, authority)) +} + +fn require_subjektiv_worker_authority( + actual: SubjektivWorkerAuthority, + expected: SubjektivWorkerAuthority, + operation: &str, +) -> ApiResult<()> { + if actual == expected { + Ok(()) + } else { + Err(Error::WorkspacePermissionDenied(format!( + "subjektiv {operation} is not authorized for this Worker role" + )) + .into()) + } } fn subjektiv_session_attribution( @@ -16583,7 +16640,13 @@ fn subjektiv_session_attribution( context: &server_api::ServerRequestContext, session_id: &str, ) -> ApiResult { - let (subject_id, worker) = subjektiv_subject_scope(api, workspace_id, context)?; + let (subject_id, worker, authority) = subjektiv_subject_scope(api, workspace_id, context)?; + if authority != SubjektivWorkerAuthority::Subject { + return Err(Error::WorkspacePermissionDenied( + "subject Session attribution requires the current subject Worker".into(), + ) + .into()); + } let session_id = session_id.trim(); if session_id.is_empty() || session_id.chars().any(char::is_control) { return Err(Error::InvalidInput("committed session_id is invalid".to_string()).into()); @@ -16824,7 +16887,7 @@ async fn scoped_subjektiv_memory_backend( Json(request): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let (subject_id, _) = subjektiv_subject_scope(&api, &path.workspace_id, &context)?; + let (subject_id, _, authority) = subjektiv_subject_scope(&api, &path.workspace_id, &context)?; let store = open_subjektiv_store(&api)?; let response = match request.operation { server_api::SubjektivMemoryBackendOperation::Query(input) => { @@ -16847,11 +16910,21 @@ async fn scoped_subjektiv_memory_backend( ) } server_api::SubjektivMemoryBackendOperation::ValidateProposal(input) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Subject, + "revision proposal validation", + )?; server_api::SubjektivMemoryBackendResponse::ProposalValidated( subjektiv_memory_validate_proposal(&store, &subject_id, input)?, ) } server_api::SubjektivMemoryBackendOperation::ReceiptStatus(input) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Subject, + "explicit candidate receipt lookup", + )?; validate_subjektiv_receipt_id(&input.receipt_id)?; let candidate = store .staging_candidate(&subject_id, &input.receipt_id) @@ -16869,6 +16942,11 @@ async fn scoped_subjektiv_memory_backend( ) } server_api::SubjektivMemoryBackendOperation::StageExplicit(input) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Subject, + "explicit candidate staging", + )?; // Re-derive the same subject and immutable Session attribution for the // write. The request never carries subject, Worker, Runtime, origin, // or arbitrary source JSON from the model. @@ -16884,10 +16962,264 @@ async fn scoped_subjektiv_memory_backend( attribution, )?) } + server_api::SubjektivMemoryBackendOperation::ListCandidates(input) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Consolidation, + "candidate listing", + )?; + let limit = bounded_subjektiv_limit(input.limit, SUBJEKTIV_QUERY_DEFAULT_LIMIT)?; + let mut candidates = store + .pending_staging_candidates(&subject_id, limit.saturating_add(1)) + .map_err(subjektiv_store_error)?; + let has_more = candidates.len() > limit; + candidates.truncate(limit); + server_api::SubjektivMemoryBackendResponse::Candidates( + server_api::SubjektivMemoryCandidateListResponse { + items: candidates + .into_iter() + .map(subjektiv_candidate_summary) + .collect(), + has_more, + }, + ) + } + server_api::SubjektivMemoryBackendOperation::ReadCandidate(input) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Consolidation, + "candidate read", + )?; + let candidate = store + .staging_candidate(&subject_id, &input.candidate_id) + .map_err(subjektiv_store_error)? + .ok_or_else(|| { + Error::InvalidInput(format!( + "candidate_not_found: pending candidate `{}` was not found", + input.candidate_id + )) + })?; + if store + .staging_resolution(&subject_id, &input.candidate_id) + .map_err(subjektiv_store_error)? + .is_some() + { + return Err(Error::RepositoryConflict(format!( + "candidate_resolved: candidate `{}` is already resolved", + input.candidate_id + )) + .into()); + } + server_api::SubjektivMemoryBackendResponse::Candidate(subjektiv_candidate(candidate)) + } + server_api::SubjektivMemoryBackendOperation::DecideCandidate(input) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Consolidation, + "candidate decision", + )?; + let receipt = store + .decide_candidate(&subject_id, subjektiv_candidate_decision(input)?) + .map_err(subjektiv_store_error)?; + server_api::SubjektivMemoryBackendResponse::CandidateDecided( + subjektiv_candidate_decision_response(receipt), + ) + } }; Ok(Json(response)) } +fn subjektiv_candidate_summary( + candidate: crate::subjektiv::SubjectStagingRecord, +) -> server_api::SubjektivMemoryCandidateSummary { + server_api::SubjektivMemoryCandidateSummary { + candidate_id: candidate.id, + kind: candidate.kind, + claim: candidate.claim, + revision_proposal: candidate.revision_proposal.map(subjektiv_proposal_metadata), + created_at: candidate.created_at, + } +} + +fn subjektiv_candidate( + candidate: crate::subjektiv::SubjectStagingRecord, +) -> server_api::SubjektivMemoryCandidate { + server_api::SubjektivMemoryCandidate { + candidate_id: candidate.id, + kind: candidate.kind, + claim: candidate.claim, + why_useful: candidate.why_useful, + staleness: candidate.staleness, + source: candidate.source, + evidence: candidate.evidence, + source_refs: candidate.source_refs, + revision_proposal: candidate.revision_proposal.map(subjektiv_proposal_metadata), + created_at: candidate.created_at, + } +} + +fn subjektiv_proposal_metadata( + proposal: crate::subjektiv::RevisionProposal, +) -> server_api::SubjektivMemoryRevisionProposalMetadata { + server_api::SubjektivMemoryRevisionProposalMetadata { + memory_id: proposal.memory_id, + expected_revision: proposal.expected_revision, + intent: match proposal.intent { + crate::subjektiv::RevisionProposalIntent::Revise => { + server_api::SubjektivMemoryRevisionIntent::Revise + } + crate::subjektiv::RevisionProposalIntent::Resolve => { + server_api::SubjektivMemoryRevisionIntent::Resolve + } + crate::subjektiv::RevisionProposalIntent::Retract => { + server_api::SubjektivMemoryRevisionIntent::Retract + } + crate::subjektiv::RevisionProposalIntent::Reopen => { + server_api::SubjektivMemoryRevisionIntent::Reopen + } + }, + change_reason: proposal.change_reason, + } +} + +fn subjektiv_candidate_decision( + input: server_api::SubjektivMemoryCandidateDecisionRequest, +) -> ApiResult { + let decision = match input.decision { + server_api::SubjektivMemoryCandidateDecision::Apply { target, memory } => { + let target = match target { + server_api::SubjektivMemoryApplyTarget::Create => { + crate::subjektiv::MemoryRevisionTarget::Create + } + server_api::SubjektivMemoryApplyTarget::Revise { + memory_id, + expected_revision, + } => crate::subjektiv::MemoryRevisionTarget::Revise { + memory_id, + expected_revision, + }, + }; + crate::subjektiv::CandidateDecision::Apply { + target, + draft: crate::subjektiv::MemoryDraft { + kind: memory.kind, + state: domain_memory_state(memory.state), + claim: memory.claim, + body_md: memory.body_md, + why_useful: memory.why_useful, + staleness: memory.staleness, + source_candidate_ids: Vec::new(), + derived_from: memory + .derived_from + .into_iter() + .map(|reference| crate::subjektiv::MemoryRevisionRef { + memory_id: reference.memory_id, + revision: reference.revision, + }) + .collect(), + change_reason: memory.change_reason, + }, + } + } + server_api::SubjektivMemoryCandidateDecision::Close { + action, + affected_memory, + } => crate::subjektiv::CandidateDecision::Close { + action: match action { + server_api::SubjektivMemoryCandidateCloseAction::Discarded => { + crate::subjektiv::StagingResolutionAction::Discarded + } + server_api::SubjektivMemoryCandidateCloseAction::Invalid => { + crate::subjektiv::StagingResolutionAction::Invalid + } + server_api::SubjektivMemoryCandidateCloseAction::Duplicate => { + crate::subjektiv::StagingResolutionAction::Duplicate + } + server_api::SubjektivMemoryCandidateCloseAction::AlreadyCovered => { + crate::subjektiv::StagingResolutionAction::AlreadyCovered + } + }, + affected_memory: affected_memory + .into_iter() + .map(|reference| crate::subjektiv::MemoryRevisionRef { + memory_id: reference.memory_id, + revision: reference.revision, + }) + .collect(), + }, + }; + Ok(crate::subjektiv::CandidateDecisionRequest { + request_id: input.request_id, + candidate_id: input.candidate_id, + reason: input.reason, + decision, + }) +} + +fn subjektiv_candidate_decision_response( + receipt: crate::subjektiv::CandidateDecisionReceipt, +) -> server_api::SubjektivMemoryCandidateDecisionResponse { + let operation = match receipt.operation { + Some(crate::subjektiv::MemoryDecisionOperation::Create) => { + server_api::SubjektivMemoryAffectedOperation::Create + } + Some(crate::subjektiv::MemoryDecisionOperation::Revise) => { + server_api::SubjektivMemoryAffectedOperation::Revise + } + Some(crate::subjektiv::MemoryDecisionOperation::Reference) | None => { + server_api::SubjektivMemoryAffectedOperation::Reference + } + }; + server_api::SubjektivMemoryCandidateDecisionResponse { + request_id: receipt.request_id, + candidate_id: receipt.candidate_id, + action: match receipt.resolution.action { + crate::subjektiv::StagingResolutionAction::Applied => { + server_api::SubjektivMemoryCandidateResolutionAction::Applied + } + crate::subjektiv::StagingResolutionAction::Discarded => { + server_api::SubjektivMemoryCandidateResolutionAction::Discarded + } + crate::subjektiv::StagingResolutionAction::Invalid => { + server_api::SubjektivMemoryCandidateResolutionAction::Invalid + } + crate::subjektiv::StagingResolutionAction::Duplicate => { + server_api::SubjektivMemoryCandidateResolutionAction::Duplicate + } + crate::subjektiv::StagingResolutionAction::AlreadyCovered => { + server_api::SubjektivMemoryCandidateResolutionAction::AlreadyCovered + } + }, + reason: receipt.resolution.reason, + affected_memory: receipt + .resolution + .affected_memory + .into_iter() + .map(|reference| server_api::SubjektivMemoryAffectedRef { + memory_id: reference.memory_id, + revision: reference.revision, + operation, + }) + .collect(), + memory: receipt + .memory + .map(|memory| server_api::SubjektivMemoryRevisionRef { + memory_id: memory.id, + revision: memory.revision, + }), + store_revision: receipt.store_revision, + surface_dirty: receipt.surface_dirty, + } +} + +fn domain_memory_state(state: server_api::SubjektivMemoryState) -> crate::subjektiv::MemoryState { + match state { + server_api::SubjektivMemoryState::Active => crate::subjektiv::MemoryState::Active, + server_api::SubjektivMemoryState::Resolved => crate::subjektiv::MemoryState::Resolved, + server_api::SubjektivMemoryState::Retracted => crate::subjektiv::MemoryState::Retracted, + } +} + fn subjektiv_memory_query( store: &crate::subjektiv::SubjektivStore, subject_id: &str, @@ -17842,6 +18174,10 @@ fn subjektiv_store_error(error: crate::subjektiv::SubjektivError) -> Error { crate::subjektiv::SubjektivError::RevisionConflict { .. } => { Error::RepositoryConflict(format!("revision_conflict: {error}")) } + crate::subjektiv::SubjektivError::CandidateResolved(_) + | crate::subjektiv::SubjektivError::DecisionRequestConflict(_) => { + Error::RepositoryConflict(format!("candidate_decision_conflict: {error}")) + } crate::subjektiv::SubjektivError::MemoryNotFound(_) => { Error::InvalidInput(format!("memory_not_found: {error}")) } @@ -17851,8 +18187,185 @@ 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( + State(api): State, + AxumPath(path): AxumPath, + context: server_api::ServerRequestContext, + Json(operation): Json, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let (subject_id, _, authority) = subjektiv_subject_scope(&api, &path.workspace_id, &context)?; + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Subject, + "consolidation request", + )?; + Ok(Json(start_subjektiv_staging_consolidation( + api, + &subject_id, + operation, + )?)) +} + +fn start_subjektiv_staging_consolidation( + api: WorkspaceApi, + subject_id: &str, + operation: MemoryConsolidateStagingOperation, +) -> ApiResult { + let (candidate_count, total_bytes) = open_subjektiv_store(&api)? + .pending_staging_backlog(subject_id) + .map_err(subjektiv_store_error)?; + if candidate_count == 0 { + return Ok(MemoryConsolidationOutput { + status: "skipped_empty".into(), + summary: "No subject Memory staging candidates are pending.".into(), + candidate_count, + total_bytes, + }); + } + const CONSOLIDATION_THRESHOLD_FILES: usize = 5; + const CONSOLIDATION_THRESHOLD_BYTES: u64 = 50_000; + if !operation.force + && candidate_count < CONSOLIDATION_THRESHOLD_FILES + && total_bytes < CONSOLIDATION_THRESHOLD_BYTES + { + return Ok(MemoryConsolidationOutput { + status: "skipped_below_threshold".into(), + summary: format!( + "Subject Memory staging backlog has {candidate_count} candidate(s), {total_bytes} byte(s), below Backend policy threshold." + ), + candidate_count, + total_bytes, + }); + } + + let runtime_id = select_memory_consolidation_runtime(&api)?; + let singleton_key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}"); + let input_content = format!( + "Process the delegated subject's {candidate_count} pending Memory candidate(s) ({total_bytes} bytes) with MemoryStagingList, MemoryStagingRead, subject Memory reads, and MemoryApplyCandidate." + ); + if let Some(output) = try_reuse_subject_consolidation_worker( + &api, + &singleton_key, + &input_content, + candidate_count, + total_bytes, + )? { + return Ok(output); + } + let result = api.spawn_workspace_worker( + &runtime_id, + WorkerSpawnRequest { + requested_worker_name: Some(SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE.to_string()), + singleton_key: Some(singleton_key), + intent: WorkerSpawnIntent::WorkspaceOrchestrator, + acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted { + expected_segments: 1, + }, + profile: ProfileSelector::Builtin(SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE.to_string()), + ticket_assignment: None, + initial_submit: vec![Segment::text(input_content)], + 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, + }, + )?; + if result.state != InternalWorkerOperationState::Accepted { + return Ok(MemoryConsolidationOutput { + status: "skipped_spawn_rejected".into(), + summary: "Runtime rejected subject Memory consolidater spawn.".into(), + candidate_count, + total_bytes, + }); + } + let Some(worker) = result.worker else { + return Ok(MemoryConsolidationOutput { + status: "skipped_spawn_missing_worker".into(), + summary: "Runtime accepted subject Memory consolidater spawn without returning a Worker summary.".into(), + candidate_count, + total_bytes, + }); + }; + Ok(MemoryConsolidationOutput { + status: "started".into(), + summary: format!( + "Started subject Memory consolidater '{}' for {candidate_count} staging candidate(s).", + worker.worker.worker_id + ), + candidate_count, + total_bytes, + }) +} + +fn try_reuse_subject_consolidation_worker( + api: &WorkspaceApi, + singleton_key: &str, + input_content: &str, + candidate_count: usize, + total_bytes: u64, +) -> ApiResult> { + let Some(owner) = api + .store + .current_worker_singleton_owner(&api.config.workspace_id, singleton_key)? + else { + return Ok(None); + }; + let mut worker = api + .runtime + .worker(&owner.worker) + .map_err(|error| error.into_error())?; + worker.singleton_key = Some(owner.key); + if worker.state != "idle" { + return Ok(Some(MemoryConsolidationOutput { + status: "skipped_existing_not_idle".into(), + summary: format!( + "Existing subject Memory consolidater '{}' is '{}', not confirmed idle.", + worker.worker.worker_id, worker.state + ), + candidate_count, + total_bytes, + })); + } + let input = api + .runtime + .send_input( + &worker.worker, + WorkerInputRequest { + kind: WorkerInputKind::User, + content: input_content.to_string(), + submission_request_id: None, + segments: None, + }, + ) + .map_err(|error| error.into_error())?; + if input.state != InternalWorkerOperationState::Accepted { + return Ok(Some(MemoryConsolidationOutput { + status: "skipped_existing_input_rejected".into(), + summary: "Existing subject Memory consolidater rejected new input.".into(), + candidate_count, + total_bytes, + })); + } + Ok(Some(MemoryConsolidationOutput { + status: "reused".into(), + summary: format!( + "Reused subject Memory consolidater '{}' for {candidate_count} staging candidate(s).", + worker.worker.worker_id + ), + candidate_count, + total_bytes, + })) +} + async fn scoped_memory_consolidation( State(api): State, AxumPath(path): AxumPath, @@ -50647,12 +51160,26 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3); #[test] fn subjektiv_singleton_scope_is_explicit_and_opaque() { assert_eq!( - subjektiv_subject_id_from_singleton_key("subjektiv:subject-42").unwrap(), + subjektiv_scope_from_singleton_key("subjektiv:subject-42") + .unwrap() + .0, "subject-42" ); - assert!(subjektiv_subject_id_from_singleton_key("workspace-orchestrator").is_err()); - assert!(subjektiv_subject_id_from_singleton_key("subjektiv:").is_err()); - assert!(subjektiv_subject_id_from_singleton_key("subjektiv: subject-42").is_err()); + assert_eq!( + subjektiv_scope_from_singleton_key("subjektiv-consolidation:subject-42") + .unwrap() + .1, + SubjektivWorkerAuthority::Consolidation + ); + assert_eq!( + subjektiv_scope_from_singleton_key("subjektiv:subject-42") + .unwrap() + .1, + SubjektivWorkerAuthority::Subject + ); + assert!(subjektiv_scope_from_singleton_key("workspace-orchestrator").is_err()); + assert!(subjektiv_scope_from_singleton_key("subjektiv:").is_err()); + assert!(subjektiv_scope_from_singleton_key("subjektiv: subject-42").is_err()); assert_eq!( subjektiv_singleton_key("subject-42").unwrap(), "subjektiv:subject-42" diff --git a/crates/workspace-server/src/subjektiv.rs b/crates/workspace-server/src/subjektiv.rs index 9c3d4c59b..a5b92a4f9 100644 --- a/crates/workspace-server/src/subjektiv.rs +++ b/crates/workspace-server/src/subjektiv.rs @@ -42,6 +42,8 @@ pub enum SubjektivError { CandidateConflict(String), #[error("staging candidate `{0}` is already resolved")] CandidateResolved(String), + #[error("candidate decision request `{0}` was already used with different input")] + DecisionRequestConflict(String), #[error( "session `{session_id}` is already attributed to subject `{existing_subject_id}` and cannot be attributed to `{requested_subject_id}`" )] @@ -336,7 +338,8 @@ pub struct MemoryRecord { pub updated_at: String, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct MemoryDraft { pub kind: CandidateKind, pub state: MemoryState, @@ -411,7 +414,8 @@ pub struct SurfaceSnapshot { pub created_at: String, } -#[derive(Debug, Clone)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] pub enum MemoryRevisionTarget { Create, Revise { @@ -420,6 +424,51 @@ pub enum MemoryRevisionTarget { }, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum MemoryDecisionOperation { + Create, + Revise, + Reference, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum CandidateDecision { + Apply { + target: MemoryRevisionTarget, + draft: MemoryDraft, + }, + Close { + action: StagingResolutionAction, + #[serde(default)] + affected_memory: Vec, + }, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CandidateDecisionRequest { + pub request_id: String, + pub candidate_id: String, + pub reason: String, + pub decision: CandidateDecision, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CandidateDecisionReceipt { + pub request_id: String, + pub candidate_id: String, + pub resolution: StagingResolution, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub memory: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub operation: Option, + pub store_revision: u64, + pub surface_dirty: bool, +} + fn create_schema(transaction: &Transaction<'_>) -> crate::feature_storage::Result<()> { transaction.execute_batch( r#" @@ -761,6 +810,37 @@ END; Ok(()) } +fn add_candidate_decision_receipts( + transaction: &Transaction<'_>, +) -> crate::feature_storage::Result<()> { + transaction.execute_batch( + r#" +CREATE TABLE candidate_decision_receipts ( + subject_id TEXT NOT NULL, + request_id TEXT NOT NULL, + candidate_id TEXT NOT NULL, + request_json TEXT NOT NULL, + receipt_json TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (subject_id, request_id), + UNIQUE (subject_id, candidate_id), + FOREIGN KEY (subject_id, candidate_id) + REFERENCES staging_resolutions(subject_id, candidate_id) ON DELETE RESTRICT +); + +CREATE TRIGGER candidate_decision_receipts_no_update +BEFORE UPDATE ON candidate_decision_receipts BEGIN + SELECT RAISE(ABORT, 'subjektiv candidate decision receipts are immutable'); +END; +CREATE TRIGGER candidate_decision_receipts_no_delete +BEFORE DELETE ON candidate_decision_receipts BEGIN + SELECT RAISE(ABORT, 'subjektiv candidate decision receipts are retained'); +END; +"#, + )?; + Ok(()) +} + static MIGRATIONS: &[FeatureMigration] = &[ FeatureMigration::new(1, "create subjektiv subject memory store", create_schema), FeatureMigration::new( @@ -768,6 +848,11 @@ static MIGRATIONS: &[FeatureMigration] = &[ "add immutable subject session attribution", add_subject_session_attribution, ), + FeatureMigration::new( + 3, + "add idempotent atomic candidate decision receipts", + add_candidate_decision_receipts, + ), ]; pub const REGISTRATION: FeatureRegistration = @@ -1024,6 +1109,56 @@ impl SubjektivStore { }) } + pub fn pending_staging_candidates( + &self, + subject_id: &str, + limit: usize, + ) -> Result> { + self.database.try_with_connection(|connection| { + require_subject_in_connection(connection, subject_id)?; + let mut statement = connection.prepare( + "SELECT s.record_json + FROM staging_records s + LEFT JOIN staging_resolutions r + ON r.subject_id = s.subject_id AND r.candidate_id = s.candidate_id + WHERE s.subject_id = ?1 AND r.candidate_id IS NULL + ORDER BY s.created_at ASC, s.candidate_id ASC + LIMIT ?2", + )?; + let rows = statement.query_map(params![subject_id, to_i64(limit as u64)?], |row| { + row.get::<_, String>(0) + })?; + let mut records = Vec::new(); + for row in rows { + records.push(parse_staging(self.workspace_id(), &row?)?); + } + Ok(records) + }) + } + + pub fn pending_staging_backlog(&self, subject_id: &str) -> Result<(usize, u64)> { + self.database.try_with_connection(|connection| { + require_subject_in_connection(connection, subject_id)?; + let (count, bytes) = connection.query_row( + "SELECT COUNT(*), COALESCE(SUM(LENGTH(s.record_json)), 0) + FROM staging_records s + LEFT JOIN staging_resolutions r + ON r.subject_id = s.subject_id AND r.candidate_id = s.candidate_id + WHERE s.subject_id = ?1 AND r.candidate_id IS NULL", + [subject_id], + |row| Ok((row.get::<_, i64>(0)?, row.get::<_, i64>(1)?)), + )?; + Ok(( + usize::try_from(count).map_err(|_| { + SubjektivError::InvalidRecord("pending candidate count exceeds usize".into()) + })?, + u64::try_from(bytes).map_err(|_| { + SubjektivError::InvalidRecord("pending candidate bytes are negative".into()) + })?, + )) + }) + } + pub fn create_memory(&self, subject_id: &str, draft: MemoryDraft) -> Result { validate_direct_memory_draft(&draft)?; if draft.state != MemoryState::Active { @@ -1080,7 +1215,7 @@ impl SubjektivStore { subject_id: &str, candidate_ids: &[String], target: MemoryRevisionTarget, - mut draft: MemoryDraft, + draft: MemoryDraft, reason: impl Into, ) -> Result<(MemoryRecord, Vec)> { if candidate_ids.is_empty() { @@ -1092,52 +1227,127 @@ impl SubjektivStore { for candidate_id in candidate_ids { validate_label("candidate id", candidate_id)?; } - draft.source_candidate_ids = candidate_ids.to_vec(); let reason = reason.into(); validate_nonempty("resolution reason", &reason)?; - let new_memory_id = issued_id("memory"); self.database.try_transaction(|transaction| { - for candidate_id in candidate_ids { - require_unresolved_candidate(transaction, subject_id, candidate_id)?; - } - let memory = match &target { - MemoryRevisionTarget::Create => { - if draft.state != MemoryState::Active { - return Err(SubjektivError::InvalidRecord( - "a new Memory must start in active state".to_string(), - )); - } - write_memory_revision(transaction, subject_id, &new_memory_id, None, draft)? + apply_candidates_in_transaction( + transaction, + subject_id, + candidate_ids, + &target, + draft, + &reason, + ) + }) + } + + pub fn decide_candidate( + &self, + subject_id: &str, + request: CandidateDecisionRequest, + ) -> Result { + validate_label("candidate decision request id", &request.request_id)?; + validate_label("candidate id", &request.candidate_id)?; + validate_nonempty("resolution reason", &request.reason)?; + let request_raw = serde_json::to_string(&request)?; + self.database.try_transaction(|transaction| { + if let Some((stored_request, stored_receipt)) = transaction + .query_row( + "SELECT request_json, receipt_json + FROM candidate_decision_receipts + WHERE subject_id = ?1 AND request_id = ?2", + params![subject_id, request.request_id], + |row| Ok((row.get::<_, String>(0)?, row.get::<_, String>(1)?)), + ) + .optional()? + { + if stored_request == request_raw { + return serde_json::from_str(&stored_receipt).map_err(Into::into); } - MemoryRevisionTarget::Revise { - memory_id, - expected_revision, - } => write_memory_revision( - transaction, - subject_id, - memory_id, - Some(*expected_revision), - draft, - )?, - }; - let reference = MemoryRevisionRef { - memory_id: memory.id.clone(), - revision: memory.revision, - }; - let resolutions = candidate_ids - .iter() - .map(|candidate_id| { - insert_resolution( + return Err(SubjektivError::DecisionRequestConflict( + request.request_id.clone(), + )); + } + if transaction + .query_row( + "SELECT 1 FROM candidate_decision_receipts + WHERE subject_id = ?1 AND candidate_id = ?2", + params![subject_id, request.candidate_id], + |_| Ok(()), + ) + .optional()? + .is_some() + { + return Err(SubjektivError::CandidateResolved( + request.candidate_id.clone(), + )); + } + + let (memory, operation, resolution) = match request.decision.clone() { + CandidateDecision::Apply { target, draft } => { + let operation = match &target { + MemoryRevisionTarget::Create => MemoryDecisionOperation::Create, + MemoryRevisionTarget::Revise { .. } => MemoryDecisionOperation::Revise, + }; + let candidate_ids = [request.candidate_id.clone()]; + let (memory, mut resolutions) = apply_candidates_in_transaction( transaction, subject_id, - candidate_id, - StagingResolutionAction::Applied, - &reason, - std::slice::from_ref(&reference), - ) - }) - .collect::>>()?; - Ok((memory, resolutions)) + &candidate_ids, + &target, + draft, + &request.reason, + )?; + (Some(memory), Some(operation), resolutions.remove(0)) + } + CandidateDecision::Close { + action, + affected_memory, + } => { + if action == StagingResolutionAction::Applied { + return Err(SubjektivError::InvalidRecord( + "applied candidates require an apply decision".into(), + )); + } + let resolution = insert_resolution( + transaction, + subject_id, + &request.candidate_id, + action, + &request.reason, + &affected_memory, + )?; + let operation = + (!affected_memory.is_empty()).then_some(MemoryDecisionOperation::Reference); + (None, operation, resolution) + } + }; + let store_revision = require_subject(transaction, subject_id)?.store_revision; + let receipt = CandidateDecisionReceipt { + request_id: request.request_id.clone(), + candidate_id: request.candidate_id.clone(), + surface_dirty: memory.is_some(), + memory, + operation, + resolution, + store_revision, + }; + let receipt_raw = serde_json::to_string(&receipt)?; + transaction.execute( + "INSERT INTO candidate_decision_receipts ( + subject_id, request_id, candidate_id, request_json, + receipt_json, created_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![ + subject_id, + request.request_id, + request.candidate_id, + request_raw, + receipt_raw, + now() + ], + )?; + Ok(receipt) }) } @@ -1390,6 +1600,64 @@ impl SubjektivStore { } } +fn apply_candidates_in_transaction( + transaction: &Transaction<'_>, + subject_id: &str, + candidate_ids: &[String], + target: &MemoryRevisionTarget, + mut draft: MemoryDraft, + reason: &str, +) -> Result<(MemoryRecord, Vec)> { + let mut candidates = Vec::with_capacity(candidate_ids.len()); + for candidate_id in candidate_ids { + let raw = require_unresolved_candidate(transaction, subject_id, candidate_id)?; + candidates.push(parse_staging_without_workspace(&raw)?); + } + for candidate in &candidates { + validate_candidate_application(transaction, subject_id, candidate, target, &draft)?; + } + draft.source_candidate_ids = candidate_ids.to_vec(); + let new_memory_id = issued_id("memory"); + let memory = match target { + MemoryRevisionTarget::Create => { + if draft.state != MemoryState::Active { + return Err(SubjektivError::InvalidRecord( + "a new Memory must start in active state".into(), + )); + } + write_memory_revision(transaction, subject_id, &new_memory_id, None, draft)? + } + MemoryRevisionTarget::Revise { + memory_id, + expected_revision, + } => write_memory_revision( + transaction, + subject_id, + memory_id, + Some(*expected_revision), + draft, + )?, + }; + let reference = MemoryRevisionRef { + memory_id: memory.id.clone(), + revision: memory.revision, + }; + let resolutions = candidate_ids + .iter() + .map(|candidate_id| { + insert_resolution( + transaction, + subject_id, + candidate_id, + StagingResolutionAction::Applied, + reason, + std::slice::from_ref(&reference), + ) + }) + .collect::>>()?; + Ok((memory, resolutions)) +} + fn write_memory_revision( transaction: &Transaction<'_>, subject_id: &str, @@ -1744,8 +2012,26 @@ fn require_unresolved_candidate( params![subject_id, candidate_id], |row| row.get::<_, String>(0), ) - .optional()? - .ok_or_else(|| SubjektivError::CandidateNotFound(candidate_id.to_string()))?; + .optional()?; + let raw = match raw { + Some(raw) => raw, + None => { + let foreign = transaction + .query_row( + "SELECT subject_id FROM staging_records WHERE candidate_id = ?1 LIMIT 1", + [candidate_id], + |row| row.get::<_, String>(0), + ) + .optional()?; + if foreign.is_some() { + return Err(SubjektivError::SubjectScopeMismatch { + subject_id: subject_id.to_string(), + reference: candidate_id.to_string(), + }); + } + return Err(SubjektivError::CandidateNotFound(candidate_id.to_string())); + } + }; let resolved = transaction .query_row( "SELECT 1 FROM staging_resolutions @@ -2001,6 +2287,74 @@ fn validate_revision_proposal_metadata(proposal: &RevisionProposal) -> Result<() validate_nonempty("revision proposal change_reason", &proposal.change_reason) } +fn validate_candidate_application( + transaction: &Transaction<'_>, + subject_id: &str, + candidate: &SubjectStagingRecord, + target: &MemoryRevisionTarget, + draft: &MemoryDraft, +) -> Result<()> { + let Some(proposal) = &candidate.revision_proposal else { + return Ok(()); + }; + let MemoryRevisionTarget::Revise { + memory_id, + expected_revision, + } = target + else { + return Err(SubjektivError::InvalidRecord( + "a revision proposal cannot create a different Memory".into(), + )); + }; + if memory_id != &proposal.memory_id || expected_revision != &proposal.expected_revision { + return Err(SubjektivError::InvalidRecord( + "candidate application target must exactly match revision proposal metadata".into(), + )); + } + let current = scoped_memory_in_connection(transaction, subject_id, memory_id)?; + if current.revision != *expected_revision { + return Err(SubjektivError::RevisionConflict { + memory_id: memory_id.clone(), + expected: *expected_revision, + actual: current.revision, + }); + } + let expected_state = match proposal.intent { + RevisionProposalIntent::Revise => current.state, + RevisionProposalIntent::Resolve => MemoryState::Resolved, + RevisionProposalIntent::Retract => MemoryState::Retracted, + RevisionProposalIntent::Reopen => MemoryState::Active, + }; + let intent_allowed = match proposal.intent { + RevisionProposalIntent::Revise => { + matches!(current.state, MemoryState::Active | MemoryState::Resolved) + } + RevisionProposalIntent::Resolve => current.state == MemoryState::Active, + RevisionProposalIntent::Retract => { + matches!(current.state, MemoryState::Active | MemoryState::Resolved) + } + RevisionProposalIntent::Reopen => current.state == MemoryState::Resolved, + }; + if !intent_allowed || draft.state != expected_state { + return Err(SubjektivError::InvalidStateTransition { + memory_id: memory_id.clone(), + from: current.state, + to: draft.state, + }); + } + if draft.kind != current.kind || draft.kind != candidate.kind { + return Err(SubjektivError::InvalidRecord( + "revision proposal application must preserve the target Memory kind".into(), + )); + } + if draft.change_reason != proposal.change_reason { + return Err(SubjektivError::InvalidRecord( + "revision proposal application must preserve change_reason".into(), + )); + } + Ok(()) +} + fn validate_staged_revision_proposal( transaction: &Transaction<'_>, record: &SubjectStagingRecord, @@ -2802,6 +3156,321 @@ mod tests { ); } + #[test] + fn atomic_candidate_decision_is_idempotent_and_preserves_proposal_metadata() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(&temp.path().join("storage"), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let memory = store + .create_memory(&subject.id, draft("Original", "initial observation")) + .unwrap(); + let change_reason = "The remembered condition has ended"; + let staged = store + .stage_candidate( + candidate(&subject.id, "candidate-resolve-atomic", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Resolve, + &memory.id, + memory.revision, + change_reason, + )) + .unwrap(), + ) + .unwrap(); + let mut revised = draft("Resolved", change_reason); + revised.state = MemoryState::Resolved; + let request = CandidateDecisionRequest { + request_id: "decision-resolve-1".into(), + candidate_id: staged.id.clone(), + reason: "Accepted exact resolution proposal".into(), + decision: CandidateDecision::Apply { + target: MemoryRevisionTarget::Revise { + memory_id: memory.id.clone(), + expected_revision: memory.revision, + }, + draft: revised, + }, + }; + + let first = store + .decide_candidate(&subject.id, request.clone()) + .unwrap(); + let retried = store.decide_candidate(&subject.id, request).unwrap(); + + assert_eq!(first.memory.as_ref().unwrap().revision, 2); + assert_eq!(retried.memory.as_ref().unwrap().revision, 2); + assert_eq!(first.resolution.id, retried.resolution.id); + assert_eq!(first.store_revision, 2); + assert!(first.surface_dirty); + assert_eq!( + first + .resolution + .candidate + .revision_proposal + .unwrap() + .change_reason, + change_reason + ); + assert_eq!( + store + .list_memory_revisions(&subject.id, &memory.id) + .unwrap() + .len(), + 2 + ); + } + + #[test] + fn candidate_decision_revalidates_stale_proposals_and_intent_transitions() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(&temp.path().join("storage"), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let memory = store + .create_memory(&subject.id, draft("Lifecycle", "initial")) + .unwrap(); + let stale = store + .stage_candidate( + candidate(&subject.id, "candidate-stale-apply", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Revise, + &memory.id, + 1, + "stale correction", + )) + .unwrap(), + ) + .unwrap(); + store + .revise_memory(&subject.id, &memory.id, 1, draft("Advanced", "advance")) + .unwrap(); + let stale_result = store.decide_candidate( + &subject.id, + CandidateDecisionRequest { + request_id: "decision-stale".into(), + candidate_id: stale.id.clone(), + reason: "must not auto-rebase".into(), + decision: CandidateDecision::Apply { + target: MemoryRevisionTarget::Revise { + memory_id: memory.id.clone(), + expected_revision: 1, + }, + draft: draft("Stale", "stale correction"), + }, + }, + ); + assert!(matches!( + stale_result, + Err(SubjektivError::RevisionConflict { + expected: 1, + actual: 2, + .. + }) + )); + assert!( + store + .staging_resolution(&subject.id, &stale.id) + .unwrap() + .is_none() + ); + + let apply_intent = |store: &SubjektivStore, + candidate_id: &str, + intent: RevisionProposalIntent, + expected_revision: u64, + state: MemoryState, + reason: &str| { + let staged = store + .stage_candidate( + candidate(&subject.id, candidate_id, "workspace-a") + .with_revision_proposal(proposal( + intent, + &memory.id, + expected_revision, + reason, + )) + .unwrap(), + ) + .unwrap(); + let mut next = draft(candidate_id, reason); + next.state = state; + store + .decide_candidate( + &subject.id, + CandidateDecisionRequest { + request_id: format!("decision-{candidate_id}"), + candidate_id: staged.id, + reason: format!("apply {candidate_id}"), + decision: CandidateDecision::Apply { + target: MemoryRevisionTarget::Revise { + memory_id: memory.id.clone(), + expected_revision, + }, + draft: next, + }, + }, + ) + .unwrap() + .memory + .unwrap() + }; + let resolved = apply_intent( + &store, + "resolve", + RevisionProposalIntent::Resolve, + 2, + MemoryState::Resolved, + "resolve reason", + ); + let reopened = apply_intent( + &store, + "reopen", + RevisionProposalIntent::Reopen, + resolved.revision, + MemoryState::Active, + "reopen reason", + ); + let retracted = apply_intent( + &store, + "retract", + RevisionProposalIntent::Retract, + reopened.revision, + MemoryState::Retracted, + "retract reason", + ); + assert_eq!(retracted.state, MemoryState::Retracted); + assert!(matches!( + store.stage_candidate( + candidate(&subject.id, "post-retract", "workspace-a") + .with_revision_proposal(proposal( + RevisionProposalIntent::Reopen, + &memory.id, + retracted.revision, + "cannot revive", + )) + .unwrap() + ), + Err(SubjektivError::InvalidStateTransition { .. }) + )); + } + + #[test] + fn candidate_decisions_reject_conflicts_and_roll_back_partial_writes() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(&temp.path().join("storage"), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let other = store.create_subject(role()).unwrap(); + let staged = store + .stage_candidate(candidate(&subject.id, "candidate-rollback", "workspace-a")) + .unwrap(); + let foreign = store + .stage_candidate(candidate(&other.id, "candidate-foreign", "workspace-a")) + .unwrap(); + let mut invalid = draft("Would be written", "must roll back"); + invalid.derived_from.push(MemoryRevisionRef { + memory_id: "missing-memory".into(), + revision: 1, + }); + let failed = store.decide_candidate( + &subject.id, + CandidateDecisionRequest { + request_id: "decision-rollback".into(), + candidate_id: staged.id.clone(), + reason: "invalid derivation".into(), + decision: CandidateDecision::Apply { + target: MemoryRevisionTarget::Create, + draft: invalid, + }, + }, + ); + assert!(matches!( + failed, + Err(SubjektivError::SubjectScopeMismatch { .. }) + )); + assert!( + store + .staging_resolution(&subject.id, &staged.id) + .unwrap() + .is_none() + ); + assert!(store.list_memories(&subject.id).unwrap().is_empty()); + + let cross_subject = store.decide_candidate( + &subject.id, + CandidateDecisionRequest { + request_id: "decision-foreign".into(), + candidate_id: foreign.id, + reason: "must remain scoped".into(), + decision: CandidateDecision::Close { + action: StagingResolutionAction::Invalid, + affected_memory: vec![], + }, + }, + ); + assert!(matches!( + cross_subject, + Err(SubjektivError::SubjectScopeMismatch { .. }) + )); + + let closed = store + .decide_candidate( + &subject.id, + CandidateDecisionRequest { + request_id: "decision-close".into(), + candidate_id: staged.id.clone(), + reason: "already represented elsewhere".into(), + decision: CandidateDecision::Close { + action: StagingResolutionAction::AlreadyCovered, + affected_memory: vec![], + }, + }, + ) + .unwrap(); + assert!(!closed.surface_dirty); + assert!(closed.memory.is_none()); + for (suffix, action) in [ + ("discarded", StagingResolutionAction::Discarded), + ("invalid", StagingResolutionAction::Invalid), + ("duplicate", StagingResolutionAction::Duplicate), + ] { + let candidate_id = format!("candidate-{suffix}"); + store + .stage_candidate(candidate(&subject.id, &candidate_id, "workspace-a")) + .unwrap(); + let receipt = store + .decide_candidate( + &subject.id, + CandidateDecisionRequest { + request_id: format!("decision-{suffix}"), + candidate_id, + reason: format!("specific {suffix} reason"), + decision: CandidateDecision::Close { + action, + affected_memory: vec![], + }, + }, + ) + .unwrap(); + assert_eq!(receipt.resolution.action, action); + assert!(!receipt.surface_dirty); + } + let conflicting_retry = store.decide_candidate( + &subject.id, + CandidateDecisionRequest { + request_id: "decision-close".into(), + candidate_id: staged.id, + reason: "different disposition".into(), + decision: CandidateDecision::Close { + action: StagingResolutionAction::Discarded, + affected_memory: vec![], + }, + }, + ); + assert!(matches!( + conflicting_retry, + Err(SubjektivError::DecisionRequestConflict(_)) + )); + } + #[test] fn revision_proposals_enforce_scope_revision_and_state_transitions() { let temp = tempfile::tempdir().unwrap(); diff --git a/openapi/server-api.json b/openapi/server-api.json index 69ebe5b89..968d40110 100644 --- a/openapi/server-api.json +++ b/openapi/server-api.json @@ -14081,7 +14081,7 @@ "info": { "title": "Yoi Server API", "version": "0.1.0", - "x-yoi-source-digest": "sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f" + "x-yoi-source-digest": "sha256:c77c3a923b6eb888bc80137894313f668ed14e11248695ca5fb07cb00281fef9" }, "openapi": "3.1.0", "paths": { diff --git a/resources/profiles/subjektiv-memory-consolidation.dcdl b/resources/profiles/subjektiv-memory-consolidation.dcdl new file mode 100644 index 000000000..676610f7e --- /dev/null +++ b/resources/profiles/subjektiv-memory-consolidation.dcdl @@ -0,0 +1,17 @@ +import "./base.dcdl" // { + slug = "subjektiv-memory-consolidation"; + description = "Delegated subject Memory staging consolidation profile."; + scope = "workspace_read"; + model = { ref = "codex-oauth/gpt-5.6-luna"; }; + engine = { reasoning = "medium"; }; + + feature = { + task = { enabled = false; }; + memory = { enabled = true; staging_tools = true; }; + web = { enabled = false; }; + sub_worker = { enabled = false; }; + worker = { enabled = false; }; + objective = { enabled = false; }; + ticket = { enabled = false; thread = false; }; + }; +} diff --git a/resources/prompts/catalog.dcdl b/resources/prompts/catalog.dcdl index 7d61dcce9..271420d6c 100644 --- a/resources/prompts/catalog.dcdl +++ b/resources/prompts/catalog.dcdl @@ -23,6 +23,7 @@ internalFlowVerifierSystem = import "./internal/flow_verifier_system.md"; internalBackendJobSystem = import "./internal/backend_job_system.md"; internalTicketItemChecker = import "./internal/ticket_item_checker.md"; internalMemoryConsolidationSystem = import "./internal/memory_consolidation_system.md"; +internalSubjektivMemoryConsolidationSystem = import "./internal/subjektiv_memory_consolidation_system.md"; internalMemoryExtractSystem = import "./internal/memory_extract_system.md"; internalWorkspaceOrchestratorQueueAttention = import "./internal/workspace_orchestrator_queue_attention.md"; internalNotifyWrapper = import "./internal/notify_wrapper.md"; @@ -60,6 +61,7 @@ in backend_job_system = internalBackendJobSystem.content; ticket_item_checker = internalTicketItemChecker.content; memory_consolidation_system = internalMemoryConsolidationSystem.content; + subjektiv_memory_consolidation_system = internalSubjektivMemoryConsolidationSystem.content; memory_extract_system = internalMemoryExtractSystem.content; workspace_orchestrator_queue_attention = internalWorkspaceOrchestratorQueueAttention.content; notify_wrapper = internalNotifyWrapper.content; diff --git a/resources/prompts/internal/subjektiv_memory_consolidation_system.md b/resources/prompts/internal/subjektiv_memory_consolidation_system.md new file mode 100644 index 000000000..b04cef77b --- /dev/null +++ b/resources/prompts/internal/subjektiv_memory_consolidation_system.md @@ -0,0 +1,41 @@ +# Memory staging consolidater + +You are the delegated internal Memory staging consolidater for one subject. + +Your job is to consume that subject's pending Memory candidates through tools. The host binds every tool call to the subject; never infer or provide a subject identifier yourself. + +Use this loop: + +1. Call `MemoryStagingList` to inspect pending candidates. +2. Pick one candidate and call `MemoryStagingRead` for the full immutable record, including any typed `revision_proposal`. +3. Use `SubjektivMemoryQuery`, `SubjektivMemoryRead`, and when useful `SubjektivMemoryListRevisions` to compare it with confirmed Memory. +4. Decide whether to create a Memory, revise/refine one exact current revision, resolve/retract/reopen according to typed proposal metadata, or reject the candidate. +5. Call `MemoryApplyCandidate` exactly once for every decision. Supply a stable `request_id`; retry the exact same request after an ambiguous transport failure. + +`MemoryApplyCandidate` atomically writes any new Memory revision, records concrete affected revision references, preserves the original candidate and provenance, and closes the pending candidate. A successful applied response is the only evidence that confirmed Memory changed. + +Rules: + +- A candidate is evidence, not authority. Do not promote derived/model claims to observations without warranted provenance. +- For a candidate with `revision_proposal`, use its `memory_id`, `expected_revision`, `intent`, and `change_reason`; do not infer a different target from claim prose. +- Proposal application must preserve `change_reason` exactly and use the state implied by `intent`: `resolve` -> `resolved`, `retract` -> `retracted`, `reopen` -> `active`, and `revise` keeps the current state. +- Never rebase a stale proposal. On `revision_conflict` or another typed conflict, leave it pending for later reread/reconciliation. +- Never revive a `retracted` Memory. New experience after retraction becomes a new Memory with an exact `derived_from` revision when justified. +- `resolved` -> `active` is allowed only for a valid `reopen` proposal. +- New or revised prose must remain concise and useful. Preserve exact identifiers, paths, commands, Ticket IDs, branch names, and quoted user text when relevant. +- `source_candidate_ids` are host-derived from the applied candidate. Do not attempt to author them. +- `derived_from` must contain only exact existing Memory revision refs returned by reads. +- Applied responses return affected refs with `{memory_id, revision, operation}`. `duplicate` and `already_covered` may reference exact existing revisions; `discarded` and `invalid` require a concrete reason. +- Surface/snapshot generation is deferred. `surface_dirty: true` reports that a later process should rebuild it; never treat surface generation failure as failure of an already returned committed decision. +- An empty model response or transport failure is not a candidate disposition. Only a successful `MemoryApplyCandidate` response closes a candidate. + +Valid non-applied close actions are `discarded`, `invalid`, `duplicate`, and `already_covered`. Always provide a specific reason. + +## Language + +- `language`: `{{language}}` +- Write confirmed Memory prose in this language. +- Preserve literal identifiers, paths, commands, branch names, issue IDs, tool names, model names, and quoted user/system text as-is. +- If the configured language is unclear, use English. + +Do not create Knowledge, Skill, Ticket, documentation, child Worker, or filesystem records. Use only the supplied Memory tools. diff --git a/web/workspace/src/lib/generated/companion-api.ts b/web/workspace/src/lib/generated/companion-api.ts index 77a81d918..7e782b48b 100644 --- a/web/workspace/src/lib/generated/companion-api.ts +++ b/web/workspace/src/lib/generated/companion-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-companion-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) +// Input contract: 0.1.0 (sha256:c77c3a923b6eb888bc80137894313f668ed14e11248695ca5fb07cb00281fef9) // Options: companion-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/companion-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_companion_api_types -- --check diff --git a/web/workspace/src/lib/generated/repository-api.ts b/web/workspace/src/lib/generated/repository-api.ts index 256443914..3ca4ddb28 100644 --- a/web/workspace/src/lib/generated/repository-api.ts +++ b/web/workspace/src/lib/generated/repository-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-repository-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) +// Input contract: 0.1.0 (sha256:c77c3a923b6eb888bc80137894313f668ed14e11248695ca5fb07cb00281fef9) // Options: repository-operation-schema-closure,openapi-3.1,format:uint64=safe-number,strict-safe-integers,sorted-output // Output: web/workspace/src/lib/generated/repository-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_repository_openapi_types -- --check diff --git a/web/workspace/src/lib/generated/runtime-api.ts b/web/workspace/src/lib/generated/runtime-api.ts index 6bc86ddb3..79b0d1bf0 100644 --- a/web/workspace/src/lib/generated/runtime-api.ts +++ b/web/workspace/src/lib/generated/runtime-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-runtime-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) +// Input contract: 0.1.0 (sha256:c77c3a923b6eb888bc80137894313f668ed14e11248695ca5fb07cb00281fef9) // Options: runtime-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/runtime-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_runtime_api_types -- --check diff --git a/web/workspace/src/lib/generated/ticket-api.ts b/web/workspace/src/lib/generated/ticket-api.ts index c6e12177f..e687625f0 100644 --- a/web/workspace/src/lib/generated/ticket-api.ts +++ b/web/workspace/src/lib/generated/ticket-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-ticket-browser-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) +// Input contract: 0.1.0 (sha256:c77c3a923b6eb888bc80137894313f668ed14e11248695ca5fb07cb00281fef9) // Options: ticket-objective-merge-request-browser-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/ticket-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_ticket_api_types -- --check diff --git a/web/workspace/src/lib/generated/worker-launch-api.ts b/web/workspace/src/lib/generated/worker-launch-api.ts index b9d390390..0824a0267 100644 --- a/web/workspace/src/lib/generated/worker-launch-api.ts +++ b/web/workspace/src/lib/generated/worker-launch-api.ts @@ -1,6 +1,6 @@ // @generated by yoi-worker-openapi-typescript@1. Do not edit. // Input: openapi/server-api.json -// Input contract: 0.1.0 (sha256:cacb143d65459cf22331ee637382620ae2a3462dbee51c0e5caf56752895105f) +// Input contract: 0.1.0 (sha256:c77c3a923b6eb888bc80137894313f668ed14e11248695ca5fb07cb00281fef9) // Options: worker-operation-schema-closure,openapi-3.1,strict-integer-ranges,sorted-output // Output: web/workspace/src/lib/generated/worker-launch-api.ts // Regenerate/check: cargo run -q -p server-api --example generate_worker_launch_api_types -- --check From ba98a6afe6459621b0bbfe3c948a7db4639da607 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 2 Oct 2026 11:13:11 +0900 Subject: [PATCH 2/2] 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; }; }; }