From 046e606a38b1a48958cb82c2c5d7b08e4ce3f709 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 2 Oct 2026 10:46:12 +0900 Subject: [PATCH] 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