From 6091fafcacbc01227252eabc9de825a64ed5f6bc Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 2 Oct 2026 06:00:31 +0900 Subject: [PATCH] feat: connect subjektiv extraction lifecycle --- crates/manifest/src/config.rs | 107 +++- crates/manifest/src/lib.rs | 83 ++- crates/manifest/src/profile.rs | 5 + crates/server-api/src/lib.rs | 152 +++++ crates/worker-runtime/src/worker_backend.rs | 63 +- crates/worker-runtime/src/worker_source.rs | 12 + crates/worker/src/controller.rs | 28 + .../src/feature/builtin/memory_lifecycle.rs | 411 ++++++++++++- .../feature/builtin/memory_staging_output.rs | 186 +++++- crates/worker/src/worker.rs | 8 + crates/workspace-server/src/server.rs | 521 ++++++++++++++++ crates/workspace-server/src/subjektiv.rs | 562 +++++++++++++++++- docs/design/subjektiv-store.md | 47 +- 13 files changed, 2082 insertions(+), 103 deletions(-) diff --git a/crates/manifest/src/config.rs b/crates/manifest/src/config.rs index 75caeee9d..d4e135111 100644 --- a/crates/manifest/src/config.rs +++ b/crates/manifest/src/config.rs @@ -19,9 +19,10 @@ use crate::{ CompactionConfig, EngineManifest, FeatureConfig, FeatureFlagConfig, FileUploadLimits, McpConfig, McpEnvValue, McpStdioCwdPolicy, MemoryConsolidationProfileConfig, MemoryExtractionProfileConfig, MemoryFeatureProfileConfig, MemoryResidentProfileConfig, - MergeRequestFeatureConfig, ResolvedMemoryFeatureConfig, ScopeConfig, SessionConfig, - SkillsConfig, TicketFeatureConfig, ToolOutputLimits, ToolPermissionConfig, ToolPermissionRule, - WebConfig, WorkerFeatureConfig, WorkerManifest, WorkerMeta, + MergeRequestFeatureConfig, ResolvedMemoryFeatureConfig, ResolvedSubjektivFeatureConfig, + ScopeConfig, SessionConfig, SkillsConfig, SubjektivFeatureProfileConfig, TicketFeatureConfig, + ToolOutputLimits, ToolPermissionConfig, ToolPermissionRule, WebConfig, WorkerFeatureConfig, + WorkerManifest, WorkerMeta, }; /// Partial-form Worker manifest. Every field is optional; one or more @@ -76,6 +77,8 @@ pub struct FeatureConfigPartial { #[serde(default)] pub memory: Option, #[serde(default)] + pub subjektiv: Option, + #[serde(default)] pub web: Option, #[serde(default)] pub image: Option, @@ -104,6 +107,11 @@ impl FeatureConfigPartial { Self { task: merge_option(self.task, other.task, FeatureFlagConfigPartial::merge), memory: merge_option(self.memory, other.memory, MemoryFeatureConfigPartial::merge), + subjektiv: merge_option( + self.subjektiv, + other.subjektiv, + SubjektivFeatureConfigPartial::merge, + ), web: merge_option(self.web, other.web, FeatureFlagConfigPartial::merge), image: merge_option(self.image, other.image, FeatureFlagConfigPartial::merge), sub_worker: merge_option( @@ -227,6 +235,28 @@ pub struct MemoryConsolidationProfileConfigPartial { pub request_enabled: Option, } +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SubjektivFeatureConfigPartial { + #[serde(default)] + pub enabled: Option, + #[serde(default)] + pub extraction: Option, +} + +impl SubjektivFeatureConfigPartial { + fn merge(self, other: Self) -> Self { + Self { + enabled: other.enabled.or(self.enabled), + extraction: merge_option( + self.extraction, + other.extraction, + MemoryExtractionProfileConfigPartial::merge, + ), + } + } +} + impl MemoryFeatureConfigPartial { fn merge(self, other: Self) -> Self { Self { @@ -331,6 +361,10 @@ impl From for FeatureConfig { .memory .map(ResolvedMemoryFeatureConfig::from) .unwrap_or_default(), + subjektiv: value + .subjektiv + .map(ResolvedSubjektivFeatureConfig::from) + .unwrap_or_default(), web: value.web.map(FeatureFlagConfig::from).unwrap_or_default(), image: value.image.map(FeatureFlagConfig::from).unwrap_or_default(), sub_worker: value @@ -447,6 +481,42 @@ impl From for MemoryFeatureConfigPartial { } } +impl From for ResolvedSubjektivFeatureConfig { + fn from(value: SubjektivFeatureConfigPartial) -> Self { + let extraction = value.extraction.unwrap_or_default(); + Self { + profile: SubjektivFeatureProfileConfig { + enabled: value.enabled.unwrap_or_default(), + extraction: MemoryExtractionProfileConfig { + enabled: extraction.enabled.unwrap_or(true), + model: extraction.model, + reasoning: extraction.reasoning, + threshold: extraction.threshold.or(Some(50_000)), + worker_max_turns: extraction + .worker_max_turns + .or(defaults::MEMORY_EXTRACT_WORKER_MAX_TURNS), + }, + }, + workspace_settings: None, + } + } +} + +impl From for SubjektivFeatureConfigPartial { + fn from(value: ResolvedSubjektivFeatureConfig) -> Self { + Self { + enabled: Some(value.profile.enabled), + extraction: Some(MemoryExtractionProfileConfigPartial { + enabled: Some(value.profile.extraction.enabled), + model: value.profile.extraction.model, + reasoning: value.profile.extraction.reasoning, + threshold: value.profile.extraction.threshold, + worker_max_turns: value.profile.extraction.worker_max_turns, + }), + } + } +} + impl From for TicketFeatureConfig { fn from(value: TicketFeatureConfigPartial) -> Self { Self { @@ -500,6 +570,7 @@ impl From for FeatureConfigPartial { Self { task: Some(value.task.into()), memory: Some(value.memory.into()), + subjektiv: Some(value.subjektiv.into()), web: Some(value.web.into()), image: Some(value.image.into()), sub_worker: Some(value.sub_worker.into()), @@ -2490,6 +2561,36 @@ directories = [".claude/skills", ".cursor/skills"] ); } + #[test] + fn subjektiv_config_is_explicit_and_roundtrips_without_subject_identity() { + let config = WorkerManifestConfig::from_toml( + r#" +[worker] +name = "subject-worker" + +[feature.subjektiv] +enabled = true + +[feature.subjektiv.extraction] +enabled = true +threshold = 321 +worker_max_turns = 7 +"#, + ) + .unwrap(); + let partial = config.feature.subjektiv.unwrap(); + assert_eq!(partial.enabled, Some(true)); + assert_eq!(partial.extraction.as_ref().unwrap().threshold, Some(321)); + + let resolved = ResolvedSubjektivFeatureConfig::from(partial); + assert!(resolved.profile.enabled); + assert_eq!(resolved.profile.extraction.threshold, Some(321)); + assert_eq!(resolved.profile.extraction.worker_max_turns, Some(7)); + let encoded = serde_json::to_value(SubjektivFeatureConfigPartial::from(resolved)).unwrap(); + assert!(encoded.get("subject_id").is_none()); + assert_eq!(encoded["enabled"], true); + } + #[test] fn merge_preserves_ref() { let lower = WorkerManifestConfig { diff --git a/crates/manifest/src/lib.rs b/crates/manifest/src/lib.rs index 5a53f084c..9986dad68 100644 --- a/crates/manifest/src/lib.rs +++ b/crates/manifest/src/lib.rs @@ -108,6 +108,11 @@ pub struct FeatureConfig { pub task: FeatureFlagConfig, #[serde(default)] pub memory: ResolvedMemoryFeatureConfig, + /// Subject-scoped Memory extraction. Subject identity is resolved by the + /// Workspace host from the Worker's keyed singleton lease, never from this + /// model-visible manifest. + #[serde(default)] + pub subjektiv: ResolvedSubjektivFeatureConfig, #[serde(default)] pub web: FeatureFlagConfig, #[serde(default)] @@ -139,6 +144,7 @@ impl Default for FeatureConfig { Self { task: FeatureFlagConfig::disabled(), memory: ResolvedMemoryFeatureConfig::default(), + subjektiv: ResolvedSubjektivFeatureConfig::default(), web: FeatureFlagConfig::disabled(), image: FeatureFlagConfig::disabled(), sub_worker: FeatureFlagConfig::disabled(), @@ -350,6 +356,79 @@ impl ResolvedMemoryFeatureConfig { } } +/// Profile-owned subjektiv activation and extraction policy. The subject id is +/// deliberately absent: the Workspace host binds it from `subjektiv:` +/// singleton ownership for each operation. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)] +#[serde(default, deny_unknown_fields)] +pub struct SubjektivFeatureProfileConfig { + pub enabled: bool, + pub extraction: MemoryExtractionProfileConfig, +} + +impl Default for SubjektivFeatureProfileConfig { + fn default() -> Self { + Self { + enabled: false, + extraction: MemoryExtractionProfileConfig::default(), + } + } +} + +/// Immutable subjektiv execution configuration persisted in a resolved Worker +/// Manifest. Workspace settings provide only trusted extraction language and +/// are not subject identity. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Default)] +#[serde(default, deny_unknown_fields)] +pub struct ResolvedSubjektivFeatureConfig { + pub profile: SubjektivFeatureProfileConfig, + #[serde(skip_serializing_if = "Option::is_none")] + pub workspace_settings: Option, +} + +impl ResolvedSubjektivFeatureConfig { + pub fn enabled(&self) -> bool { + self.profile.enabled + } + + pub fn bind_workspace_settings( + &mut self, + settings: WorkspaceMemorySettingsSnapshot, + ) -> Result<(), &'static str> { + if !self.profile.enabled { + if self.workspace_settings.is_some() { + return Err("disabled subjektiv feature must not carry Workspace settings"); + } + return Ok(()); + } + if self.workspace_settings.is_some() { + return Err("subjektiv Workspace settings are already bound"); + } + self.workspace_settings = Some(settings); + Ok(()) + } + + pub fn workspace_settings(&self) -> Option { + self.workspace_settings.clone() + } + + pub fn validate_execution(&self) -> Result<(), &'static str> { + if self.profile.enabled && self.workspace_settings.is_none() { + return Err("enabled subjektiv feature requires trusted Workspace settings"); + } + if !self.profile.enabled && self.workspace_settings.is_some() { + return Err("disabled subjektiv feature must not carry Workspace settings"); + } + if let Some(settings) = &self.workspace_settings + && (settings.settings_revision == 0 + || !is_normalized_workspace_memory_language(&settings.language)) + { + return Err("subjektiv Workspace settings snapshot metadata is invalid"); + } + Ok(()) + } +} + #[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq, Default)] pub struct TicketFeatureConfig { #[serde(default)] @@ -963,7 +1042,9 @@ impl Default for CompactionConfig { impl WorkerManifest { pub fn requires_persisted_execution_snapshot(&self) -> bool { - self.profile.is_some() || self.feature.memory.workspace_settings.is_some() + self.profile.is_some() + || self.feature.memory.workspace_settings.is_some() + || self.feature.subjektiv.workspace_settings.is_some() } /// Parse a manifest from a TOML string. diff --git a/crates/manifest/src/profile.rs b/crates/manifest/src/profile.rs index b55df5678..2e59e18ff 100644 --- a/crates/manifest/src/profile.rs +++ b/crates/manifest/src/profile.rs @@ -144,6 +144,7 @@ pub enum WorkspaceAuthorityRequirement { Flow, ManageWorkdir, Memory, + Subjektiv, MergeRequest, Objective, Orchestration, @@ -157,6 +158,7 @@ impl fmt::Display for WorkspaceAuthorityRequirement { Self::Flow => formatter.write_str("feature.flow"), Self::ManageWorkdir => formatter.write_str("feature.manage_workdir"), Self::Memory => formatter.write_str("feature.memory"), + Self::Subjektiv => formatter.write_str("feature.subjektiv"), Self::MergeRequest => formatter.write_str("feature.merge_request"), Self::Objective => formatter.write_str("feature.objective"), Self::Orchestration => formatter.write_str("feature.orchestration"), @@ -185,6 +187,9 @@ pub fn validate_profile_execution_target( if feature.memory.profile.enabled || feature.memory.profile.staging_tools { requirements.insert(WorkspaceAuthorityRequirement::Memory); } + if feature.subjektiv.profile.enabled { + requirements.insert(WorkspaceAuthorityRequirement::Subjektiv); + } if feature.merge_request.show || feature.merge_request.open || feature.merge_request.review diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 7bed30902..cf2fbdb9c 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -234,6 +234,13 @@ impl_openapi_schema!( MemoryBackendResponse, MemoryConsolidateStagingRequest, MemoryConsolidationResponse, + SubjektivSubjectCreateRequest, + SubjektivSubjectResponse, + SubjektivSubjectWorkerStartRequest, + SubjektivStageCandidateRequest, + SubjektivStageCandidateResponse, + SubjektivRecordSessionRequest, + SubjektivRecordSessionResponse, SkillCatalogResponse, SkillDetailResponse, SkillActivationResponse, @@ -1089,6 +1096,87 @@ pub trait ServerApi { #[body] request: MemoryConsolidateStagingRequest, ) -> Result; + #[post( + "/api/w/{workspace_id}/subjektiv/subjects", + status = 201, + error_status = 400, + additional_error_statuses = [401, 403, 404, 409, 500], + bearer_auth = true, + browser_auth = true, + )] + async fn subjektiv_subject_create( + &self, + #[path] workspace_id: String, + #[body] request: SubjektivSubjectCreateRequest, + ) -> Result; + #[get( + "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}", + status = 200, + error_status = 404, + additional_error_statuses = [400, 401, 403, 500], + bearer_auth = true, + browser_auth = true, + )] + async fn subjektiv_subject_get( + &self, + #[path] workspace_id: String, + #[path] subject_id: String, + ) -> Result; + #[post( + "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/retire", + status = 200, + error_status = 400, + additional_error_statuses = [401, 403, 404, 409, 500], + bearer_auth = true, + browser_auth = true, + )] + async fn subjektiv_subject_retire( + &self, + #[path] workspace_id: String, + #[path] subject_id: String, + ) -> Result; + #[post( + "/api/w/{workspace_id}/subjektiv/subjects/{subject_id}/worker", + status = 200, + error_status = 400, + additional_error_statuses = [401, 403, 404, 409, 500, 502, 503], + bearer_auth = true, + browser_auth = true, + )] + async fn subjektiv_subject_worker_start( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[path] subject_id: String, + #[body] request: SubjektivSubjectWorkerStartRequest, + ) -> Result; + #[post( + "/api/w/{workspace_id}/subjektiv/staging", + status = 200, + error_status = 400, + additional_error_statuses = [401, 403, 404, 409, 500], + openapi = false, + )] + async fn subjektiv_stage_candidate( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[body] request: SubjektivStageCandidateRequest, + ) -> Result; + #[post( + "/api/w/{workspace_id}/subjektiv/sessions", + status = 200, + error_status = 400, + additional_error_statuses = [401, 403, 404, 409, 500], + openapi = false, + )] + async fn subjektiv_record_session( + &self, + #[extension] context: ServerRequestContext, + #[path] workspace_id: String, + #[body] request: SubjektivRecordSessionRequest, + ) -> Result; + #[get( "/api/w/{workspace_id}/skills", status = 200, @@ -4356,6 +4444,68 @@ pub struct MemoryConsolidationResponse { pub total_bytes: u64, } +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum SubjektivSubjectState { + Active, + Retired, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSubjectCreateRequest { + pub role: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSubjectResponse { + pub id: String, + pub role: String, + pub state: SubjektivSubjectState, + #[schemars(range(min = 0, max = 9_007_199_254_740_991_u64))] + pub store_revision: u64, + pub created_at: String, + pub updated_at: String, + /// Current keyed-singleton owner, when one exists. This is a live Yoi + /// projection, not state duplicated in the subjektiv store. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub current_worker: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSubjectWorkerStartRequest { + pub worker: CreateWorkspaceWorkerRequest, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivStageCandidateRequest { + /// Committed parent Session identity captured by the Worker host. + pub session_id: String, + pub operation: memory::backend::MemoryStageCandidateOperation, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivStageCandidateResponse { + pub staging_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivRecordSessionRequest { + pub session_id: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivRecordSessionResponse { + pub subject_id: String, + pub session_id: String, +} + pub const WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES: usize = 128; pub const WORKSPACE_DELETION_MAX_REVISION_BYTES: usize = 128; pub const WORKSPACE_DELETION_MAX_CONFIRMATION_BYTES: usize = 256; @@ -11521,6 +11671,8 @@ mod openapi_artifact_tests { "worker_control_stop", "worker_observation_capture", "worker_observation_sessions", + "subjektiv_stage_candidate", + "subjektiv_record_session", "workspace_worker_discovery", "workspace_worker_remove", ]; diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 381a972a2..5c1673a16 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -960,14 +960,28 @@ fn bind_workspace_memory_settings( snapshot.workspace_id, workspace_api.workspace_id )); } + if manifest.feature.memory.profile.enabled { + manifest + .feature + .memory + .bind_workspace_settings(snapshot.clone()) + .map_err(str::to_string)?; + } + if manifest.feature.subjektiv.profile.enabled { + manifest + .feature + .subjektiv + .bind_workspace_settings(snapshot.clone()) + .map_err(str::to_string)?; + } manifest .feature .memory - .bind_workspace_settings(snapshot.clone()) + .validate_execution() .map_err(str::to_string)?; manifest .feature - .memory + .subjektiv .validate_execution() .map_err(str::to_string)?; Ok(()) @@ -985,24 +999,35 @@ fn validate_worker_memory_settings( .memory .validate_execution() .map_err(str::to_string)?; - if !manifest.feature.memory.profile.enabled { - return Ok(()); - } - let actual = manifest + manifest .feature - .memory - .workspace_settings() - .ok_or_else(|| { - "Workspace Worker restored without its bound Memory settings snapshot".to_string() - })?; - if &actual != expected { - return Err(format!( - "Workspace Worker Memory settings snapshot mismatch: expected {} revision {}, restored {} revision {}", - expected.workspace_id, - expected.settings_revision, - actual.workspace_id, - actual.settings_revision - )); + .subjektiv + .validate_execution() + .map_err(str::to_string)?; + for (feature, enabled, actual) in [ + ( + "Memory", + manifest.feature.memory.profile.enabled, + manifest.feature.memory.workspace_settings(), + ), + ( + "subjektiv", + manifest.feature.subjektiv.profile.enabled, + manifest.feature.subjektiv.workspace_settings(), + ), + ] { + if enabled && actual.as_ref() != Some(expected) { + let actual = actual.ok_or_else(|| { + format!("Workspace Worker restored without its bound {feature} settings snapshot") + })?; + return Err(format!( + "Workspace Worker {feature} settings snapshot mismatch: expected {} revision {}, restored {} revision {}", + expected.workspace_id, + expected.settings_revision, + actual.workspace_id, + actual.settings_revision + )); + } } Ok(()) } diff --git a/crates/worker-runtime/src/worker_source.rs b/crates/worker-runtime/src/worker_source.rs index 1a7e5d4da..3c6dbfccf 100644 --- a/crates/worker-runtime/src/worker_source.rs +++ b/crates/worker-runtime/src/worker_source.rs @@ -564,6 +564,18 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient { .await, ) } + WorkspaceServerOperation::SubjektivStageCandidate(request) => { + generated_workspace_response( + client + .subjektiv_stage_candidate(workspace_id, request) + .await, + ) + } + WorkspaceServerOperation::SubjektivRecordSession(request) => { + generated_workspace_response( + client.subjektiv_record_session(workspace_id, request).await, + ) + } } }) }) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index f925b7e60..6b5b2efe9 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1242,6 +1242,18 @@ where ), ); } + let memory_profile = &worker.manifest().feature.memory.profile; + let subjektiv_profile = &worker.manifest().feature.subjektiv.profile; + if memory_profile.enabled + && memory_profile.extraction.enabled + && subjektiv_profile.enabled + && subjektiv_profile.extraction.enabled + { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "Workspace Memory and subjektiv automatic extraction cannot both be enabled for one Worker", + )); + } let memory_install_plan = crate::feature::builtin::memory::MemoryFeatureInstallPlan::prepare( worker.manifest(), worker.workspace_client_handle(), @@ -1276,6 +1288,22 @@ where { feature_registry.add_module(memory_lifecycle); } + if let Some(subjektiv_lifecycle) = + crate::feature::builtin::memory_lifecycle::SubjektivLifecycleFeature::from_resolved_config( + worker.manifest_lifecycle_features_enabled(), + worker.manifest().feature.subjektiv.clone(), + worker.committed_session_capture_handle(), + worker.session_extension_handle(), + worker.workspace_client_handle(), + spawner_manifest.clone(), + worker.llm_client_handle(), + prompts.clone(), + spawner_workspace_context.clone(), + worker.working_event_sender(), + )? + { + feature_registry.add_module(subjektiv_lifecycle); + } if sub_worker_enabled && !worker_enabled { feature_registry.add_module( crate::feature::builtin::manage_worker::sub_worker_control_feature( diff --git a/crates/worker/src/feature/builtin/memory_lifecycle.rs b/crates/worker/src/feature/builtin/memory_lifecycle.rs index a6b095c58..6e3027737 100644 --- a/crates/worker/src/feature/builtin/memory_lifecycle.rs +++ b/crates/worker/src/feature/builtin/memory_lifecycle.rs @@ -39,8 +39,16 @@ use manifest::WorkerManifest; use protocol::Event; const TASK_NAME: &str = "memory-lifecycle"; +const SUBJEKTIV_TASK_NAME: &str = "subjektiv-memory-lifecycle"; +const SUBJEKTIV_EXTRACT_DOMAIN: &str = "subjektiv.extract.v1"; const TASK_TIMEOUT: Duration = Duration::from_secs(300); +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ExtractionTarget { + WorkspaceMemory, + Subjektiv, +} + /// Parent-Worker lifecycle Feature that observes committed runs and schedules /// bounded extraction work. It owns the Memory pointer, audit, restricted /// Internal Worker, and staging disposition; Worker core owns only generic @@ -50,9 +58,17 @@ pub(crate) struct MemoryLifecycleFeature { task: MemoryLifecycleTask, } +/// Subject-scoped extraction is a distinct Feature module even though it reuses +/// the mature committed-capture/extract/pointer lifecycle implementation. +#[derive(Clone)] +pub(crate) struct SubjektivLifecycleFeature { + task: MemoryLifecycleTask, +} + #[derive(Clone)] struct MemoryLifecycleTask { config: manifest::ResolvedMemoryFeatureConfig, + target: ExtractionTarget, capture: CommittedSessionCaptureHandle, extensions: SessionExtensionHandle, workspace_client: Arc, @@ -120,6 +136,7 @@ impl MemoryLifecycleFeature { Self { task: MemoryLifecycleTask { config, + target: ExtractionTarget::WorkspaceMemory, capture, extensions, workspace_client, @@ -133,6 +150,63 @@ impl MemoryLifecycleFeature { } } +impl SubjektivLifecycleFeature { + #[allow(clippy::too_many_arguments)] + pub(crate) fn from_resolved_config( + lifecycle_enabled: bool, + config: manifest::ResolvedSubjektivFeatureConfig, + capture: CommittedSessionCaptureHandle, + extensions: SessionExtensionHandle, + workspace_client: Arc, + manifest: WorkerManifest, + client: Box, + prompts: Arc>, + workspace_context: WorkerWorkspaceContext, + event_tx: Option>, + ) -> std::io::Result> { + if !lifecycle_enabled || !config.profile.enabled || !config.profile.extraction.enabled { + return Ok(None); + } + config + .validate_execution() + .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?; + if !workspace_client.is_available() || workspace_client.workspace_id().is_none() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "subjektiv extraction requires Backend Workspace API authority", + )); + } + let memory_config = manifest::ResolvedMemoryFeatureConfig { + profile: manifest::MemoryFeatureProfileConfig { + enabled: true, + staging_tools: false, + resident: manifest::MemoryResidentProfileConfig { + inject_summary: false, + }, + extraction: config.profile.extraction, + consolidation: manifest::MemoryConsolidationProfileConfig { + request_enabled: false, + }, + }, + workspace_settings: config.workspace_settings, + }; + Ok(Some(Self { + task: MemoryLifecycleTask { + config: memory_config, + target: ExtractionTarget::Subjektiv, + capture, + extensions, + workspace_client, + manifest, + client, + prompts, + workspace_context, + event_tx, + }, + })) + } +} + impl FeatureModule for MemoryLifecycleFeature { fn descriptor(&self) -> FeatureDescriptor { FeatureDescriptor::builtin("memory-lifecycle", "Memory Lifecycle") @@ -152,17 +226,54 @@ impl FeatureModule for MemoryLifecycleFeature { } } +impl FeatureModule for SubjektivLifecycleFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("subjektiv-memory-lifecycle", "subjektiv Memory Lifecycle") + .with_description( + "Observes terminal committed parent runs and stages bounded Memory candidates for the host-connected subject.", + ) + .with_background_task(BackgroundTaskDeclaration::worker_managed( + SUBJEKTIV_TASK_NAME, + "Extract provenance-preserving candidates into subject-scoped staging after committed runs.", + )) + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + context + .background_tasks() + .register(subjektiv_lifecycle_task_spec(), self.task.clone()) + } +} + fn memory_lifecycle_task_spec() -> BackgroundTaskSpec { - let declaration = BackgroundTaskDeclaration::worker_managed( + lifecycle_task_spec( TASK_NAME, "Extract provenance-preserving Memory candidates and request Backend consolidation after committed runs.", - ); + ) +} + +fn subjektiv_lifecycle_task_spec() -> BackgroundTaskSpec { + lifecycle_task_spec( + SUBJEKTIV_TASK_NAME, + "Extract provenance-preserving candidates into subject-scoped staging after committed runs.", + ) +} + +fn lifecycle_task_spec(name: &'static str, description: &'static str) -> BackgroundTaskSpec { + let declaration = BackgroundTaskDeclaration::worker_managed(name, description); let mut spec = BackgroundTaskSpec::single_flight(declaration, TASK_TIMEOUT); spec.trigger = BackgroundTaskTrigger::RunCommitted; spec } impl MemoryLifecycleTask { + fn pointer_domain(&self) -> &'static str { + match self.target { + ExtractionTarget::WorkspaceMemory => extract::EXTRACT_DOMAIN, + ExtractionTarget::Subjektiv => SUBJEKTIV_EXTRACT_DOMAIN, + } + } + fn extraction_manifest(&self) -> WorkerManifest { let mut manifest = self.manifest.clone(); if let Some(model) = self.config.profile.extraction.model.clone() { @@ -192,6 +303,11 @@ impl MemoryLifecycleTask { .map(model_audit_from_manifest), ) .with_memory_settings(&self.config); + let audit = if self.target == ExtractionTarget::Subjektiv { + audit.without_legacy_backend() + } else { + audit + }; let capture = match self.capture.capture() { Ok(capture) => capture, Err(error) => { @@ -223,7 +339,7 @@ impl MemoryLifecycleTask { .await; return Ok(()); } - let pointer = match extract_pointer(&capture) { + let pointer = match extract_pointer(&capture, self.pointer_domain()) { Ok(pointer) => pointer, Err(error) => { audit @@ -333,12 +449,21 @@ impl MemoryLifecycleTask { None, ) .await; - let output_state = MemoryStagingOutputState::new( - view.clone(), - Arc::clone(&self.workspace_client), - source, - audit.run_id.to_string(), - ); + let output_state = match self.target { + ExtractionTarget::WorkspaceMemory => MemoryStagingOutputState::new( + view.clone(), + Arc::clone(&self.workspace_client), + source, + audit.run_id.to_string(), + ), + ExtractionTarget::Subjektiv => MemoryStagingOutputState::new_subjektiv( + view.clone(), + Arc::clone(&self.workspace_client), + source, + audit.run_id.to_string(), + capture.session_id.clone(), + ), + }; let client = if let Some(model) = self.config.profile.extraction.model.as_ref() { match crate::model_client::build_client(model) { Ok(client) => client, @@ -485,6 +610,23 @@ impl MemoryLifecycleTask { ExtractionDisposition::Completed => {} } + if self.target == ExtractionTarget::Subjektiv + && let Err(reason) = self.record_subjektiv_session(&capture.session_id) + { + audit + .emit( + self.workspace_client.as_ref(), + self.event_tx.as_ref(), + memory::audit::WorkerLifecycleStatus::Failed, + reason, + usage_audit, + extract_audit, + None, + ) + .await; + return Ok(()); + } + context.generation_fence.ensure_current()?; let next_pointer = memory::ExtractPointerPayload { processed_through_entry: capture.entry_count - 1, @@ -494,7 +636,7 @@ impl MemoryLifecycleTask { let payload = serde_json::to_value(&next_pointer).map_err(hook_internal)?; if !self .extensions - .append_if_current(&capture.location(), extract::EXTRACT_DOMAIN, payload) + .append_if_current(&capture.location(), self.pointer_domain(), payload) .map_err(hook_internal)? { audit @@ -549,6 +691,35 @@ impl FeatureBackgroundTask for MemoryLifecycleTask { } impl MemoryLifecycleTask { + fn record_subjektiv_session(&self, session_id: &str) -> Result<(), String> { + let response = self + .workspace_client + .execute_server_operation( + crate::worker::WorkspaceServerOperation::SubjektivRecordSession( + server_api::SubjektivRecordSessionRequest { + session_id: session_id.to_string(), + }, + ), + ) + .map_err(|error| format!("record subjektiv Session attribution failed: {error}"))?; + if !response.is_success() { + return Err(format!( + "record subjektiv Session attribution returned HTTP {}: {}", + response.status, response.body + )); + } + let output: server_api::SubjektivRecordSessionResponse = + serde_json::from_str(&response.body).map_err(|error| { + format!("decode subjektiv Session attribution response: {error}") + })?; + if output.session_id != session_id { + return Err( + "subjektiv Session attribution response changed session identity".to_string(), + ); + } + Ok(()) + } + async fn request_consolidation(&self) { let audit = WorkerAuditBase::new( memory::audit::AuditWorker::MemoryConsolidation, @@ -668,13 +839,19 @@ fn hook_internal(error: impl std::fmt::Display) -> HookError { fn extract_pointer( capture: &CommittedSessionCapture, + domain: &str, ) -> Result, HookError> { - let pointer = memory::extract::fold_pointer(&capture.extensions); + let pointer = capture + .extensions + .iter() + .rev() + .find(|(candidate, _)| candidate == domain) + .and_then(|(_, payload)| serde_json::from_value(payload.clone()).ok()); if pointer.is_none() && capture .extensions .iter() - .any(|(domain, _)| domain == extract::EXTRACT_DOMAIN) + .any(|(candidate, _)| candidate == domain) { return Err(hook_internal( "latest committed Memory extraction pointer is malformed", @@ -732,6 +909,7 @@ struct WorkerAuditBase { trigger: memory::audit::AuditTrigger, memory_settings: Option, model: Option, + persist_legacy_backend: bool, } impl WorkerAuditBase { @@ -746,9 +924,15 @@ impl WorkerAuditBase { trigger, memory_settings: None, model, + persist_legacy_backend: true, } } + fn without_legacy_backend(mut self) -> Self { + self.persist_legacy_backend = false; + self + } + fn with_memory_settings(mut self, config: &manifest::ResolvedMemoryFeatureConfig) -> Self { self.memory_settings = config @@ -784,15 +968,19 @@ impl WorkerAuditBase { extract, consolidation, }; - let _ = workspace_client - .execute_memory_backend_operation(memory::backend::MemoryBackendOperation::AppendAudit( - memory::backend::MemoryAppendAuditOperation { - event: memory::audit::AuditEvent::new( - memory::audit::AuditPayload::WorkerLifecycle(payload), + if self.persist_legacy_backend { + let _ = workspace_client + .execute_memory_backend_operation( + memory::backend::MemoryBackendOperation::AppendAudit( + memory::backend::MemoryAppendAuditOperation { + event: memory::audit::AuditEvent::new( + memory::audit::AuditPayload::WorkerLifecycle(payload), + ), + }, ), - }, - )) - .await; + ) + .await; + } if let Some(tx) = event_tx { let _ = tx.send(Event::MemoryWorker(protocol::MemoryWorkerEvent { worker: self.worker.label().to_string(), @@ -859,6 +1047,8 @@ mod tests { &self, request: crate::worker::WorkspaceRequest, ) -> Result { + let is_subjektiv_stage = request.path.ends_with("/subjektiv/staging"); + let is_subjektiv_session = request.path.ends_with("/subjektiv/sessions"); let is_stage_candidate = request .body .as_deref() @@ -868,6 +1058,25 @@ mod tests { .as_deref() .is_some_and(|body| body.contains("append_audit")); self.requests.lock().unwrap().push(request); + if is_subjektiv_session { + return Ok(crate::worker::WorkspaceResponse { + status: 200, + body: serde_json::to_string(&server_api::SubjektivRecordSessionResponse { + subject_id: "subject-1".to_string(), + session_id: "session-1".to_string(), + }) + .unwrap(), + }); + } + if is_subjektiv_stage { + return Ok(crate::worker::WorkspaceResponse { + status: 200, + body: serde_json::to_string(&server_api::SubjektivStageCandidateResponse { + staging_id: "subject-candidate-1".to_string(), + }) + .unwrap(), + }); + } if is_stage_candidate { return Ok(crate::worker::WorkspaceResponse { status: 200, @@ -1072,6 +1281,7 @@ permission = "write" }); MemoryLifecycleTask { config: test_config(), + target: ExtractionTarget::WorkspaceMemory, capture: capture_handle, extensions, workspace_client, @@ -1122,6 +1332,30 @@ permission = "write" registry.shutdown().await.unwrap(); } + #[test] + fn disabled_subjektiv_feature_installs_nothing_without_workspace_side_effects() { + let capture = CommittedSessionCaptureHandle::new(|| { + panic!("disabled subjektiv must not capture a Session") + }); + let extensions = SessionExtensionHandle::new(|_, _, _| { + panic!("disabled subjektiv must not write a pointer") + }); + let feature = SubjektivLifecycleFeature::from_resolved_config( + true, + manifest::ResolvedSubjektivFeatureConfig::default(), + capture, + extensions, + crate::worker::marker_workspace_client(None, "disabled"), + test_manifest(), + Box::new(ScriptClient::new(Vec::new())), + Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())), + WorkerWorkspaceContext::no_workspace(), + None, + ) + .unwrap(); + assert!(feature.is_none()); + } + #[test] fn extraction_manifest_applies_memory_model_and_reasoning_overrides() { use agen::llm_client::capability::{ReasoningControl, ReasoningEffort}; @@ -1236,6 +1470,136 @@ permission = "write" ); } + #[tokio::test] + async fn subjektiv_extraction_stages_once_and_commits_its_independent_pointer() { + let source = capture(2, 250); + let entry_ref = + SessionCapture::from_history_entries(source.segment_id.clone(), source.history.clone()) + .overview()[0] + .id + .to_string(); + let client = ScriptClient::new(vec![ + stage_candidate_events("stage-1", &entry_ref), + finish_events("finish-1", 1), + completed_events(), + ]); + let extension_writes = Arc::new(Mutex::new(Vec::new())); + let (event_tx, _) = broadcast::channel(64); + let workspace_client = Arc::new(RecordingWorkspaceClient::default()); + let mut task = test_task( + source, + Box::new(client), + Arc::clone(&extension_writes), + event_tx, + workspace_client.clone(), + ); + task.target = ExtractionTarget::Subjektiv; + task.config.profile.consolidation.request_enabled = false; + run_background_task(task).await; + + let writes = extension_writes.lock().unwrap(); + assert_eq!(writes.len(), 1); + assert_eq!(writes[0].1, SUBJEKTIV_EXTRACT_DOMAIN); + let pointer: memory::ExtractPointerPayload = + serde_json::from_value(writes[0].2.clone()).unwrap(); + assert_eq!(pointer.staging_id, "subject-candidate-1"); + let requests = workspace_client.requests.lock().unwrap(); + assert_eq!( + requests + .iter() + .filter(|request| request.path.ends_with("/subjektiv/staging")) + .count(), + 1 + ); + assert_eq!( + requests + .iter() + .filter(|request| request.path.ends_with("/subjektiv/sessions")) + .count(), + 1 + ); + assert!(requests.iter().all(|request| { + request + .body + .as_deref() + .is_none_or(|body| !body.contains("append_audit") && body != "{\"force\":false}") + })); + } + + #[tokio::test] + async fn empty_subjektiv_extraction_records_session_and_commits_pointer_without_candidate() { + let client = ScriptClient::new(vec![finish_empty_events("finish-1"), completed_events()]); + let extension_writes = Arc::new(Mutex::new(Vec::new())); + let (event_tx, _) = broadcast::channel(32); + let workspace_client = Arc::new(RecordingWorkspaceClient::default()); + let mut task = test_task( + capture(2, 250), + Box::new(client), + Arc::clone(&extension_writes), + event_tx, + workspace_client.clone(), + ); + task.target = ExtractionTarget::Subjektiv; + task.config.profile.consolidation.request_enabled = false; + run_background_task(task).await; + + let writes = extension_writes.lock().unwrap(); + assert_eq!(writes.len(), 1); + assert_eq!(writes[0].1, SUBJEKTIV_EXTRACT_DOMAIN); + let pointer: memory::ExtractPointerPayload = + serde_json::from_value(writes[0].2.clone()).unwrap(); + assert!(pointer.staging_id.is_empty()); + let requests = workspace_client.requests.lock().unwrap(); + assert!( + !requests + .iter() + .any(|request| request.path.ends_with("/subjektiv/staging")) + ); + assert!( + requests + .iter() + .any(|request| request.path.ends_with("/subjektiv/sessions")) + ); + } + + #[tokio::test] + async fn subjektiv_candidate_remains_staged_when_later_extraction_fails_without_pointer() { + let source = capture(2, 250); + let entry_ref = + SessionCapture::from_history_entries(source.segment_id.clone(), source.history.clone()) + .overview()[0] + .id + .to_string(); + let client = ScriptClient::new(vec![stage_candidate_events("stage-1", &entry_ref)]); + let extension_writes = Arc::new(Mutex::new(Vec::new())); + let (event_tx, _) = broadcast::channel(32); + let workspace_client = Arc::new(RecordingWorkspaceClient::default()); + let mut task = test_task( + source, + Box::new(client), + Arc::clone(&extension_writes), + event_tx, + workspace_client.clone(), + ); + task.target = ExtractionTarget::Subjektiv; + task.config.profile.consolidation.request_enabled = false; + run_background_task(task).await; + + assert!(extension_writes.lock().unwrap().is_empty()); + let requests = workspace_client.requests.lock().unwrap(); + assert_eq!( + requests + .iter() + .filter(|request| request.path.ends_with("/subjektiv/staging")) + .count(), + 1, + "the immutable partial candidate remains staged for retry/deduplication" + ); + assert!(!requests + .iter() + .any(|request| request.path.ends_with("/subjektiv/sessions"))); + } + #[tokio::test] async fn failed_extraction_emits_failure_event_and_durable_audit_without_pointer() { let client = ScriptClient::new(Vec::new()); @@ -1562,7 +1926,10 @@ permission = "write" serde_json::to_value(&latest).unwrap(), ), ]; - assert_eq!(extract_pointer(&capture).unwrap(), Some(latest)); + assert_eq!( + extract_pointer(&capture, extract::EXTRACT_DOMAIN).unwrap(), + Some(latest) + ); } #[test] @@ -1583,7 +1950,7 @@ permission = "write" serde_json::json!({"invalid": true}), ), ]; - assert!(extract_pointer(&capture).is_err()); + assert!(extract_pointer(&capture, extract::EXTRACT_DOMAIN).is_err()); } #[test] diff --git a/crates/worker/src/feature/builtin/memory_staging_output.rs b/crates/worker/src/feature/builtin/memory_staging_output.rs index 4ccc90afe..2b9b7a09b 100644 --- a/crates/worker/src/feature/builtin/memory_staging_output.rs +++ b/crates/worker/src/feature/builtin/memory_staging_output.rs @@ -33,10 +33,17 @@ pub(crate) struct MemoryStagingOutputState { workspace_client: Arc, source: SourceRef, extract_run_id: String, + destination: MemoryStagingDestination, staged: Arc>>, finished: Arc>>, } +#[derive(Clone)] +enum MemoryStagingDestination { + WorkspaceMemory, + Subjektiv { session_id: String }, +} + impl MemoryStagingOutputState { pub(crate) fn new( view: SessionCapture, @@ -49,6 +56,25 @@ impl MemoryStagingOutputState { workspace_client, source, extract_run_id, + destination: MemoryStagingDestination::WorkspaceMemory, + staged: Arc::new(Mutex::new(Vec::new())), + finished: Arc::new(Mutex::new(None)), + } + } + + pub(crate) fn new_subjektiv( + view: SessionCapture, + workspace_client: Arc, + source: SourceRef, + extract_run_id: String, + session_id: String, + ) -> Self { + Self { + view: Arc::new(view), + workspace_client, + source, + extract_run_id, + destination: MemoryStagingDestination::Subjektiv { session_id }, staged: Arc::new(Mutex::new(Vec::new())), finished: Arc::new(Mutex::new(None)), } @@ -201,34 +227,72 @@ impl Tool for StageMemoryCandidateTool { staleness: params.staleness, evidence_ids: params.entry_refs, }; - let result = self - .state - .workspace_client - .execute_memory_backend_operation(MemoryBackendOperation::StageCandidate( - MemoryStageCandidateOperation { - source: self.state.source.clone(), - extract_run_id: self.state.extract_run_id.clone(), - candidate, - evidence, - source_refs, - }, - )) - .await - .map_err(map_memory_stage_error)?; - let staging_ids = match result { - MemoryBackendOperationResult::StagingWritten(output) if output.staging_count == 1 => { - output.staging_ids + let operation = MemoryStageCandidateOperation { + source: self.state.source.clone(), + extract_run_id: self.state.extract_run_id.clone(), + candidate, + evidence, + source_refs, + }; + let staging_ids = match &self.state.destination { + MemoryStagingDestination::WorkspaceMemory => { + let result = self + .state + .workspace_client + .execute_memory_backend_operation(MemoryBackendOperation::StageCandidate( + operation, + )) + .await + .map_err(map_memory_stage_error)?; + match result { + MemoryBackendOperationResult::StagingWritten(output) + if output.staging_count == 1 => + { + output.staging_ids + } + MemoryBackendOperationResult::StagingWritten(output) => { + return Err(ToolError::ExecutionFailed(format!( + "StageMemoryCandidate expected one staging record, backend wrote {}", + output.staging_count + ))); + } + other => { + return Err(ToolError::ExecutionFailed(format!( + "unexpected Memory backend result for StageMemoryCandidate: {other:?}" + ))); + } + } } - MemoryBackendOperationResult::StagingWritten(output) => { - return Err(ToolError::ExecutionFailed(format!( - "StageMemoryCandidate expected one staging record, backend wrote {}", - output.staging_count - ))); - } - other => { - return Err(ToolError::ExecutionFailed(format!( - "unexpected Memory backend result for StageMemoryCandidate: {other:?}" - ))); + MemoryStagingDestination::Subjektiv { session_id } => { + let response = self + .state + .workspace_client + .execute_server_operation( + crate::worker::WorkspaceServerOperation::SubjektivStageCandidate( + server_api::SubjektivStageCandidateRequest { + session_id: session_id.clone(), + operation, + }, + ), + ) + .map_err(|error| { + ToolError::ExecutionFailed(format!( + "write subjektiv staging failed: {error}" + )) + })?; + if !response.is_success() { + return Err(ToolError::ExecutionFailed(format!( + "write subjektiv staging returned HTTP {}: {}", + response.status, response.body + ))); + } + let output: server_api::SubjektivStageCandidateResponse = + serde_json::from_str(&response.body).map_err(|error| { + ToolError::ExecutionFailed(format!( + "decode subjektiv staging response: {error}" + )) + })?; + vec![output.staging_id] } }; let staging_id = staging_ids.into_iter().next().ok_or_else(|| { @@ -425,10 +489,45 @@ fn truncate_line(text: &str, max_chars: usize) -> String { #[cfg(test)] mod tests { + use std::sync::Mutex; + use agen::Item; use super::*; + #[derive(Debug, Default)] + struct RecordingSubjektivClient { + requests: Mutex>, + } + + impl WorkspaceClient for RecordingSubjektivClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-1") + } + + fn kind(&self) -> &str { + "subjektiv-test" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &self, + request: crate::worker::WorkspaceRequest, + ) -> Result { + self.requests.lock().unwrap().push(request); + Ok(crate::worker::WorkspaceResponse { + status: 200, + body: serde_json::to_string(&server_api::SubjektivStageCandidateResponse { + staging_id: "subject-candidate-1".to_string(), + }) + .unwrap(), + }) + } + } + fn state() -> MemoryStagingOutputState { MemoryStagingOutputState::new( SessionCapture::new("segment-1", vec![Item::user_message("durable decision")]), @@ -503,6 +602,39 @@ mod tests { assert!(format!("{error:?}").contains("exclusively HumanInput evidence")); } + #[tokio::test] + async fn subjektiv_destination_sends_committed_session_without_subject_scope() { + let client = Arc::new(RecordingSubjektivClient::default()); + let state = MemoryStagingOutputState::new_subjektiv( + SessionCapture::new("segment-1", vec![Item::user_message("durable decision")]), + client.clone(), + SourceRef { + segment_id: "segment-1".to_string(), + range: [0, 0], + }, + "run-1".to_string(), + "session-committed".to_string(), + ); + let tool = StageMemoryCandidateTool { + state: state.clone(), + }; + tool.execute( + r#"{"kind":"decision","claim":"claim","why_useful":"useful","entry_refs":["E00000000"]}"#, + agen::tool::ToolExecutionContext::direct(), + ) + .await + .unwrap(); + + assert_eq!(state.staged(), vec!["subject-candidate-1"]); + let requests = client.requests.lock().unwrap(); + assert_eq!(requests.len(), 1); + assert!(requests[0].path.ends_with("/subjektiv/staging")); + let body = requests[0].body.as_deref().unwrap(); + assert!(body.contains("session-committed")); + assert!(body.contains("why_useful")); + assert!(!body.contains("subject_id")); + } + #[tokio::test] async fn stage_rejects_entry_ref_outside_capture_before_backend_mutation() { let tool = StageMemoryCandidateTool { state: state() }; diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 4313a184f..c44521d75 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -624,6 +624,8 @@ pub enum WorkspaceServerOperation { }, WorkerObservationSessions, WorkerObservationCapture(server_api::WorkerObservationSubjectRef), + SubjektivStageCandidate(server_api::SubjektivStageCandidateRequest), + SubjektivRecordSession(server_api::SubjektivRecordSessionRequest), } fn workspace_server_json_request( @@ -702,6 +704,12 @@ fn workspace_server_operation_request( WorkspaceServerOperation::WorkerObservationCapture(request) => { workspace_server_json_request(format!("{base}/worker-observation/session"), &request) } + WorkspaceServerOperation::SubjektivStageCandidate(request) => { + workspace_server_json_request(format!("{base}/subjektiv/staging"), &request) + } + WorkspaceServerOperation::SubjektivRecordSession(request) => { + workspace_server_json_request(format!("{base}/subjektiv/sessions"), &request) + } } } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 08e09b32d..d76521768 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -1388,6 +1388,7 @@ pub struct WorkspaceApi { config_schema_registry: crate::config_source::WorkspaceConfigSchemaRegistry, prompt_projection_cache: crate::prompt_settings::WorkspacePromptProjectionCache, feature_storage: crate::WorkspaceFeatureStorage, + subjektiv_registration: crate::RegisteredFeature, authority: SqliteWorkspaceAuthority, _worker_projection_shutdown: Arc, runtime: Arc, @@ -3215,6 +3216,10 @@ impl WorkspaceApi { .map_err(|error| { Error::Store(format!("failed to initialize Feature storage: {error}")) })?; + let subjektiv_registration = crate::subjektiv::SubjektivStore::register(&feature_storage) + .map_err(|error| { + Error::Store(format!("failed to register subjektiv storage: {error}")) + })?; let api = Self { config_store, repository_secrets, @@ -3223,6 +3228,7 @@ impl WorkspaceApi { prompt_projection_cache: crate::prompt_settings::WorkspacePromptProjectionCache::default(), feature_storage, + subjektiv_registration, authority: SqliteWorkspaceAuthority::new( config.database_path.clone(), config.workspace_id.clone(), @@ -5432,6 +5438,24 @@ fn generated_workspace_contract_router(service: ServerApiContractService) -> Rou .merge(server_api::server_api_axum::memory_consolidation( service.clone(), )) + .merge(server_api::server_api_axum::subjektiv_subject_create( + service.clone(), + )) + .merge(server_api::server_api_axum::subjektiv_subject_get( + service.clone(), + )) + .merge(server_api::server_api_axum::subjektiv_subject_retire( + service.clone(), + )) + .merge(server_api::server_api_axum::subjektiv_subject_worker_start( + service.clone(), + )) + .merge(server_api::server_api_axum::subjektiv_stage_candidate( + service.clone(), + )) + .merge(server_api::server_api_axum::subjektiv_record_session( + service.clone(), + )) .merge(server_api::server_api_axum::skill_list(service.clone())) .merge(server_api::server_api_axum::skill_lint(service.clone())) .merge(server_api::server_api_axum::skill_get(service.clone())) @@ -7006,6 +7030,119 @@ impl server_api::ServerApi for ServerApiContractService { .map_err(ApiError::into_repository_api_error) } + async fn subjektiv_subject_create( + &self, + workspace_id: String, + request: server_api::SubjektivSubjectCreateRequest, + ) -> std::result::Result + { + scoped_create_subjektiv_subject( + State(self.workspace_api()?.clone()), + AxumPath(ScopedWorkspacePath { workspace_id }), + Json(request), + ) + .await + .map(|(_, Json(response))| response) + .map_err(ApiError::into_repository_api_error) + } + + async fn subjektiv_subject_get( + &self, + workspace_id: String, + subject_id: String, + ) -> std::result::Result + { + scoped_get_subjektiv_subject( + State(self.workspace_api()?.clone()), + AxumPath(ScopedSubjektivSubjectPath { + workspace_id, + subject_id, + }), + ) + .await + .map(|Json(response)| response) + .map_err(ApiError::into_repository_api_error) + } + + async fn subjektiv_subject_retire( + &self, + workspace_id: String, + subject_id: String, + ) -> std::result::Result + { + scoped_retire_subjektiv_subject( + State(self.workspace_api()?.clone()), + AxumPath(ScopedSubjektivSubjectPath { + workspace_id, + subject_id, + }), + ) + .await + .map(|Json(response)| response) + .map_err(ApiError::into_repository_api_error) + } + + async fn subjektiv_subject_worker_start( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + subject_id: String, + request: server_api::SubjektivSubjectWorkerStartRequest, + ) -> std::result::Result { + scoped_start_subjektiv_subject_worker( + State(self.workspace_api()?.clone()), + AxumPath(ScopedSubjektivSubjectPath { + workspace_id, + subject_id, + }), + contract_request_headers(&context)?, + Json(request), + ) + .await + .map(|Json(response)| response) + .map_err(ApiError::into_repository_api_error) + } + + async fn subjektiv_stage_candidate( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + request: server_api::SubjektivStageCandidateRequest, + ) -> std::result::Result< + server_api::SubjektivStageCandidateResponse, + server_api::RepositoryApiError, + > { + scoped_stage_subjektiv_candidate( + State(self.workspace_api()?.clone()), + AxumPath(ScopedWorkspacePath { workspace_id }), + context, + Json(request), + ) + .await + .map(|Json(response)| response) + .map_err(ApiError::into_repository_api_error) + } + + async fn subjektiv_record_session( + &self, + context: server_api::ServerRequestContext, + workspace_id: String, + request: server_api::SubjektivRecordSessionRequest, + ) -> std::result::Result< + server_api::SubjektivRecordSessionResponse, + server_api::RepositoryApiError, + > { + scoped_record_subjektiv_session( + State(self.workspace_api()?.clone()), + AxumPath(ScopedWorkspacePath { workspace_id }), + context, + Json(request), + ) + .await + .map(|Json(response)| response) + .map_err(ApiError::into_repository_api_error) + } + async fn skill_list( &self, workspace_id: String, @@ -10217,6 +10354,12 @@ struct ScopedWorkspacePath { workspace_id: String, } +#[derive(Debug, Clone, Deserialize)] +struct ScopedSubjektivSubjectPath { + workspace_id: String, + subject_id: String, +} + #[derive(Debug, Deserialize)] struct ScopedFlowPath { workspace_id: String, @@ -16352,6 +16495,255 @@ async fn scoped_memory_backend_operation( Ok(Json(response)) } +const SUBJEKTIV_SINGLETON_PREFIX: &str = "subjektiv:"; + +fn open_subjektiv_store(api: &WorkspaceApi) -> ApiResult { + crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration) + .map_err(|error| ApiError::from(Error::Store(error.to_string()))) +} + +fn subjektiv_subject_id_from_singleton_key(key: &str) -> ApiResult<&str> { + key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) + .filter(|subject_id| { + !subject_id.is_empty() + && !subject_id.chars().any(char::is_control) + && subject_id.trim() == *subject_id + }) + .ok_or_else(|| { + Error::WorkspacePermissionDenied( + "current Worker is not connected to a subjektiv subject".to_string(), + ) + .into() + }) +} + +fn subjektiv_singleton_key(subject_id: &str) -> ApiResult { + if subject_id.trim().is_empty() || subject_id.chars().any(char::is_control) { + return Err(Error::InvalidInput("subjektiv subject id is invalid".to_string()).into()); + } + let key = format!("{SUBJEKTIV_SINGLETON_PREFIX}{subject_id}"); + crate::store::validate_worker_singleton_key(&key)?; + Ok(key) +} + +fn subjektiv_session_attribution( + api: &WorkspaceApi, + workspace_id: &str, + context: &server_api::ServerRequestContext, + session_id: &str, +) -> ApiResult { + let source = context.runtime_source.as_ref().ok_or_else(|| { + Error::WorkspacePermissionDenied( + "subjektiv extraction requires authenticated Runtime-owned source authority" + .to_string(), + ) + })?; + let worker_id = source.worker_id.as_deref().ok_or_else(|| { + Error::WorkspacePermissionDenied( + "subjektiv extraction requires Runtime-bound Worker identity".to_string(), + ) + })?; + let worker = RuntimeWorkerRef::new(&source.runtime_id, worker_id); + let lease = api + .store + .require_current_worker_singleton_owner(workspace_id, &worker)? + .ok_or_else(|| { + Error::WorkspacePermissionDenied( + "current Worker has no keyed singleton ownership".to_string(), + ) + })?; + let subject_id = subjektiv_subject_id_from_singleton_key(&lease.key)?; + 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()); + } + crate::subjektiv::SubjectSessionAttribution::new( + subject_id, + worker.runtime_id, + worker.worker_id, + session_id, + ) + .map_err(|error| Error::InvalidInput(error.to_string()).into()) +} + +fn subjektiv_subject_response( + api: &WorkspaceApi, + subject: crate::subjektiv::SubjectRecord, +) -> server_api::SubjektivSubjectResponse { + let current_worker = subjektiv_singleton_key(&subject.id) + .ok() + .and_then(|key| { + api.store + .current_worker_singleton_owner(&api.config.workspace_id, &key) + .ok() + .flatten() + }) + .and_then(|lease| api.runtime.worker(&lease.worker).ok()) + .map(worker_launch_worker_summary); + server_api::SubjektivSubjectResponse { + id: subject.id, + role: subject.role.as_str().to_string(), + state: match subject.state { + crate::subjektiv::SubjectState::Active => server_api::SubjektivSubjectState::Active, + crate::subjektiv::SubjectState::Retired => server_api::SubjektivSubjectState::Retired, + }, + store_revision: subject.store_revision, + created_at: subject.created_at, + updated_at: subject.updated_at, + current_worker, + } +} + +async fn scoped_create_subjektiv_subject( + State(api): State, + AxumPath(path): AxumPath, + Json(request): Json, +) -> ApiResult<(StatusCode, Json)> { + validate_workspace_scope(&api, &path.workspace_id)?; + let role = crate::subjektiv::SubjectRole::new(request.role) + .map_err(|error| Error::InvalidInput(error.to_string()))?; + let subject = open_subjektiv_store(&api)? + .create_subject(role) + .map_err(|error| Error::Store(error.to_string()))?; + Ok(( + StatusCode::CREATED, + Json(subjektiv_subject_response(&api, subject)), + )) +} + +async fn scoped_get_subjektiv_subject( + State(api): State, + AxumPath(path): AxumPath, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let subject = open_subjektiv_store(&api)? + .subject(&path.subject_id) + .map_err(|error| Error::Store(error.to_string()))? + .ok_or_else(|| { + Error::InvalidInput(format!("unknown subjektiv subject `{}`", path.subject_id)) + })?; + Ok(Json(subjektiv_subject_response(&api, subject))) +} + +async fn scoped_retire_subjektiv_subject( + State(api): State, + AxumPath(path): AxumPath, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let subject = open_subjektiv_store(&api)? + .retire_subject(&path.subject_id) + .map_err(|error| Error::InvalidInput(error.to_string()))?; + // Retirement deliberately does not stop or remove the current Worker. The + // Worker lease and subject lifecycle are independent authorities. + Ok(Json(subjektiv_subject_response(&api, subject))) +} + +async fn scoped_start_subjektiv_subject_worker( + State(api): State, + AxumPath(path): AxumPath, + headers: HeaderMap, + Json(request): Json, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let subject = open_subjektiv_store(&api)? + .subject(&path.subject_id) + .map_err(|error| Error::Store(error.to_string()))? + .ok_or_else(|| { + Error::InvalidInput(format!("unknown subjektiv subject `{}`", path.subject_id)) + })?; + if subject.state != crate::subjektiv::SubjectState::Active { + return Err(Error::InvalidInput(format!( + "subjektiv subject `{}` is retired", + path.subject_id + )) + .into()); + } + let mut worker = request.worker; + if worker.singleton_key.is_some() { + return Err(Error::InvalidInput( + "subjektiv subject Worker start owns singleton selection; singleton_key must be omitted" + .to_string(), + ) + .into()); + } + worker.singleton_key = Some(subjektiv_singleton_key(&path.subject_id)?); + create_workspace_worker(State(api), headers, Json(worker)).await +} + +async fn scoped_stage_subjektiv_candidate( + State(api): State, + AxumPath(path): AxumPath, + context: server_api::ServerRequestContext, + Json(mut request): Json, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let attribution = + subjektiv_session_attribution(&api, &path.workspace_id, &context, &request.session_id)?; + let subject_id = attribution.subject_id.clone(); + let session_id = attribution.session_id.as_str(); + for source_ref in &mut request.operation.source_refs { + if source_ref + .session_id + .as_deref() + .is_some_and(|value| value != session_id) + { + return Err(Error::InvalidInput( + "candidate source_ref belongs to a different Session".to_string(), + ) + .into()); + } + source_ref.session_id = Some(session_id.to_string()); + } + if matches!( + request.operation.candidate.kind, + memory::extract::CandidateKind::Preference + ) && request.operation.evidence.iter().any(|evidence| { + !matches!( + evidence.origin.as_ref().map(|origin| &origin.kind), + Some(memory::schema::EvidenceOriginKind::HumanInput) + ) + }) { + return Err(Error::InvalidInput( + "preference candidates require exclusively HumanInput evidence".to_string(), + ) + .into()); + } + let candidate_id = Uuid::now_v7().to_string(); + let record = memory::extract::StagingRecord::from_candidate( + candidate_id, + request.operation.extract_run_id, + request.operation.source, + request.operation.candidate, + request.operation.evidence, + request.operation.source_refs, + ); + let record = crate::subjektiv::SubjectStagingRecord::attach(&subject_id, record); + let (staged, _) = open_subjektiv_store(&api)? + .stage_candidate_with_attribution(record, attribution) + .map_err(|error| Error::InvalidInput(error.to_string()))?; + Ok(Json(server_api::SubjektivStageCandidateResponse { + staging_id: staged.id, + })) +} + +async fn scoped_record_subjektiv_session( + State(api): State, + AxumPath(path): AxumPath, + context: server_api::ServerRequestContext, + Json(request): Json, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let attribution = + subjektiv_session_attribution(&api, &path.workspace_id, &context, &request.session_id)?; + let attribution = open_subjektiv_store(&api)? + .record_session_attribution(attribution) + .map_err(|error| Error::InvalidInput(error.to_string()))?; + Ok(Json(server_api::SubjektivRecordSessionResponse { + subject_id: attribution.subject_id, + session_id: attribution.session_id, + })) +} + const MEMORY_CONSOLIDATION_PROFILE: &str = "memory-consolidation"; const MEMORY_CONSOLIDATION_SINGLETON_KEY: &str = "workspace-memory-consolidation"; @@ -33523,6 +33915,120 @@ mod tests { ); } + #[tokio::test] + async fn subjektiv_subject_worker_and_staging_derive_scope_from_singleton_authority() { + let workspace = tempfile::tempdir().unwrap(); + init_clean_git_workspace(workspace.path()); + let api = test_api(workspace.path()).await; + let subject = open_subjektiv_store(&api) + .unwrap() + .create_subject(crate::subjektiv::SubjectRole::new("companion").unwrap()) + .unwrap(); + let request = server_api::SubjektivSubjectWorkerStartRequest { + worker: CreateWorkspaceWorkerRequest { + runtime_id: EMBEDDED_WORKER_RUNTIME_ID.to_string(), + display_name: "Subject Worker".to_string(), + singleton_key: None, + profile: Some("builtin:companion".to_string()), + ticket_assignment: None, + initial_submit: Vec::new(), + workdir_attachments: Vec::new(), + control_operation_id: None, + }, + }; + let Json(created) = scoped_start_subjektiv_subject_worker( + State(api.clone()), + AxumPath(ScopedSubjektivSubjectPath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + subject_id: subject.id.clone(), + }), + HeaderMap::new(), + Json(request), + ) + .await + .unwrap(); + assert_eq!( + created.worker.singleton_key.as_deref(), + Some(format!("subjektiv:{}", subject.id).as_str()) + ); + let context = server_api::ServerRequestContext { + actor: None, + worker_source: None, + runtime_source: Some(server_api::ServerRuntimeSource { + runtime_id: created.runtime_id.clone(), + worker_id: Some(created.worker_id.clone()), + }), + origin: None, + transport_headers: Vec::new(), + }; + let operation = memory::backend::MemoryStageCandidateOperation { + source: memory::schema::SourceRef { + segment_id: "segment-1".to_string(), + range: [0, 0], + }, + extract_run_id: "extract-1".to_string(), + candidate: memory::extract::ExtractedCandidate { + kind: memory::extract::CandidateKind::Decision, + claim: "Keep subject scope host-derived".to_string(), + why_useful: "Prevents cross-subject writes".to_string(), + staleness: None, + evidence_ids: vec!["E00000000".to_string()], + }, + evidence: vec![memory::extract::StagingEvidence { + id: "E00000000".to_string(), + kind: memory::schema::EvidenceKind::new(memory::schema::EvidenceKind::MESSAGE), + entry_range: Some([0, 0]), + origin: Some(memory::schema::EvidenceOrigin { + kind: memory::schema::EvidenceOriginKind::HumanInput, + account_id: None, + workspace_id: None, + runtime_id: None, + worker_id: None, + flow_selector: None, + flow_definition_id: None, + flow_definition_revision: None, + }), + excerpt: Some("Keep subject scope host-derived".to_string()), + summary: Some("Scope decision".to_string()), + }], + source_refs: vec![memory::schema::SourceEvidenceRef { + segment_id: Some("segment-1".to_string()), + entry_range: Some([0, 0]), + evidence_id: Some("E00000000".to_string()), + evidence_kind: Some(memory::schema::EvidenceKind::new( + memory::schema::EvidenceKind::MESSAGE, + )), + ..Default::default() + }], + }; + let Json(staged) = scoped_stage_subjektiv_candidate( + State(api.clone()), + AxumPath(ScopedWorkspacePath { + workspace_id: TEST_WORKSPACE_ID.to_string(), + }), + context, + Json(server_api::SubjektivStageCandidateRequest { + session_id: "session-1".to_string(), + operation, + }), + ) + .await + .unwrap(); + let store = open_subjektiv_store(&api).unwrap(); + let candidate = store + .staging_candidate(&subject.id, &staged.staging_id) + .unwrap() + .unwrap(); + assert_eq!(candidate.subject_id, subject.id); + assert_eq!( + candidate.source_refs[0].session_id.as_deref(), + Some("session-1") + ); + let attribution = store.session_attribution("session-1").unwrap().unwrap(); + assert_eq!(attribution.subject_id, subject.id); + assert_eq!(attribution.worker_id, created.worker_id); + } + #[tokio::test] async fn singleton_retry_disconnect_replacement_and_restore_are_generation_fenced() { let workspace = tempfile::tempdir().unwrap(); @@ -48089,6 +48595,21 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3); serde_json::from_slice(&bytes).unwrap() } + #[test] + fn subjektiv_singleton_scope_is_explicit_and_opaque() { + assert_eq!( + subjektiv_subject_id_from_singleton_key("subjektiv:subject-42").unwrap(), + "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_singleton_key("subject-42").unwrap(), + "subjektiv:subject-42" + ); + } + fn write_ticket( database_path: &Path, workspace_id: &str, diff --git a/crates/workspace-server/src/subjektiv.rs b/crates/workspace-server/src/subjektiv.rs index 6c95c3188..639b12a66 100644 --- a/crates/workspace-server/src/subjektiv.rs +++ b/crates/workspace-server/src/subjektiv.rs @@ -41,6 +41,16 @@ pub enum SubjektivError { CandidateConflict(String), #[error("staging candidate `{0}` is already resolved")] CandidateResolved(String), + #[error( + "session `{session_id}` is already attributed to subject `{existing_subject_id}` and cannot be attributed to `{requested_subject_id}`" + )] + SessionSubjectConflict { + session_id: String, + existing_subject_id: String, + requested_subject_id: String, + }, + #[error("session `{0}` already has different historical attribution")] + SessionAttributionConflict(String), #[error("memory `{0}` was not found for this subject")] MemoryNotFound(String), #[error("memory `{memory_id}` revision conflict: expected {expected}, current {actual}")] @@ -117,6 +127,40 @@ pub struct SubjectRecord { pub updated_at: String, } +/// Immutable host-recorded history that attributes one durable Session to the +/// subject and Worker that produced it. This is archival provenance only: it is +/// deliberately not a current-Worker link or execution ownership record. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SubjectSessionAttribution { + pub schema_version: u32, + pub subject_id: String, + pub runtime_id: String, + pub worker_id: String, + pub session_id: String, + pub attributed_at: String, +} + +impl SubjectSessionAttribution { + pub fn new( + subject_id: impl Into, + runtime_id: impl Into, + worker_id: impl Into, + session_id: impl Into, + ) -> Result { + let record = Self { + schema_version: SUBJEKTIV_SCHEMA_VERSION, + subject_id: subject_id.into(), + runtime_id: runtime_id.into(), + worker_id: worker_id.into(), + session_id: session_id.into(), + attributed_at: now(), + }; + validate_session_attribution(&record)?; + Ok(record) + } +} + /// The existing extraction schema with only host-owned subject scope and /// persistence time added. Model input still contains neither field. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -602,11 +646,45 @@ END; Ok(()) } -static MIGRATIONS: &[FeatureMigration] = &[FeatureMigration::new( - 1, - "create subjektiv subject memory store", - create_schema, -)]; +fn add_subject_session_attribution( + transaction: &Transaction<'_>, +) -> crate::feature_storage::Result<()> { + transaction.execute_batch( + r#" +CREATE TABLE subject_session_attributions ( + session_id TEXT PRIMARY KEY, + subject_id TEXT NOT NULL, + runtime_id TEXT NOT NULL, + worker_id TEXT NOT NULL, + record_json TEXT NOT NULL, + attributed_at TEXT NOT NULL, + FOREIGN KEY (subject_id) REFERENCES subjects(subject_id) ON DELETE RESTRICT +); + +CREATE INDEX subject_session_attributions_by_subject +ON subject_session_attributions(subject_id, attributed_at, session_id); + +CREATE TRIGGER subject_session_attributions_no_update +BEFORE UPDATE ON subject_session_attributions BEGIN + SELECT RAISE(ABORT, 'subjektiv session attribution is immutable'); +END; +CREATE TRIGGER subject_session_attributions_no_delete +BEFORE DELETE ON subject_session_attributions BEGIN + SELECT RAISE(ABORT, 'subjektiv session attribution is retained'); +END; +"#, + )?; + Ok(()) +} + +static MIGRATIONS: &[FeatureMigration] = &[ + FeatureMigration::new(1, "create subjektiv subject memory store", create_schema), + FeatureMigration::new( + 2, + "add immutable subject session attribution", + add_subject_session_attribution, + ), +]; pub const REGISTRATION: FeatureRegistration = FeatureRegistration::new(SUBJEKTIV_FEATURE_ID, MIGRATIONS); @@ -740,33 +818,89 @@ impl SubjektivStore { let raw = serde_json::to_string(&record)?; self.database.try_transaction(|transaction| { require_active_subject(transaction, &record.subject_id)?; - let existing = transaction + write_staging_candidate(transaction, self.workspace_id(), record, raw) + }) + } + + /// Atomically records immutable host attribution for the committed Session + /// and stages its extracted candidate. Exact retries return the first stored + /// records, including their original host timestamps. + pub fn stage_candidate_with_attribution( + &self, + mut record: SubjectStagingRecord, + attribution: SubjectSessionAttribution, + ) -> Result<(SubjectStagingRecord, SubjectSessionAttribution)> { + validate_session_attribution(&attribution)?; + if record.subject_id != attribution.subject_id { + return Err(SubjektivError::SubjectScopeMismatch { + subject_id: record.subject_id, + reference: format!( + "session {} is attributed to subject {}", + attribution.session_id, attribution.subject_id + ), + }); + } + attach_session_to_candidate(&mut record, &attribution.session_id)?; + validate_staging_record(self.workspace_id(), &record)?; + let candidate_raw = serde_json::to_string(&record)?; + let attribution_raw = serde_json::to_string(&attribution)?; + + self.database.try_transaction(|transaction| { + require_active_subject(transaction, &record.subject_id)?; + let attribution = write_session_attribution(transaction, attribution, attribution_raw)?; + let candidate = + write_staging_candidate(transaction, self.workspace_id(), record, candidate_raw)?; + Ok((candidate, attribution)) + }) + } + + pub fn record_session_attribution( + &self, + attribution: SubjectSessionAttribution, + ) -> Result { + validate_session_attribution(&attribution)?; + let raw = serde_json::to_string(&attribution)?; + self.database.try_transaction(|transaction| { + require_active_subject(transaction, &attribution.subject_id)?; + write_session_attribution(transaction, attribution, raw) + }) + } + + pub fn session_attribution( + &self, + session_id: &str, + ) -> Result> { + validate_label("session id", session_id)?; + self.database.try_with_connection(|connection| { + let raw = connection .query_row( - "SELECT record_json FROM staging_records - WHERE subject_id = ?1 AND candidate_id = ?2", - params![record.subject_id, record.id], + "SELECT record_json FROM subject_session_attributions + WHERE session_id = ?1", + [session_id], |row| row.get::<_, String>(0), ) .optional()?; - if let Some(existing) = existing { - if equivalent_staging_payload(&existing, &raw)? { - return parse_staging(self.workspace_id(), &existing); - } - return Err(SubjektivError::CandidateConflict(record.id.clone())); - } - transaction.execute( - "INSERT INTO staging_records ( - subject_id, candidate_id, kind, record_json, created_at - ) VALUES (?1, ?2, ?3, ?4, ?5)", - params![ - record.subject_id, - record.id, - candidate_kind_name(&record.kind), - raw, - record.created_at - ], + raw.map(|raw| parse_session_attribution(&raw)).transpose() + }) + } + + pub fn subject_session_attributions( + &self, + subject_id: &str, + ) -> Result> { + validate_label("subject id", subject_id)?; + self.database.try_with_connection(|connection| { + let mut statement = connection.prepare( + "SELECT record_json FROM subject_session_attributions + WHERE subject_id = ?1 + ORDER BY attributed_at ASC, session_id ASC", )?; - Ok(record) + let rows = statement.query_map([subject_id], |row| row.get::<_, String>(0))?; + let mut records = Vec::new(); + for row in rows { + records.push(parse_session_attribution(&row?)?); + } + Ok(records) }) } @@ -1228,6 +1362,85 @@ fn write_memory_revision( Ok(record) } +fn write_staging_candidate( + transaction: &Transaction<'_>, + workspace_id: &str, + record: SubjectStagingRecord, + raw: String, +) -> Result { + let existing = transaction + .query_row( + "SELECT record_json FROM staging_records + WHERE subject_id = ?1 AND candidate_id = ?2", + params![record.subject_id, record.id], + |row| row.get::<_, String>(0), + ) + .optional()?; + if let Some(existing) = existing { + if equivalent_staging_payload(&existing, &raw)? { + return parse_staging(workspace_id, &existing); + } + return Err(SubjektivError::CandidateConflict(record.id.clone())); + } + transaction.execute( + "INSERT INTO staging_records ( + subject_id, candidate_id, kind, record_json, created_at + ) VALUES (?1, ?2, ?3, ?4, ?5)", + params![ + record.subject_id, + record.id, + candidate_kind_name(&record.kind), + raw, + record.created_at + ], + )?; + Ok(record) +} + +fn write_session_attribution( + transaction: &Transaction<'_>, + record: SubjectSessionAttribution, + raw: String, +) -> Result { + let existing = transaction + .query_row( + "SELECT record_json FROM subject_session_attributions WHERE session_id = ?1", + [&record.session_id], + |row| row.get::<_, String>(0), + ) + .optional()?; + if let Some(existing) = existing { + let existing = parse_session_attribution(&existing)?; + if existing.subject_id != record.subject_id { + return Err(SubjektivError::SessionSubjectConflict { + session_id: record.session_id, + existing_subject_id: existing.subject_id, + requested_subject_id: record.subject_id, + }); + } + if equivalent_session_attribution(&existing, &record) { + return Ok(existing); + } + return Err(SubjektivError::SessionAttributionConflict( + record.session_id, + )); + } + transaction.execute( + "INSERT INTO subject_session_attributions ( + session_id, subject_id, runtime_id, worker_id, record_json, attributed_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![ + record.session_id, + record.subject_id, + record.runtime_id, + record.worker_id, + raw, + record.attributed_at + ], + )?; + Ok(record) +} + fn insert_resolution( transaction: &Transaction<'_>, subject_id: &str, @@ -1436,6 +1649,48 @@ fn validate_memory_refs( Ok(()) } +fn validate_session_attribution(record: &SubjectSessionAttribution) -> Result<()> { + validate_schema_version(record.schema_version, "subject Session attribution")?; + validate_label("subject id", &record.subject_id)?; + validate_label("runtime id", &record.runtime_id)?; + validate_label("worker id", &record.worker_id)?; + validate_label("session id", &record.session_id)?; + validate_label("Session attribution timestamp", &record.attributed_at)?; + chrono::DateTime::parse_from_rfc3339(&record.attributed_at).map_err(|error| { + SubjektivError::InvalidRecord(format!( + "Session attribution timestamp is not RFC 3339: {error}" + )) + })?; + Ok(()) +} + +fn attach_session_to_candidate(record: &mut SubjectStagingRecord, session_id: &str) -> Result<()> { + if record.source_refs.is_empty() { + record.source_refs.push(SourceEvidenceRef { + session_id: Some(session_id.to_string()), + segment_id: Some(record.source.segment_id.clone()), + entry_range: Some(record.source.range), + ..SourceEvidenceRef::default() + }); + return Ok(()); + } + for source in &mut record.source_refs { + match source.session_id.as_deref() { + Some(existing) if existing != session_id => { + return Err(SubjektivError::SubjectScopeMismatch { + subject_id: record.subject_id.clone(), + reference: format!( + "candidate source Session `{existing}` does not match attributed Session `{session_id}`" + ), + }); + } + Some(_) => {} + None => source.session_id = Some(session_id.to_string()), + } + } + Ok(()) +} + fn validate_staging_record(workspace_id: &str, record: &SubjectStagingRecord) -> Result<()> { if record.schema_version != STAGING_SCHEMA_VERSION { return Err(SubjektivError::InvalidRecord(format!( @@ -1450,6 +1705,18 @@ fn validate_staging_record(workspace_id: &str, record: &SubjectStagingRecord) -> validate_nonempty("candidate claim", &record.claim)?; validate_nonempty("candidate why_useful", &record.why_useful)?; validate_range("candidate source", record.source.range)?; + if matches!(record.kind, CandidateKind::Preference) + && record.evidence.iter().any(|evidence| { + !matches!( + evidence.origin.as_ref().map(|origin| &origin.kind), + Some(memory::schema::EvidenceOriginKind::HumanInput) + ) + }) + { + return Err(SubjektivError::InvalidRecord( + "preference candidates require exclusively HumanInput evidence".to_string(), + )); + } let mut evidence_ids = HashSet::new(); for evidence in &record.evidence { @@ -1552,6 +1819,17 @@ fn parse_subject(raw: &str) -> Result { Ok(record) } +fn equivalent_session_attribution( + left: &SubjectSessionAttribution, + right: &SubjectSessionAttribution, +) -> bool { + left.schema_version == right.schema_version + && left.subject_id == right.subject_id + && left.runtime_id == right.runtime_id + && left.worker_id == right.worker_id + && left.session_id == right.session_id +} + fn equivalent_staging_payload(left: &str, right: &str) -> Result { let mut left = serde_json::from_str::(left)?; let mut right = serde_json::from_str::(right)?; @@ -1569,6 +1847,12 @@ fn equivalent_staging_payload(left: &str, right: &str) -> Result { Ok(left == right) } +fn parse_session_attribution(raw: &str) -> Result { + let record: SubjectSessionAttribution = serde_json::from_str(raw)?; + validate_session_attribution(&record)?; + Ok(record) +} + fn parse_staging(workspace_id: &str, raw: &str) -> Result { let record = parse_staging_without_workspace(raw)?; validate_staging_record(workspace_id, &record)?; @@ -1736,6 +2020,12 @@ mod tests { use super::*; use crate::FeatureStorage; + static V1_MIGRATIONS: &[FeatureMigration] = &[FeatureMigration::new( + 1, + "create subjektiv subject memory store", + create_schema, + )]; + fn role() -> SubjectRole { SubjectRole::new("workspace_companion").unwrap() } @@ -1787,6 +2077,19 @@ mod tests { SubjectStagingRecord::attach_at(subject_id, record, "2026-09-28T10:00:00.000Z".into()) } + fn attribution( + subject_id: &str, + runtime_id: &str, + worker_id: &str, + session_id: &str, + attributed_at: &str, + ) -> SubjectSessionAttribution { + let mut record = + SubjectSessionAttribution::new(subject_id, runtime_id, worker_id, session_id).unwrap(); + record.attributed_at = attributed_at.to_string(); + record + } + fn draft(claim: &str, reason: &str) -> MemoryDraft { MemoryDraft::active( CandidateKind::Constraint, @@ -1851,12 +2154,217 @@ mod tests { assert!(serde_json::from_str::(&raw.replace("active", "unknown")).is_err()); } + #[test] + fn v1_store_migrates_before_attributed_staging() { + let temp = tempfile::tempdir().unwrap(); + let root = temp.path().join("storage"); + let subject_id = { + let manager = FeatureStorage::new(&root); + let workspace = manager.workspace("workspace-a").unwrap(); + let registration = workspace + .register(FeatureRegistration::new( + SUBJEKTIV_FEATURE_ID, + V1_MIGRATIONS, + )) + .unwrap(); + let store = SubjektivStore::open(&workspace, ®istration).unwrap(); + store.create_subject(role()).unwrap().id + }; + + let (_manager, _workspace, store) = open_store(&root, "workspace-a"); + let (staged, stored_attribution) = store + .stage_candidate_with_attribution( + candidate(&subject_id, "candidate-after-migration", "workspace-a"), + attribution( + &subject_id, + "runtime-1", + "worker-1", + "session-1", + "2026-09-28T10:00:00.000Z", + ), + ) + .unwrap(); + + assert_eq!(staged.subject_id, subject_id); + assert_eq!(stored_attribution.session_id, "session-1"); + assert_eq!( + store.session_attribution("session-1").unwrap().unwrap(), + stored_attribution + ); + } + + #[test] + fn attributed_staging_is_atomic_and_retry_idempotent() { + 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 first_attribution = attribution( + &subject.id, + "runtime-1", + "worker-1", + "session-1", + "2026-09-28T10:00:00.000Z", + ); + let (first_candidate, first_attribution) = store + .stage_candidate_with_attribution( + candidate(&subject.id, "candidate-1", "workspace-a"), + first_attribution, + ) + .unwrap(); + + let mut retried_candidate = first_candidate.clone(); + retried_candidate.created_at = "2026-09-28T11:00:00.000Z".into(); + let retried_attribution = attribution( + &subject.id, + "runtime-1", + "worker-1", + "session-1", + "2026-09-28T11:00:00.000Z", + ); + let (retried_candidate, retried_attribution) = store + .stage_candidate_with_attribution(retried_candidate, retried_attribution) + .unwrap(); + assert_eq!(retried_candidate.created_at, first_candidate.created_at); + assert_eq!(retried_attribution, first_attribution); + + let mut conflicting_candidate = candidate(&subject.id, "candidate-1", "workspace-a"); + conflicting_candidate.claim = "different content".into(); + conflicting_candidate.source_refs[0].session_id = Some("session-2".into()); + let result = store.stage_candidate_with_attribution( + conflicting_candidate, + attribution( + &subject.id, + "runtime-1", + "worker-1", + "session-2", + "2026-09-28T12:00:00.000Z", + ), + ); + assert!(matches!( + result, + Err(SubjektivError::CandidateConflict(candidate_id)) + if candidate_id == "candidate-1" + )); + assert!(store.session_attribution("session-2").unwrap().is_none()); + } + + #[test] + fn worker_replacement_preserves_subject_history_and_rejects_cross_subject_session() { + 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_subject = store.create_subject(role()).unwrap(); + + store + .stage_candidate_with_attribution( + candidate(&subject.id, "candidate-1", "workspace-a"), + attribution( + &subject.id, + "runtime-1", + "worker-1", + "session-1", + "2026-09-28T10:00:00.000Z", + ), + ) + .unwrap(); + let mut replacement_candidate = candidate(&subject.id, "candidate-2", "workspace-a"); + replacement_candidate.source_refs[0].session_id = None; + let (replacement_candidate, _) = store + .stage_candidate_with_attribution( + replacement_candidate, + attribution( + &subject.id, + "runtime-2", + "worker-2", + "session-2", + "2026-09-28T11:00:00.000Z", + ), + ) + .unwrap(); + assert_eq!( + replacement_candidate.source_refs[0].session_id.as_deref(), + Some("session-2") + ); + + let history = store.subject_session_attributions(&subject.id).unwrap(); + assert_eq!(history.len(), 2); + assert_eq!(history[0].worker_id, "worker-1"); + assert_eq!(history[1].worker_id, "worker-2"); + assert!(history.iter().all(|entry| entry.subject_id == subject.id)); + + let update = store + .database + .try_with_connection::<_, SubjektivError>(|connection| { + connection.execute( + "UPDATE subject_session_attributions SET worker_id = 'worker-other' + WHERE session_id = 'session-1'", + [], + )?; + Ok(()) + }); + assert!(update.is_err()); + let delete = store + .database + .try_with_connection::<_, SubjektivError>(|connection| { + connection.execute( + "DELETE FROM subject_session_attributions WHERE session_id = 'session-1'", + [], + )?; + Ok(()) + }); + assert!(delete.is_err()); + + let mut foreign_candidate = + candidate(&other_subject.id, "candidate-foreign", "workspace-a"); + foreign_candidate.source_refs[0].session_id = Some("session-1".into()); + let result = store.stage_candidate_with_attribution( + foreign_candidate, + attribution( + &other_subject.id, + "runtime-1", + "worker-1", + "session-1", + "2026-09-28T12:00:00.000Z", + ), + ); + assert!(matches!( + result, + Err(SubjektivError::SessionSubjectConflict { + session_id, + existing_subject_id, + requested_subject_id, + }) if session_id == "session-1" + && existing_subject_id == subject.id + && requested_subject_id == other_subject.id + )); + assert!( + store + .staging_candidate(&other_subject.id, "candidate-foreign") + .unwrap() + .is_none() + ); + } + #[test] fn malformed_nested_provenance_is_rejected_without_staging() { 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 mut non_human_preference = + candidate(&subject.id, "candidate-preference", "workspace-a"); + non_human_preference.kind = CandidateKind::Preference; + non_human_preference.evidence[0] + .origin + .as_mut() + .unwrap() + .kind = EvidenceOriginKind::ModelOutput; + assert!(matches!( + store.stage_candidate(non_human_preference), + Err(SubjektivError::InvalidRecord(message)) + if message.contains("HumanInput") + )); + let mut reversed = candidate(&subject.id, "candidate-reversed", "workspace-a"); reversed.evidence[0].entry_range = Some([12, 10]); assert!(matches!( diff --git a/docs/design/subjektiv-store.md b/docs/design/subjektiv-store.md index 5d59a1e7c..50d9c6357 100644 --- a/docs/design/subjektiv-store.md +++ b/docs/design/subjektiv-store.md @@ -19,14 +19,17 @@ cutover of the existing Workspace Memory authority. Async callers must run these synchronous repository calls on a blocking executor. - The store contains no `current_worker_id` or equivalent current link. Yoi's - keyed singleton remains the sole authority for the current Worker. Historical - Worker/Session attribution is provenance and will be connected separately. + keyed singleton remains the sole authority for the current Worker. T-668 + connects Workers using the opaque `subjektiv:` key and records + immutable historical Worker/Session attribution as provenance; that history is + never consulted as a current Worker link. Registration is deliberately separate from open. Trusted Server construction registers `REGISTRATION` once per `FeatureStorage` manager, then reuses its `RegisteredFeature` for each Workspace. Merely adding this module does not open -or create a subjektiv database; product enablement and old-Memory cutover are -later work. +or create a subjektiv database. T-668 opens it lazily for explicit subject +operations or an authenticated subject extraction write; leaving the Feature +disabled has no subject-store side effect. ## Versioned records @@ -120,3 +123,39 @@ online backup/restore includes the entire subjektiv database, including immutabl history and provenance. Workspace deletion and shutdown use the common lifecycle fences and close retained repository handles. Disabling subjektiv does not erase its database. + +## Extraction connection (T-668) + +`feature.subjektiv` is an explicit Worker profile boundary. It reuses the +committed-run threshold, restricted Internal extraction Worker, +`SessionEntryRef` evidence resolution, explicit finish, generation fence, and +success-only pointer progression from the established Memory lifecycle. Its +pointer domain is separate from legacy Workspace Memory, and a Worker is rejected +at installation when both automatic extraction destinations are enabled. Ordinary +SubWorkers and Reviewer SubWorkers have lifecycle Features disabled by their host; +only evidence actually committed into the parent Session can enter the parent's +capture. + +The staging request contains the existing model contract (`kind`, `claim`, +`why_useful`, optional `staleness`, and entry references after host resolution) +plus the committed parent `session_id`. It contains no subject id. The Server +derives subject scope from authenticated Runtime/Worker source proof and the +current keyed-singleton lease, then attaches the session id to every source +reference and records immutable historical `{subject, runtime, worker, session}` +provenance atomically with staging. A Session already attributed to another +subject is rejected. Preference candidates still require exclusively HumanInput +evidence. + +Subject retirement does not stop a Worker, and stopping/replacing a Worker does +not retire the subject. A replacement uses the same `subjektiv:` key; +retries therefore recover through Yoi's singleton authority without a second +current-worker mapping in this database. + +Candidates are immutable writes that occur before `FinishMemoryExtraction`. If a +run stages one or more candidates and then fails, is cancelled, or loses its +generation fence, those candidates remain recorded and the extraction pointer +does not advance. A retry may therefore produce semantically duplicate candidates; +subsequent consolidation must compare them using the existing candidate +identity/content boundary. The system must not claim that the partial writes were +rolled back or silently delete them. Subject recall, consolidation, and legacy +Memory migration remain separate work.