feat: connect subject memory consolidation

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)]

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,7 +1312,8 @@ mod tests {
             );
         }
 
-        let consolidation = resolve("memory-consolidation");
+        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")
@@ -1321,6 +1323,7 @@ mod tests {
                 Some(ReasoningControl::Effort(ReasoningEffort::Medium))
             );
         }
+    }
 
     #[test]
     fn profile_rejects_workspace_memory_snapshot_authority_fields() {

crates/server-api/src/lib.rs

@@ -1190,6 +1190,19 @@ pub trait ServerApi {
         #[path] workspace_id: String,
         #[body] request: SubjektivMemoryBackendRequest,
     ) -> Result<SubjektivMemoryBackendResponse, RepositoryApiError>;
+    #[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<MemoryConsolidationResponse, RepositoryApiError>;
 
     #[get(
         "/api/w/{workspace_id}/skills",
@@ -4650,6 +4663,168 @@ pub struct SubjektivMemoryStageExplicitRequest {
     pub proposal: Option<SubjektivMemoryRevisionProposal>,
 }
 
+#[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<usize>,
+}
+
+#[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<String>,
+    pub source: memory::schema::SourceRef,
+    #[serde(default)]
+    pub evidence: Vec<memory::extract::StagingEvidence>,
+    #[serde(default)]
+    pub source_refs: Vec<memory::schema::SourceEvidenceRef>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub revision_proposal: Option<SubjektivMemoryRevisionProposalMetadata>,
+    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<SubjektivMemoryRevisionProposalMetadata>,
+    pub created_at: String,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)]
+#[serde(deny_unknown_fields)]
+pub struct SubjektivMemoryCandidateListResponse {
+    pub items: Vec<SubjektivMemoryCandidateSummary>,
+    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<String>,
+    #[serde(default)]
+    pub derived_from: Vec<SubjektivMemoryRevisionRef>,
+    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<SubjektivMemoryRevisionRef>,
+    },
+}
+
+#[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<SubjektivMemoryAffectedRef>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub memory: Option<SubjektivMemoryRevisionRef>,
+    #[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",
         ];

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

@@ -84,7 +84,14 @@ impl dyn WorkspaceClient + '_ {
         &self,
         operation: MemoryConsolidateStagingOperation,
     ) -> Result<MemoryConsolidationOutput, WorkspaceMemoryBackendError> {
-        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<MemoryConsolidationOutput, WorkspaceMemoryBackendError> {
+        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<MemoryConsolidationOutput, WorkspaceMemoryBackendError> {
     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<dyn WorkspaceClient>) -> Vec<Tool
     ]
 }
 
+pub fn workspace_http_subjektiv_consolidation_tools(
+    client: Arc<dyn WorkspaceClient>,
+) -> Vec<ToolDefinition> {
+    use SubjectConsolidationOperation as Operation;
+    [
+        (
+            "SubjektivMemoryQuery",
+            "Search current Memory revisions for the delegated subject.",
+            schema_for::<server_api::SubjektivMemoryQueryRequest>(),
+            Operation::Query,
+        ),
+        (
+            "SubjektivMemoryRead",
+            "Read one current or historical Memory revision for the delegated subject.",
+            schema_for::<server_api::SubjektivMemoryReadRequest>(),
+            Operation::Read,
+        ),
+        (
+            "SubjektivMemoryListRevisions",
+            "List immutable revisions of one Memory for the delegated subject.",
+            schema_for::<server_api::SubjektivMemoryListRevisionsRequest>(),
+            Operation::ListRevisions,
+        ),
+        (
+            "MemoryStagingList",
+            "List pending staging candidates for the delegated subject.",
+            schema_for::<server_api::SubjektivMemoryCandidateListRequest>(),
+            Operation::ListCandidates,
+        ),
+        (
+            "MemoryStagingRead",
+            "Read one pending candidate, including typed revision proposal metadata and provenance.",
+            schema_for::<server_api::SubjektivMemoryCandidateReadRequest>(),
+            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::<server_api::SubjektivMemoryCandidateDecisionRequest>(),
+            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<dyn Tool>,
+            )
+        }) as ToolDefinition
+    })
+    .collect()
+}
+
+#[derive(Clone, Copy)]
+enum SubjectConsolidationOperation {
+    Query,
+    Read,
+    ListRevisions,
+    ListCandidates,
+    ReadCandidate,
+    DecideCandidate,
+}
+
+struct SubjectConsolidationTool {
+    client: Arc<dyn WorkspaceClient>,
+    operation: SubjectConsolidationOperation,
+}
+
+#[async_trait]
+impl Tool for SubjectConsolidationTool {
+    async fn execute(
+        &self,
+        input_json: &str,
+        _ctx: ToolExecutionContext,
+    ) -> Result<ToolOutput, ToolError> {
+        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::<server_api::RepositoryApiError>(&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<dyn WorkspaceClient>,
 ) -> Vec<ToolDefinition> {
@@ -407,16 +621,21 @@ impl MemoryFeatureInstallPlan {
         prompts: Arc<crate::prompt::catalog::PromptCatalog>,
         profile: Option<manifest::ProfileManifestSnapshot>,
     ) -> std::io::Result<Option<Self>> {
-        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)
+                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<ToolDefinition>) -> 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"));
+    }
 }

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

@@ -164,7 +164,7 @@ impl SubjektivLifecycleFeature {
         workspace_context: WorkerWorkspaceContext,
         event_tx: Option<broadcast::Sender<Event>>,
     ) -> std::io::Result<Option<Self>> {
-        if !lifecycle_enabled || !config.profile.enabled || !config.profile.extraction.enabled {
+        if !lifecycle_enabled || !config.profile.enabled {
             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 },
-            )
+        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<crate::worker::WorkspaceResponse, crate::worker::WorkspaceClientError> {
             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]

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<String, CatalogError> {
+        self.render(
+            WorkerPrompt::SubjektivMemoryConsolidationSystem,
+            single("language", language),
+        )
+    }
     pub fn flow_verifier_system(&self) -> Result<String, CatalogError> {
         self.render(WorkerPrompt::FlowVerifierSystem, Value::UNDEFINED)
     }

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<server_api::MemoryConsolidationResponse, server_api::RepositoryApiError>
+    {
+        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> {
     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> {
+#[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)
-        .filter(|subject_id| {
-            !subject_id.is_empty()
-                && !subject_id.chars().any(char::is_control)
-                && subject_id.trim() == *subject_id
-        })
-        .ok_or_else(|| {
-            Error::WorkspacePermissionDenied(
+    {
+        (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()
-        })
+        .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<String> {
@@ -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<crate::subjektiv::SubjectSessionAttribution> {
-    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<server_api::SubjektivMemoryBackendRequest>,
 ) -> ApiResult<Json<server_api::SubjektivMemoryBackendResponse>> {
     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<crate::subjektiv::CandidateDecisionRequest> {
+    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<WorkspaceApi>,
+    AxumPath(path): AxumPath<ScopedWorkspacePath>,
+    context: server_api::ServerRequestContext,
+    Json(operation): Json<MemoryConsolidateStagingOperation>,
+) -> ApiResult<Json<MemoryConsolidationOutput>> {
+    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<MemoryConsolidationOutput> {
+    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<Option<MemoryConsolidationOutput>> {
+    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<WorkspaceApi>,
     AxumPath(path): AxumPath<ScopedWorkspacePath>,
@@ -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"

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<MemoryRevisionRef>,
+    },
+}
+
+#[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<MemoryRecord>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub operation: Option<MemoryDecisionOperation>,
+    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<Vec<SubjectStagingRecord>> {
+        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<MemoryRecord> {
         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<String>,
     ) -> Result<(MemoryRecord, Vec<StagingResolution>)> {
         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)?
-                }
-                MemoryRevisionTarget::Revise {
-                    memory_id,
-                    expected_revision,
-                } => write_memory_revision(
+            apply_candidates_in_transaction(
                 transaction,
                 subject_id,
-                    memory_id,
-                    Some(*expected_revision),
+                candidate_ids,
+                &target,
                 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::<Result<Vec<_>>>()?;
-            Ok((memory, resolutions))
+    }
+
+    pub fn decide_candidate(
+        &self,
+        subject_id: &str,
+        request: CandidateDecisionRequest,
+    ) -> Result<CandidateDecisionReceipt> {
+        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);
+                }
+                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_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<StagingResolution>)> {
+    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::<Result<Vec<_>>>()?;
+    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();

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": {

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; };
+    };
+}

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;

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.

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

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

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

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

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