feat: integrate standalone subjektiv support

crates/worker/src/bootstrap.rs

@@ -52,6 +52,7 @@ pub struct WorkerBootstrap<St> {
     transport: WorkerControllerTransport,
     model_client: Option<Box<dyn LlmClient>>,
     workdir_session: Option<WorkdirSessionHandle>,
+    subjektiv_host: Option<std::sync::Arc<dyn crate::subjektiv::SubjektivHost>>,
 }
 
 /// A constructed Worker whose host-owned live bindings can still be installed
@@ -106,6 +107,7 @@ where
             transport,
             model_client: None,
             workdir_session: None,
+            subjektiv_host: None,
         }
     }
 
@@ -127,6 +129,16 @@ where
         self
     }
 
+    /// Bind an explicit, process-owned subject capability. Profile policy alone
+    /// is inert; the connection is never persisted into the Worker manifest.
+    pub fn with_subjektiv_host(
+        mut self,
+        host: std::sync::Arc<dyn crate::subjektiv::SubjektivHost>,
+    ) -> Self {
+        self.subjektiv_host = Some(host);
+        self
+    }
+
     /// Construct the Worker without exposing a controller handle. Runtime hosts
     /// use this seam to bind Workdir, observation, Flow, and other live services
     /// before [`PreparedWorker::start`] performs Feature installation.
@@ -147,6 +159,24 @@ where
         if let Some(workdir_session) = self.workdir_session {
             worker.bind_single_workdir_session(Some(workdir_session));
         }
+        let explicit_subjektiv_host = self.subjektiv_host.is_some();
+        worker.bind_subjektiv_host(self.subjektiv_host);
+        if explicit_subjektiv_host
+            && worker.manifest().feature.subjektiv.profile.enabled
+            && !worker
+                .manifest()
+                .feature
+                .subjektiv
+                .profile
+                .consolidation_tools
+        {
+            worker
+                .finalize_subjektiv_session_attribution(
+                    true,
+                    crate::SubjektivSessionAttributionLifecycle::NewSession,
+                )
+                .map_err(WorkerBootstrapError::Worker)?;
+        }
         Ok(PreparedWorker::new(worker, self.layout, self.transport))
     }
 
@@ -170,6 +200,24 @@ where
         if let Some(workdir_session) = self.workdir_session {
             worker.bind_single_workdir_session(Some(workdir_session));
         }
+        let explicit_subjektiv_host = self.subjektiv_host.is_some();
+        worker.bind_subjektiv_host(self.subjektiv_host);
+        if explicit_subjektiv_host
+            && worker.manifest().feature.subjektiv.profile.enabled
+            && !worker
+                .manifest()
+                .feature
+                .subjektiv
+                .profile
+                .consolidation_tools
+        {
+            worker
+                .finalize_subjektiv_session_attribution(
+                    true,
+                    crate::SubjektivSessionAttributionLifecycle::RestoredSession,
+                )
+                .map_err(WorkerBootstrapError::Worker)?;
+        }
         Ok(PreparedWorker::new(worker, self.layout, self.transport))
     }
 

crates/worker/src/controller.rs

@@ -1377,6 +1377,7 @@ where
         None => durable_parent_notifications,
     };
     let backend_job = worker.backend_job().cloned();
+    let subjektiv_host = worker.subjektiv_host_connection()?;
     // Bounded Jobs complete their lifecycle synchronously, never by post-run background work.
     let lifecycle_enabled = worker.manifest_lifecycle_features_enabled() && backend_job.is_none();
     let prompts = worker.prompts().clone();
@@ -1443,8 +1444,8 @@ where
                 workspace_client.clone(),
             );
         if binding.subjektiv_consolidation {
-            let surface = crate::feature::builtin::memory_surface_lifecycle::SubjektivSurfaceLifecycleFeature::for_job(
-                workspace_client, spawner_manifest.clone(), worker.llm_client_handle(),
+            let surface = crate::feature::builtin::memory_surface_lifecycle::SubjektivSurfaceLifecycleFeature::for_host(
+                subjektiv_host.clone().ok_or_else(|| std::io::Error::other("Backend Job requires subjektiv Host capability"))?, spawner_manifest.clone(), worker.llm_client_handle(),
                 prompts.clone(), spawner_workspace_context.clone(),
             )?;
             result_feature = result_feature.with_surface(surface);
@@ -1453,7 +1454,9 @@ where
     }
     let memory_profile = &worker.manifest().feature.memory.profile;
     let subjektiv_profile = &worker.manifest().feature.subjektiv.profile;
-    validate_memory_lifecycle_targets(memory_profile, subjektiv_profile)?;
+    if subjektiv_host.is_some() {
+        validate_memory_lifecycle_targets(memory_profile, subjektiv_profile)?;
+    }
     let memory_install_plan = crate::feature::builtin::memory::MemoryFeatureInstallPlan::prepare(
         worker.manifest(),
         worker.workspace_client_handle(),
@@ -1476,7 +1479,7 @@ where
     let subjektiv_consolidation_plan =
         crate::feature::builtin::memory::SubjektivConsolidationFeatureInstallPlan::prepare(
             worker.manifest(),
-            worker.workspace_client_handle(),
+            subjektiv_host.clone(),
             worker.prompts().load_full(),
         )?;
     if let Some(plan) = subjektiv_consolidation_plan {
@@ -1485,22 +1488,20 @@ where
         }
         feature_registry.add_module(plan.module);
     }
-    let ordinary_subjektiv_features_enabled =
-        crate::feature::builtin::memory::ordinary_subjektiv_features_enabled(worker.manifest());
+    let ordinary_subjektiv_features_enabled = subjektiv_host.is_some()
+        && crate::feature::builtin::memory::ordinary_subjektiv_features_enabled(worker.manifest());
     if let Some(resident_summary_source) =
         crate::feature::builtin::memory::ordinary_subjektiv_resident_summary_source(
             worker.manifest(),
-            worker.workspace_client_handle(),
+            subjektiv_host.clone(),
             worker.prompts(),
         )?
     {
-        let workspace_id = worker
-            .workspace_client()
-            .workspace_id()
-            .expect("validated subject resident source has Workspace identity")
-            .to_string();
+        let workspace_id = subjektiv_host
+            .as_ref()
+            .and_then(|host| host.host.workspace_id());
         if let Some(refresh_feature) =
-            crate::feature::builtin::memory::SubjektivResidentRestoreRefreshFeature::for_host(
+            crate::feature::builtin::memory::SubjektivResidentRestoreRefreshFeature::for_connection(
                 lifecycle_enabled,
                 Arc::clone(&resident_summary_source),
                 worker.prompts(),
@@ -1516,7 +1517,7 @@ where
             crate::feature::builtin::subjektiv_memory::SubjektivMemoryFeature::from_resolved_config(
                 &worker.manifest().feature.subjektiv,
                 worker.committed_session_capture_handle(),
-                worker.workspace_client_handle(),
+                subjektiv_host.clone(),
             )?
     {
         feature_registry.add_module(subjektiv_memory);
@@ -1525,7 +1526,7 @@ where
         && let Some(subjektiv_sessions) =
             crate::feature::builtin::subjektiv_session::SubjektivSessionFeature::from_resolved_config(
                 &worker.manifest().feature.subjektiv,
-                worker.workspace_client_handle(),
+                subjektiv_host.clone(),
             )?
     {
         feature_registry.add_module(subjektiv_sessions);
@@ -1548,7 +1549,7 @@ where
     }
     if backend_job.is_none() && let Some(surface_lifecycle) = crate::feature::builtin::memory_surface_lifecycle::SubjektivSurfaceLifecycleFeature::from_manifest(
         lifecycle_enabled,
-        worker.workspace_client_handle(),
+        subjektiv_host.clone(),
         spawner_manifest.clone(),
         worker.llm_client_handle(),
         prompts.clone(),
@@ -1563,7 +1564,7 @@ where
                 worker.manifest().feature.subjektiv.clone(),
                 worker.committed_session_capture_handle(),
                 worker.session_extension_handle(),
-                worker.workspace_client_handle(),
+                subjektiv_host.clone(),
                 spawner_manifest.clone(),
                 worker.llm_client_handle(),
                 prompts.clone(),

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

@@ -4,8 +4,10 @@
 //! authority. In that case model-visible Memory tools must go through the
 //! workspace backend instead of resolving `.yoi/memory` from a Worker workdir.
 
+use crate::subjektiv::{SubjektivHostConnection, SubjektivHostError};
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
+#[cfg(test)]
 use std::time::Duration;
 
 use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
@@ -158,6 +160,7 @@ impl dyn WorkspaceClient + '_ {
     }
 }
 
+#[cfg(test)]
 const RESIDENT_SUMMARY_REQUEST_TIMEOUT: Duration = Duration::from_secs(5);
 
 async fn execute_subjektiv_memory_operation(
@@ -327,9 +330,7 @@ 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> {
+fn subjektiv_consolidation_tools(host: SubjektivHostConnection) -> Vec<ToolDefinition> {
     use SubjectConsolidationOperation as Operation;
     [
         (
@@ -371,14 +372,14 @@ pub fn workspace_http_subjektiv_consolidation_tools(
     ]
     .into_iter()
     .map(|(name, description, schema, operation)| {
-        let client = Arc::clone(&client);
+        let host = host.clone();
         Arc::new(move || {
             (
                 ToolMeta::new(name)
                     .description(description)
                     .input_schema(schema.clone()),
                 Arc::new(SubjectConsolidationTool {
-                    client: Arc::clone(&client),
+                    host: host.clone(),
                     operation,
                 }) as Arc<dyn Tool>,
             )
@@ -398,7 +399,7 @@ enum SubjectConsolidationOperation {
 }
 
 struct SubjectConsolidationTool {
-    client: Arc<dyn WorkspaceClient>,
+    host: SubjektivHostConnection,
     operation: SubjectConsolidationOperation,
 }
 
@@ -434,59 +435,17 @@ impl Tool for SubjectConsolidationTool {
                 )?)
             }
         };
-        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()))
-            }) {
+        let response = match self.host.memory(operation) {
+            Ok(response) => response,
+            Err(SubjektivHostError::Conflict { code, message }) => {
                 return Ok(ToolOutput {
                     summary: format!("Candidate decision requires reread: {code}."),
-                    content: Some(
-                        serde_json::json!({
-                            "status": "error",
-                            "error": { "code": code, "message": message }
-                        })
-                        .to_string(),
-                    ),
+                    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()))?;
+            Err(error) => return Err(error.into()),
+        };
         let summary = match &response {
             server_api::SubjektivMemoryBackendResponse::Query(value) => {
                 format!("Found {} subject Memory item(s).", value.items.len())
@@ -686,7 +645,7 @@ struct SubjectBehaviorRefreshState {
 }
 
 struct WorkspaceSubjektivResidentSummarySource {
-    client: Arc<dyn WorkspaceClient>,
+    host: SubjektivHostConnection,
     prompts: Arc<ArcSwap<PromptCatalog>>,
     behavior_state: Mutex<SubjectBehaviorRefreshState>,
 }
@@ -730,16 +689,12 @@ impl SystemPromptContributionSource for WorkspaceResidentSummarySource {
 impl WorkspaceSubjektivResidentSummarySource {
     async fn fetch(
         &self,
-    ) -> Result<server_api::SubjektivResidentContextOutput, WorkspaceMemoryBackendError> {
-        match execute_subjektiv_memory_operation_with(
-            self.client.as_ref(),
+    ) -> Result<server_api::SubjektivResidentContextOutput, SubjektivHostError> {
+        match self.host.memory(
             server_api::SubjektivMemoryBackendOperation::ResidentContext(Default::default()),
-            |client, request| {
-                client.execute_with_timeout(request, RESIDENT_SUMMARY_REQUEST_TIMEOUT)
-            },
         )? {
             server_api::SubjektivMemoryBackendResponse::ResidentContext(output) => Ok(output),
-            other => Err(WorkspaceMemoryBackendError::InvalidResponse(format!(
+            other => Err(SubjektivHostError::Unavailable(format!(
                 "unexpected subject resident context response: {other:?}"
             ))),
         }
@@ -862,10 +817,11 @@ const SUBJEKTIV_BEHAVIOR_REWRITE_HOOK: &str = "invalidate-subject-behavior-befor
 pub(crate) struct SubjektivResidentRestoreRefreshFeature {
     source: Arc<dyn SystemPromptContributionSource>,
     prompts: Arc<ArcSwap<PromptCatalog>>,
-    workspace_id: String,
+    workspace_id: Option<String>,
 }
 
 impl SubjektivResidentRestoreRefreshFeature {
+    #[cfg(test)]
     pub(crate) fn for_host(
         lifecycle_enabled: bool,
         source: Arc<dyn SystemPromptContributionSource>,
@@ -875,6 +831,20 @@ impl SubjektivResidentRestoreRefreshFeature {
         lifecycle_enabled.then(|| Self::new(source, prompts, workspace_id))
     }
 
+    pub(crate) fn for_connection(
+        lifecycle_enabled: bool,
+        source: Arc<dyn SystemPromptContributionSource>,
+        prompts: Arc<ArcSwap<PromptCatalog>>,
+        workspace_id: Option<String>,
+    ) -> Option<Self> {
+        lifecycle_enabled.then_some(Self {
+            source,
+            prompts,
+            workspace_id,
+        })
+    }
+
+    #[cfg(test)]
     pub(crate) fn new(
         source: Arc<dyn SystemPromptContributionSource>,
         prompts: Arc<ArcSwap<PromptCatalog>>,
@@ -883,7 +853,7 @@ impl SubjektivResidentRestoreRefreshFeature {
         Self {
             source,
             prompts,
-            workspace_id: workspace_id.into(),
+            workspace_id: Some(workspace_id.into()),
         }
     }
 }
@@ -981,7 +951,7 @@ impl Hook<PreLlmRequest> for SubjektivBehaviorRefreshHook {
 struct SubjektivResidentRestoreRefreshHook {
     source: Arc<dyn SystemPromptContributionSource>,
     prompts: Arc<ArcSwap<PromptCatalog>>,
-    workspace_id: String,
+    workspace_id: Option<String>,
 }
 
 #[async_trait]
@@ -1000,7 +970,7 @@ impl Hook<WorkerRestored> for SubjektivResidentRestoreRefreshHook {
         context.system_items().append_resident_summary_refresh(
             body,
             session_store::PromptRenderProvenance {
-                workspace_id: Some(self.workspace_id.clone()),
+                workspace_id: self.workspace_id.clone(),
                 config_revision: projection.config_revision,
                 source_digest: projection.source_digest.clone(),
                 projection_digest: projection.catalog_digest.clone(),
@@ -1114,51 +1084,25 @@ pub(crate) fn is_builtin_subjektiv_consolidation_profile(
 }
 
 pub(crate) fn ordinary_subjektiv_features_enabled(manifest: &manifest::WorkerManifest) -> bool {
-    manifest.feature.subjektiv.execution_enabled()
+    manifest.feature.subjektiv.profile.enabled
         && !manifest.feature.subjektiv.profile.consolidation_tools
 }
 
 pub(crate) fn ordinary_subjektiv_resident_summary_source(
     manifest: &manifest::WorkerManifest,
-    client: Arc<dyn WorkspaceClient>,
+    host: Option<SubjektivHostConnection>,
     prompts: Arc<ArcSwap<PromptCatalog>>,
 ) -> std::io::Result<Option<Arc<dyn SystemPromptContributionSource>>> {
-    if !ordinary_subjektiv_features_enabled(manifest) {
+    if !ordinary_subjektiv_features_enabled(manifest) || host.is_none() {
         return Ok(None);
     }
     manifest
         .feature
         .subjektiv
         .validate_execution()
-        .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?;
-    if !client.is_available() {
-        return Err(std::io::Error::new(
-            std::io::ErrorKind::InvalidInput,
-            "subjektiv resident Memory requires Backend Workspace API authority",
-        ));
-    }
-    let workspace_id = client.workspace_id().ok_or_else(|| {
-        std::io::Error::new(
-            std::io::ErrorKind::InvalidInput,
-            "subjektiv resident Memory requires Backend Workspace API authority",
-        )
-    })?;
-    let settings = manifest
-        .feature
-        .subjektiv
-        .workspace_settings()
-        .expect("validated enabled subjektiv config has Workspace settings");
-    if settings.workspace_id != workspace_id {
-        return Err(std::io::Error::new(
-            std::io::ErrorKind::InvalidInput,
-            format!(
-                "subjektiv Workspace settings belong to {} instead of {}",
-                settings.workspace_id, workspace_id
-            ),
-        ));
-    }
+        .map_err(std::io::Error::other)?;
     Ok(Some(Arc::new(WorkspaceSubjektivResidentSummarySource {
-        client,
+        host: host.expect("checked explicit connection"),
         prompts,
         behavior_state: Mutex::new(SubjectBehaviorRefreshState::default()),
     })))
@@ -1172,57 +1116,51 @@ pub(crate) struct SubjektivConsolidationFeatureInstallPlan {
 impl SubjektivConsolidationFeatureInstallPlan {
     pub(crate) fn prepare(
         manifest: &manifest::WorkerManifest,
-        client: Arc<dyn WorkspaceClient>,
+        host: Option<SubjektivHostConnection>,
         prompts: Arc<crate::prompt::catalog::PromptCatalog>,
     ) -> std::io::Result<Option<Self>> {
-        if !manifest.feature.subjektiv.profile.consolidation_tools {
+        let config = &manifest.feature.subjektiv;
+        if !config.profile.enabled || !config.profile.consolidation_tools || host.is_none() {
             return Ok(None);
         }
-        let config = &manifest.feature.subjektiv;
-        config
-            .validate_execution()
-            .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?;
-        if !config.execution_enabled() || config.profile.extraction.enabled {
+        config.validate_execution().map_err(std::io::Error::other)?;
+        let host = host.expect("checked explicit connection");
+        if config.profile.extraction.enabled || !host.host.consolidation_granted() {
             return Err(std::io::Error::new(
                 std::io::ErrorKind::InvalidInput,
-                "subjektiv consolidation profile requires trusted subject settings without automatic extraction",
-            ));
-        }
-        let workspace_id = client.workspace_id().ok_or_else(|| {
-            std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "subjektiv consolidation tools require Backend Workspace API authority",
-            )
-        })?;
-        let settings = config
-            .workspace_settings()
-            .expect("validated enabled subjektiv config has Workspace settings");
-        if settings.workspace_id != workspace_id {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                format!(
-                    "subjektiv settings belong to {} instead of {}",
-                    settings.workspace_id, workspace_id
-                ),
+                "subjektiv consolidation requires explicit Host grant without automatic extraction",
             ));
         }
         let system_prompt_override = prompts
-            .subjektiv_memory_consolidation_system(&settings.language)
+            .subjektiv_memory_consolidation_system(&host.host.settings().language)
             .map_err(|error| std::io::Error::other(error.to_string()))?;
         Ok(Some(Self {
-            module: SubjektivConsolidationToolsFeature {
-                tools: workspace_http_subjektiv_consolidation_tools(client),
-            },
+            module: SubjektivConsolidationToolsFeature::new(host)?,
             system_prompt_override,
         }))
     }
 }
 
 #[derive(Clone)]
-pub(crate) struct SubjektivConsolidationToolsFeature {
+pub struct SubjektivConsolidationToolsFeature {
     tools: Vec<ToolDefinition>,
 }
 
+impl SubjektivConsolidationToolsFeature {
+    /// Reusable installation for a generic Job with an explicit attenuated Host.
+    pub fn new(host: SubjektivHostConnection) -> std::io::Result<Self> {
+        if !host.host.consolidation_granted() {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::PermissionDenied,
+                "Host did not grant consolidation",
+            ));
+        }
+        Ok(Self {
+            tools: subjektiv_consolidation_tools(host),
+        })
+    }
+}
+
 impl FeatureModule for SubjektivConsolidationToolsFeature {
     fn descriptor(&self) -> FeatureDescriptor {
         let mut descriptor = FeatureDescriptor::builtin(
@@ -1293,6 +1231,29 @@ mod tests {
     use agen::tool::ToolDefinition;
     use std::sync::Mutex;
 
+    fn backend_resident_source(
+        manifest: &manifest::WorkerManifest,
+        client: Arc<dyn WorkspaceClient>,
+        prompts: Arc<ArcSwap<PromptCatalog>>,
+    ) -> std::io::Result<Option<Arc<dyn SystemPromptContributionSource>>> {
+        let host = crate::subjektiv::BackendSubjektivHost::from_resolved_config(
+            &manifest.feature.subjektiv,
+            client,
+        )?;
+        let host = host
+            .map(|host| {
+                SubjektivHostConnection::new(
+                    host,
+                    crate::subjektiv::SubjektivHostContext {
+                        worker_id: manifest.worker.name.clone(),
+                        session_id: "session-1".into(),
+                    },
+                )
+            })
+            .transpose()?;
+        ordinary_subjektiv_resident_summary_source(manifest, host, prompts)
+    }
+
     fn test_prompts() -> Arc<ArcSwap<PromptCatalog>> {
         Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap()))
     }
@@ -1601,7 +1562,7 @@ mod tests {
 
         let plan = SubjektivConsolidationFeatureInstallPlan::prepare(
             &manifest,
-            test_client(),
+            Some(crate::subjektiv::test_connection(test_client())),
             crate::prompt::catalog::PromptCatalog::builtins_only().unwrap(),
         )
         .unwrap()
@@ -1696,13 +1657,9 @@ permission = "write"
             memory::backend::MemoryResidentSummaryAvailability::Ready,
             Some("current subject surface"),
         ));
-        let ready = ordinary_subjektiv_resident_summary_source(
-            &manifest,
-            ready_client.clone(),
-            test_prompts(),
-        )
-        .unwrap()
-        .unwrap();
+        let ready = backend_resident_source(&manifest, ready_client.clone(), test_prompts())
+            .unwrap()
+            .unwrap();
         let loaded = ready.load().await;
         let SystemPromptContribution::Ready(loaded) = loaded else {
             panic!("subject resident context must be present");
@@ -1722,13 +1679,9 @@ permission = "write"
             memory::backend::MemoryResidentSummaryAvailability::Ready,
             None,
         ));
-        let ready_empty = ordinary_subjektiv_resident_summary_source(
-            &manifest,
-            ready_empty_client,
-            test_prompts(),
-        )
-        .unwrap()
-        .unwrap();
+        let ready_empty = backend_resident_source(&manifest, ready_empty_client, test_prompts())
+            .unwrap()
+            .unwrap();
         let SystemPromptContribution::Ready(ready_empty) = ready_empty.load().await else {
             panic!("empty surface must retain Subject behavior");
         };
@@ -1740,7 +1693,7 @@ permission = "write"
             memory::backend::MemoryResidentSummaryAvailability::Stale,
             memory::backend::MemoryResidentSummaryAvailability::Failed,
         ] {
-            let source = ordinary_subjektiv_resident_summary_source(
+            let source = backend_resident_source(
                 &manifest,
                 Arc::new(SubjectResidentClient::new(
                     availability,
@@ -1765,10 +1718,9 @@ permission = "write"
             memory::backend::MemoryResidentSummaryAvailability::Ungenerated,
             None,
         ));
-        let source =
-            ordinary_subjektiv_resident_summary_source(&manifest, client.clone(), test_prompts())
-                .unwrap()
-                .unwrap();
+        let source = backend_resident_source(&manifest, client.clone(), test_prompts())
+            .unwrap()
+            .unwrap();
 
         let initial = source.load().await;
         assert!(matches!(initial, SystemPromptContribution::Ready(_)));
@@ -1838,10 +1790,9 @@ permission = "write"
     async fn subject_resident_source_fails_closed_without_trusted_subject_scope() {
         let manifest = subject_manifest("workspace");
         let client = Arc::new(SubjectResidentClient::denied());
-        let source =
-            ordinary_subjektiv_resident_summary_source(&manifest, client.clone(), test_prompts())
-                .unwrap()
-                .unwrap();
+        let source = backend_resident_source(&manifest, client.clone(), test_prompts())
+            .unwrap()
+            .unwrap();
 
         let SystemPromptContribution::Ready(failure) = source.load().await else {
             panic!("fetch failure must be explicit in resident context");
@@ -1874,7 +1825,7 @@ permission = "write"
         )
         .unwrap();
         assert!(
-            ordinary_subjektiv_resident_summary_source(&disabled, test_client(), test_prompts())
+            backend_resident_source(&disabled, test_client(), test_prompts())
                 .unwrap()
                 .is_none()
         );
@@ -1886,30 +1837,22 @@ permission = "write"
             Some("policy-only surface must not load"),
         ));
         assert!(
-            ordinary_subjektiv_resident_summary_source(
-                &policy_only,
-                unattached_client.clone(),
-                test_prompts()
-            )
-            .unwrap()
-            .is_none()
+            backend_resident_source(&policy_only, unattached_client.clone(), test_prompts())
+                .unwrap()
+                .is_none()
         );
         assert!(unattached_client.paths.lock().unwrap().is_empty());
 
         let foreign = subject_manifest("other-workspace");
-        assert!(
-            ordinary_subjektiv_resident_summary_source(&foreign, test_client(), test_prompts())
-                .is_err()
-        );
+        assert!(backend_resident_source(&foreign, test_client(), test_prompts()).is_err());
     }
 
     #[test]
     fn subject_restore_refresh_is_suppressed_for_internal_hosts() {
         let manifest = subject_manifest("workspace");
-        let source =
-            ordinary_subjektiv_resident_summary_source(&manifest, test_client(), test_prompts())
-                .unwrap()
-                .unwrap();
+        let source = backend_resident_source(&manifest, test_client(), test_prompts())
+            .unwrap()
+            .unwrap();
         let prompts = Arc::new(ArcSwap::from(
             crate::prompt::catalog::PromptCatalog::builtins_only().unwrap(),
         ));
@@ -1923,10 +1866,9 @@ permission = "write"
     #[test]
     fn subject_restore_refresh_feature_declares_only_restore_lifecycle() {
         let manifest = subject_manifest("workspace");
-        let source =
-            ordinary_subjektiv_resident_summary_source(&manifest, test_client(), test_prompts())
-                .unwrap()
-                .unwrap();
+        let source = backend_resident_source(&manifest, test_client(), test_prompts())
+            .unwrap()
+            .unwrap();
         let feature = SubjektivResidentRestoreRefreshFeature::new(
             source,
             Arc::new(ArcSwap::from(
@@ -2034,7 +1976,7 @@ permission = "write"
             (403, "subject_scope_mismatch"),
         ] {
             let tool = SubjectConsolidationTool {
-                client: repository_error_client(status, code),
+                host: crate::subjektiv::test_connection(repository_error_client(status, code)),
                 operation: SubjectConsolidationOperation::DecideCandidate,
             };
             let output = tool
@@ -2051,7 +1993,7 @@ permission = "write"
 
     #[test]
     fn subjektiv_consolidation_has_atomic_decision_without_host_scope_fields() {
-        let tools = workspace_http_subjektiv_consolidation_tools(test_client());
+        let tools = subjektiv_consolidation_tools(crate::subjektiv::test_connection(test_client()));
         let names = tool_names(tools.clone());
         assert_eq!(
             names,

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

@@ -101,6 +101,7 @@ struct MemoryLifecycleTask {
     capture: CommittedSessionCaptureHandle,
     extensions: SessionExtensionHandle,
     workspace_client: Arc<dyn WorkspaceClient>,
+    subjektiv_host: Option<crate::subjektiv::SubjektivHostConnection>,
     manifest: WorkerManifest,
     client: Box<dyn LlmClient>,
     prompts: Arc<ArcSwap<PromptCatalog>>,
@@ -175,6 +176,7 @@ impl MemoryLifecycleFeature {
                 capture,
                 extensions,
                 workspace_client,
+                subjektiv_host: None,
                 manifest,
                 client,
                 prompts,
@@ -192,32 +194,34 @@ impl SubjektivLifecycleFeature {
         config: manifest::ResolvedSubjektivFeatureConfig,
         capture: CommittedSessionCaptureHandle,
         extensions: SessionExtensionHandle,
-        workspace_client: Arc<dyn WorkspaceClient>,
+        host: Option<crate::subjektiv::SubjektivHostConnection>,
         manifest: WorkerManifest,
         client: Box<dyn LlmClient>,
         prompts: Arc<ArcSwap<PromptCatalog>>,
         workspace_context: WorkerWorkspaceContext,
         event_tx: Option<broadcast::Sender<Event>>,
     ) -> std::io::Result<Option<Self>> {
-        if !lifecycle_enabled || !config.execution_enabled() || !config.profile.extraction.enabled {
+        if !lifecycle_enabled
+            || !config.profile.enabled
+            || host.is_none()
+            || !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",
-            ));
-        }
         Ok(Some(Self {
             task: MemoryLifecycleTask {
                 config: LifecycleConfig::from_subjektiv(config),
                 target: ExtractionTarget::Subjektiv,
                 capture,
                 extensions,
-                workspace_client,
+                workspace_client: crate::unavailable_workspace_client(
+                    None,
+                    "subjektiv Host has no legacy Memory authority",
+                ),
+                subjektiv_host: host,
                 manifest,
                 client,
                 prompts,
@@ -477,10 +481,11 @@ impl MemoryLifecycleTask {
             ),
             ExtractionTarget::Subjektiv => MemoryStagingOutputState::new_subjektiv(
                 view.clone(),
-                Arc::clone(&self.workspace_client),
+                self.subjektiv_host
+                    .clone()
+                    .expect("subjektiv task has Host connection"),
                 source,
                 audit.run_id.to_string(),
-                capture.session_id.clone(),
             ),
         };
         let client = if let Some(model) = self.config.extraction.model.as_ref() {
@@ -499,6 +504,11 @@ impl MemoryLifecycleTask {
             .config
             .workspace_settings()
             .map(|snapshot| snapshot.language)
+            .or_else(|| {
+                self.subjektiv_host
+                    .as_ref()
+                    .map(|host| host.host.settings().language)
+            })
         else {
             self.record_preparation_failure(
                 &audit,
@@ -737,14 +747,25 @@ impl MemoryLifecycleTask {
         .with_workspace_settings(self.config.workspace_settings.as_ref());
         let operation = memory::backend::MemoryConsolidateStagingOperation { force: false };
         let request = if subjektiv {
-            self.workspace_client
-                .request_subjektiv_memory_staging_consolidation(operation)
-                .await
+            let host = self
+                .subjektiv_host
+                .as_ref()
+                .expect("subjektiv task has Host connection");
+            host.host
+                .request_consolidation(&host.context, operation)
+                .map_err(|error| {
+                    super::memory::WorkspaceMemoryBackendError::Backend(error.to_string())
+                })
         } else {
             self.workspace_client
                 .request_memory_staging_consolidation(operation)
                 .await
         };
+        let audit = if subjektiv {
+            audit.without_legacy_backend()
+        } else {
+            audit
+        };
         match request {
             Ok(output) => {
                 tracing::debug!(
@@ -1319,6 +1340,7 @@ permission = "write"
             capture: capture_handle,
             extensions,
             workspace_client,
+            subjektiv_host: None,
             manifest: test_manifest(),
             client,
             prompts: Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())),
@@ -1379,7 +1401,7 @@ permission = "write"
             manifest::ResolvedSubjektivFeatureConfig::default(),
             capture,
             extensions,
-            crate::worker::marker_workspace_client(None, "disabled"),
+            None,
             test_manifest(),
             Box::new(ScriptClient::new(Vec::new())),
             Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())),
@@ -1407,7 +1429,7 @@ permission = "write"
             config,
             capture,
             extensions,
-            crate::worker::marker_workspace_client(None, "extraction-disabled"),
+            None,
             test_manifest(),
             Box::new(ScriptClient::new(Vec::new())),
             Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())),
@@ -1444,7 +1466,7 @@ permission = "write"
             config,
             capture,
             extensions,
-            crate::worker::marker_workspace_client(None, "child-disabled"),
+            None,
             test_manifest(),
             Box::new(ScriptClient::new(Vec::new())),
             Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())),
@@ -1593,6 +1615,9 @@ permission = "write"
             workspace_client.clone(),
         );
         task.target = ExtractionTarget::Subjektiv;
+        task.subjektiv_host = Some(crate::subjektiv::test_connection(
+            task.workspace_client.clone(),
+        ));
         task.config.consolidation_request_enabled = false;
         run_background_task(task).await;
 
@@ -1647,6 +1672,9 @@ permission = "write"
             workspace_client.clone(),
         );
         task.target = ExtractionTarget::Subjektiv;
+        task.subjektiv_host = Some(crate::subjektiv::test_connection(
+            task.workspace_client.clone(),
+        ));
         task.config.consolidation_request_enabled = false;
         run_background_task(task).await;
 
@@ -1689,6 +1717,9 @@ permission = "write"
             workspace_client.clone(),
         );
         task.target = ExtractionTarget::Subjektiv;
+        task.subjektiv_host = Some(crate::subjektiv::test_connection(
+            task.workspace_client.clone(),
+        ));
         task.config.consolidation_request_enabled = false;
         run_background_task(task).await;
 
@@ -1806,6 +1837,9 @@ permission = "write"
             workspace_client.clone(),
         );
         task.target = ExtractionTarget::Subjektiv;
+        task.subjektiv_host = Some(crate::subjektiv::test_connection(
+            task.workspace_client.clone(),
+        ));
         task.config.consolidation_request_enabled = false;
         let registry = start_background_task(task);
         tokio::time::timeout(Duration::from_secs(5), async {

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

@@ -41,7 +41,9 @@ pub(crate) struct MemoryStagingOutputState {
 #[derive(Clone)]
 enum MemoryStagingDestination {
     WorkspaceMemory,
-    Subjektiv { session_id: String },
+    Subjektiv {
+        host: crate::subjektiv::SubjektivHostConnection,
+    },
 }
 
 impl MemoryStagingOutputState {
@@ -64,17 +66,19 @@ impl MemoryStagingOutputState {
 
     pub(crate) fn new_subjektiv(
         view: SessionCapture,
-        workspace_client: Arc<dyn WorkspaceClient>,
+        host: crate::subjektiv::SubjektivHostConnection,
         source: SourceRef,
         extract_run_id: String,
-        session_id: String,
     ) -> Self {
         Self {
             view: Arc::new(view),
-            workspace_client,
+            workspace_client: crate::unavailable_workspace_client(
+                None,
+                "subjektiv Host has no legacy Memory authority",
+            ),
             source,
             extract_run_id,
-            destination: MemoryStagingDestination::Subjektiv { session_id },
+            destination: MemoryStagingDestination::Subjektiv { host },
             staged: Arc::new(Mutex::new(Vec::new())),
             finished: Arc::new(Mutex::new(None)),
         }
@@ -263,35 +267,11 @@ impl Tool for StageMemoryCandidateTool {
                     }
                 }
             }
-            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}"
-                        ))
-                    })?;
+            MemoryStagingDestination::Subjektiv { host } => {
+                let output = host
+                    .host
+                    .stage_candidate(&host.context, operation)
+                    .map_err(ToolError::from)?;
                 vec![output.staging_id]
             }
         };
@@ -607,13 +587,16 @@ mod tests {
         let client = Arc::new(RecordingSubjektivClient::default());
         let state = MemoryStagingOutputState::new_subjektiv(
             SessionCapture::new("segment-1", vec![Item::user_message("durable decision")]),
-            client.clone(),
+            {
+                let mut host = crate::subjektiv::test_connection(client.clone());
+                host.context.session_id = "session-committed".into();
+                host
+            },
             SourceRef {
                 segment_id: "segment-1".to_string(),
                 range: [0, 0],
             },
             "run-1".to_string(),
-            "session-committed".to_string(),
         );
         let tool = StageMemoryCandidateTool {
             state: state.clone(),

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

@@ -14,9 +14,7 @@ use crate::feature::background::{
     BackgroundTaskCancellation, BackgroundTaskContext, BackgroundTaskSpec, BackgroundTaskTrigger,
     FeatureBackgroundTask,
 };
-use crate::feature::builtin::memory::{
-    WorkspaceMemoryBackendError, is_builtin_subjektiv_consolidation_profile,
-};
+use crate::feature::builtin::memory::is_builtin_subjektiv_consolidation_profile;
 use crate::feature::builtin::memory_surface_output::{
     MemorySurfaceOutputFeature, MemorySurfaceOutputState, submit_llm_tool_definition,
 };
@@ -29,7 +27,10 @@ use crate::internal_worker::{
     InternalWorkerAuthority, InternalWorkerIdentity, InternalWorkerSpec,
     run_internal_worker_with_cancel_sender,
 };
-use crate::worker::{WorkerFilesystemAuthority, WorkerWorkspaceContext, WorkspaceClient};
+use crate::subjektiv::{SubjektivHostConnection, SubjektivHostError};
+#[cfg(test)]
+use crate::worker::WorkspaceClient;
+use crate::worker::{WorkerFilesystemAuthority, WorkerWorkspaceContext};
 use manifest::WorkerManifest;
 
 const TASK_NAME: &str = "subjektiv-memory-surface";
@@ -39,13 +40,13 @@ const EDITOR_MAX_TURNS: u32 = 3;
 const EDITOR_QUESTION: &str = "この主体が次の作業を始める際、毎回思い出しておくべきことは何か。継続的な制約、判断の前提、未解決事項、再発を避けたい教訓を、指定予算内でまとめる。";
 
 #[derive(Clone)]
-pub(crate) struct SubjektivSurfaceLifecycleFeature {
+pub struct SubjektivSurfaceLifecycleFeature {
     task: SubjektivSurfaceLifecycleTask,
 }
 
 #[derive(Clone)]
 struct SubjektivSurfaceLifecycleTask {
-    workspace_client: Arc<dyn WorkspaceClient>,
+    host: SubjektivHostConnection,
     manifest: WorkerManifest,
     client: Box<dyn LlmClient>,
     prompts: Arc<ArcSwap<PromptCatalog>>,
@@ -94,18 +95,20 @@ impl LlmClient for InputBudgetLlmClient {
 impl SubjektivSurfaceLifecycleFeature {
     pub(crate) fn from_manifest(
         lifecycle_enabled: bool,
-        workspace_client: Arc<dyn WorkspaceClient>,
+        host: Option<SubjektivHostConnection>,
         manifest: WorkerManifest,
         client: Box<dyn LlmClient>,
         prompts: Arc<ArcSwap<PromptCatalog>>,
         workspace_context: WorkerWorkspaceContext,
     ) -> std::io::Result<Option<Self>> {
-        let dedicated = is_builtin_subjektiv_consolidation_profile(&manifest);
-        if !lifecycle_enabled || !dedicated {
+        if !lifecycle_enabled
+            || !is_builtin_subjektiv_consolidation_profile(&manifest)
+            || host.is_none()
+        {
             return Ok(None);
         }
-        Self::for_job(
-            workspace_client,
+        Self::for_host(
+            host.expect("checked connection"),
             manifest,
             client,
             prompts,
@@ -114,8 +117,9 @@ impl SubjektivSurfaceLifecycleFeature {
         .map(Some)
     }
 
-    pub(crate) fn for_job(
-        workspace_client: Arc<dyn WorkspaceClient>,
+    /// Install the reusable surface Feature for an explicitly granted Host.
+    pub fn for_host(
+        host: SubjektivHostConnection,
         manifest: WorkerManifest,
         client: Box<dyn LlmClient>,
         prompts: Arc<ArcSwap<PromptCatalog>>,
@@ -125,22 +129,16 @@ impl SubjektivSurfaceLifecycleFeature {
             .feature
             .subjektiv
             .validate_execution()
-            .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?;
-        if !manifest.feature.subjektiv.execution_enabled() {
+            .map_err(std::io::Error::other)?;
+        if !manifest.feature.subjektiv.profile.enabled || !host.host.consolidation_granted() {
             return Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "subjektiv surface generation requires trusted subject attachment",
-            ));
-        }
-        if !workspace_client.is_available() || workspace_client.workspace_id().is_none() {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "subjektiv surface generation requires Backend Workspace API authority",
+                std::io::ErrorKind::PermissionDenied,
+                "surface generation requires explicit Host consolidation grant",
             ));
         }
         Ok(Self {
             task: SubjektivSurfaceLifecycleTask {
-                workspace_client,
+                host,
                 manifest,
                 client,
                 prompts,
@@ -150,7 +148,7 @@ impl SubjektivSurfaceLifecycleFeature {
     }
 
     /// The Job tool awaits this clean-context lifecycle before submitting its result.
-    pub(crate) async fn complete_for_job(
+    pub async fn complete_for_job(
         &self,
         cancellation: BackgroundTaskCancellation,
         reusable_surface: Option<&serde_json::Value>,
@@ -234,7 +232,7 @@ impl SubjektivSurfaceLifecycleTask {
                 ));
             }
             let generation = self
-                .workspace_client
+                .host
                 .prepare_subjektiv_memory_surface()
                 .await
                 .map_err(surface_hook_error)?;
@@ -299,7 +297,7 @@ impl SubjektivSurfaceLifecycleTask {
                 ));
             }
             match self
-                .workspace_client
+                .host
                 .publish_subjektiv_memory_surface(server_api::SubjektivSurfacePublishRequest {
                     generation_id: generation.generation_id.clone(),
                     points,
@@ -312,9 +310,8 @@ impl SubjektivSurfaceLifecycleTask {
                         "store_revision": output.built_from_store_revision, "snapshot_id": output.snapshot_id,
                     }));
                 }
-                Err(WorkspaceMemoryBackendError::Http { status, .. })
-                    if status == reqwest::StatusCode::CONFLICT
-                        && attempt + 1 < MAX_GENERATION_ATTEMPTS =>
+                Err(SubjektivHostError::Conflict { .. })
+                    if attempt + 1 < MAX_GENERATION_ATTEMPTS =>
                 {
                     continue;
                 }
@@ -334,18 +331,7 @@ impl SubjektivSurfaceLifecycleTask {
         generation: &server_api::SubjektivSurfacePrepareResponse,
         cancellation: BackgroundTaskCancellation,
     ) -> Result<Vec<server_api::SubjektivSurfacePoint>, SurfaceEditorFailure> {
-        let language = self
-            .manifest
-            .feature
-            .subjektiv
-            .workspace_settings()
-            .map(|settings| settings.language)
-            .ok_or_else(|| {
-                SurfaceEditorFailure::editor(HookError::new(
-                    HookErrorCategory::Internal,
-                    "surface editor requires bound subjektiv Workspace settings",
-                ))
-            })?;
+        let language = self.host.host.settings().language;
         let system_prompt = self
             .prompts
             .load_full()
@@ -424,7 +410,7 @@ impl SubjektivSurfaceLifecycleTask {
         job_confirmation_required: bool,
     ) -> Result<serde_json::Value, HookError> {
         let failure = self
-            .workspace_client
+            .host
             .fail_subjektiv_memory_surface(server_api::SubjektivSurfaceFailureRequest {
                 generation_id: generation.generation_id.clone(),
                 reason_code: reason_code.to_string(),
@@ -586,7 +572,7 @@ fn estimated_editor_request_tokens(request: &Request) -> Result<usize, HookError
     Ok(request.len().saturating_add(3) / 4)
 }
 
-fn surface_hook_error(error: WorkspaceMemoryBackendError) -> HookError {
+fn surface_hook_error(error: SubjektivHostError) -> HookError {
     HookError::new(HookErrorCategory::Internal, error.to_string())
 }
 
@@ -889,7 +875,7 @@ permission = "write"
         calls: Arc<AtomicUsize>,
     ) -> SubjektivSurfaceLifecycleTask {
         SubjektivSurfaceLifecycleTask {
-            workspace_client,
+            host: crate::subjektiv::test_connection(workspace_client),
             manifest: test_manifest(),
             client: Box::new(SurfaceEditorClient { calls }),
             prompts: Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())),

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

@@ -13,7 +13,7 @@ use memory::extract::CandidateKind;
 use schemars::JsonSchema;
 use serde::Deserialize;
 use server_api::{
-    SubjektivMemoryBackendOperation, SubjektivMemoryBackendRequest, SubjektivMemoryBackendResponse,
+    SubjektivMemoryBackendOperation, SubjektivMemoryBackendResponse,
     SubjektivMemoryListRevisionsRequest, SubjektivMemoryQueryRequest, SubjektivMemoryReadRequest,
     SubjektivMemoryReceiptStatus, SubjektivMemoryReceiptStatusRequest,
     SubjektivMemoryRevisionIntent, SubjektivMemoryRevisionProposal,
@@ -33,7 +33,11 @@ use crate::hook::{
     PreRequestContext, RunCommitted, RunCommittedContext,
 };
 use crate::session_capture::{SessionCapture, SessionEntryEvidence};
-use crate::worker::{WorkspaceClient, WorkspaceServerOperation};
+use crate::subjektiv::SubjektivHostConnection;
+#[cfg(test)]
+use crate::worker::WorkspaceClient;
+#[cfg(test)]
+use server_api::SubjektivMemoryBackendRequest;
 
 const QUERY_TOOL: &str = "SubjektivMemoryQuery";
 const READ_TOOL: &str = "SubjektivMemoryRead";
@@ -58,7 +62,7 @@ pub(crate) struct SubjektivMemoryFeature {
 
 #[derive(Clone)]
 struct SubjektivMemoryState {
-    client: Arc<dyn WorkspaceClient>,
+    host: SubjektivHostConnection,
     capture: CommittedSessionCaptureHandle,
 }
 
@@ -76,22 +80,17 @@ impl SubjektivMemoryFeature {
     pub(crate) fn from_resolved_config(
         config: &manifest::ResolvedSubjektivFeatureConfig,
         capture: CommittedSessionCaptureHandle,
-        client: Arc<dyn WorkspaceClient>,
+        host: Option<SubjektivHostConnection>,
     ) -> std::io::Result<Option<Self>> {
-        if !config.execution_enabled() {
+        if !config.profile.enabled || host.is_none() {
             return Ok(None);
         }
         config
             .validate_execution()
             .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?;
-        if !client.is_available() || client.workspace_id().is_none() {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "subjektiv Memory tools require Backend Workspace API authority",
-            ));
-        }
+        let host = host.expect("checked explicit connection");
         Ok(Some(Self {
-            state: SubjektivMemoryState { client, capture },
+            state: SubjektivMemoryState { host, capture },
         }))
     }
 }
@@ -483,48 +482,7 @@ impl SubjektivMemoryState {
         &self,
         operation: SubjektivMemoryBackendOperation,
     ) -> Result<SubjektivMemoryBackendResponse, ToolError> {
-        let response = self
-            .client
-            .execute_server_operation(WorkspaceServerOperation::SubjektivMemory(
-                SubjektivMemoryBackendRequest { operation },
-            ))
-            .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" | "stale_cursor"
-                        )
-                    })
-                    .map(|diagnostic| (diagnostic.code.as_str(), error.message.as_str()))
-            }) {
-                return Err(ToolError::StructuredConflict {
-                    code: code.to_string(),
-                    message: message.to_string(),
-                });
-            }
-            let detail = parsed
-                .map(|error| error.message)
-                .unwrap_or_else(|| response.body.clone());
-            let message = format!(
-                "subjektiv Memory backend returned HTTP {}: {detail}",
-                response.status
-            );
-            return if matches!(response.status, 400 | 404 | 409 | 422) {
-                Err(ToolError::InvalidArgument(message))
-            } else {
-                Err(ToolError::ExecutionFailed(message))
-            };
-        }
-        serde_json::from_str(&response.body).map_err(|error| {
-            ToolError::ExecutionFailed(format!("decode subjektiv response: {error}"))
-        })
+        self.host.memory(operation).map_err(Into::into)
     }
 
     fn backend_receipt_status(
@@ -628,7 +586,7 @@ struct PendingRetryHook {
 #[async_trait]
 impl Hook<PreLlmRequest> for PendingRetryHook {
     async fn call(&self, _input: &PreRequestContext) -> Result<HookPreRequestAction, HookError> {
-        replay_pending_committed_memory(&self.state)?;
+        replay_pending_committed_memory_with_policy(&self.state, true)?;
         Ok(HookPreRequestAction::Continue)
     }
 }
@@ -649,6 +607,12 @@ impl Hook<BeforeSessionRewrite> for PendingRewriteHook {
 }
 
 fn replay_pending_committed_memory(state: &SubjektivMemoryState) -> Result<(), HookError> {
+    replay_pending_committed_memory_with_policy(state, false)
+}
+fn replay_pending_committed_memory_with_policy(
+    state: &SubjektivMemoryState,
+    allow_pending: bool,
+) -> Result<(), HookError> {
     let capture = state
         .capture
         .capture()
@@ -691,7 +655,11 @@ fn replay_pending_committed_memory(state: &SubjektivMemoryState) -> Result<(), H
             vec![call.evidence],
             item.proposal,
         ) {
-            failures.push(format!("{receipt_id}: {error}"));
+            if !allow_pending
+                || !matches!(&error,ToolError::StructuredConflict {code,..} if code=="pending_commit")
+            {
+                failures.push(format!("{receipt_id}: {error}"));
+            }
         }
     }
     if failures.is_empty() {
@@ -1109,7 +1077,7 @@ mod tests {
         });
         SubjektivMemoryFeature {
             state: SubjektivMemoryState {
-                client: Arc::new(TestWorkspaceClient),
+                host: crate::subjektiv::test_connection(Arc::new(TestWorkspaceClient)),
                 capture,
             },
         }
@@ -1245,7 +1213,7 @@ mod tests {
             "\\\u{0000}".repeat(server_api::SUBJEKTIV_MEMORY_READ_MAX_TOOL_CONTENT_BYTES);
         let tool = SubjektivReadTool {
             state: SubjektivMemoryState {
-                client: Arc::new(ReadWorkspaceClient(response)),
+                host: crate::subjektiv::test_connection(Arc::new(ReadWorkspaceClient(response))),
                 capture: feature().state.capture,
             },
             operation: ReadOperation::Read,
@@ -1265,7 +1233,9 @@ mod tests {
 
         let oversized_tool = SubjektivReadTool {
             state: SubjektivMemoryState {
-                client: Arc::new(ReadWorkspaceClient(oversized_response)),
+                host: crate::subjektiv::test_connection(Arc::new(ReadWorkspaceClient(
+                    oversized_response,
+                ))),
                 capture: feature().state.capture,
             },
             operation: ReadOperation::Read,
@@ -1285,7 +1255,9 @@ mod tests {
         for code in ["revision_conflict", "stale_cursor"] {
             let tool = SubjektivReadTool {
                 state: SubjektivMemoryState {
-                    client: Arc::new(ConflictWorkspaceClient(code)),
+                    host: crate::subjektiv::test_connection(Arc::new(ConflictWorkspaceClient(
+                        code,
+                    ))),
                     capture: feature().state.capture,
                 },
                 operation: ReadOperation::Query,
@@ -1386,7 +1358,7 @@ mod tests {
         });
         PendingCommitHook {
             state: SubjektivMemoryState {
-                client: client.clone(),
+                host: crate::subjektiv::test_connection(client.clone()),
                 capture,
             },
         }
@@ -1439,7 +1411,7 @@ mod tests {
         });
         let hook = PendingCommitHook {
             state: SubjektivMemoryState {
-                client: client.clone(),
+                host: crate::subjektiv::test_connection(client.clone()),
                 capture,
             },
         };
@@ -1515,7 +1487,7 @@ mod tests {
             })
         });
         let state = SubjektivMemoryState {
-            client: client.clone(),
+            host: crate::subjektiv::test_connection(client.clone()),
             capture,
         };
         assert_eq!(

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

@@ -11,16 +11,20 @@ use async_trait::async_trait;
 use schemars::JsonSchema;
 use server_api::{
     SUBJEKTIV_SESSION_MAX_TOOL_CONTENT_BYTES, SubjektivSessionBackendOperation,
-    SubjektivSessionBackendRequest, SubjektivSessionBackendResponse, SubjektivSessionBackendResult,
-    SubjektivSessionDiagnosticCode, SubjektivSessionErrorResponse, SubjektivSessionListRequest,
-    SubjektivSessionReadRequest, SubjektivSessionSearchRequest,
+    SubjektivSessionBackendResponse, SubjektivSessionBackendResult, SubjektivSessionDiagnosticCode,
+    SubjektivSessionErrorResponse, SubjektivSessionListRequest, SubjektivSessionReadRequest,
+    SubjektivSessionSearchRequest,
 };
 
 use crate::feature::{
     FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution,
     ToolDeclaration,
 };
-use crate::worker::{WorkspaceClient, WorkspaceServerOperation};
+use crate::subjektiv::SubjektivHostConnection;
+#[cfg(test)]
+use crate::worker::WorkspaceClient;
+#[cfg(test)]
+use server_api::SubjektivSessionBackendRequest;
 
 const LIST_TOOL: &str = "SubjektivSessionList";
 const SEARCH_TOOL: &str = "SubjektivSessionSearch";
@@ -37,28 +41,23 @@ pub(crate) struct SubjektivSessionFeature {
 
 #[derive(Clone)]
 struct SubjektivSessionState {
-    client: Arc<dyn WorkspaceClient>,
+    host: SubjektivHostConnection,
 }
 
 impl SubjektivSessionFeature {
     pub(crate) fn from_resolved_config(
         config: &manifest::ResolvedSubjektivFeatureConfig,
-        client: Arc<dyn WorkspaceClient>,
+        host: Option<SubjektivHostConnection>,
     ) -> std::io::Result<Option<Self>> {
-        if !config.execution_enabled() {
+        if !config.profile.enabled || host.is_none() {
             return Ok(None);
         }
         config
             .validate_execution()
             .map_err(|message| std::io::Error::new(std::io::ErrorKind::InvalidInput, message))?;
-        if !client.is_available() || client.workspace_id().is_none() {
-            return Err(std::io::Error::new(
-                std::io::ErrorKind::InvalidInput,
-                "subjektiv Session tools require Backend Workspace API authority",
-            ));
-        }
+        let host = host.expect("checked explicit connection");
         Ok(Some(Self {
-            state: SubjektivSessionState { client },
+            state: SubjektivSessionState { host },
         }))
     }
 }
@@ -154,25 +153,7 @@ impl SubjektivSessionState {
         &self,
         operation: SubjektivSessionBackendOperation,
     ) -> Result<SubjektivSessionBackendResponse, ToolError> {
-        let response = self
-            .client
-            .execute_server_operation(WorkspaceServerOperation::SubjektivSession(
-                SubjektivSessionBackendRequest { operation },
-            ))
-            .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
-
-        if let Ok(response) =
-            serde_json::from_str::<SubjektivSessionBackendResponse>(&response.body)
-        {
-            return Ok(response);
-        }
-        let detail = serde_json::from_str::<server_api::RepositoryApiError>(&response.body)
-            .map(|error| error.message)
-            .unwrap_or(response.body);
-        Err(ToolError::ExecutionFailed(format!(
-            "subjektiv Session backend returned HTTP {}: {detail}",
-            response.status
-        )))
+        self.host.session(operation).map_err(Into::into)
     }
 }
 
@@ -293,7 +274,9 @@ mod tests {
     }
 
     fn state(client: Arc<dyn WorkspaceClient>) -> SubjektivSessionState {
-        SubjektivSessionState { client }
+        SubjektivSessionState {
+            host: crate::subjektiv::test_connection(client),
+        }
     }
 
     fn feature(client: Arc<dyn WorkspaceClient>) -> SubjektivSessionFeature {
@@ -363,7 +346,9 @@ mod tests {
         let config = manifest::ResolvedSubjektivFeatureConfig::default();
         let result = SubjektivSessionFeature::from_resolved_config(
             &config,
-            Arc::new(SessionToolClient::new(list_response())),
+            Some(crate::subjektiv::test_connection(Arc::new(
+                SessionToolClient::new(list_response()),
+            ))),
         )
         .unwrap();
         assert!(result.is_none());

crates/worker/src/job.rs

@@ -8,7 +8,7 @@
 //! cannot be preempted by dropping its tool future and must itself be bounded.
 
 mod result;
-pub use result::{JobResultFeature, JobResultSink, SUBMIT_JOB_RESULT_TOOL};
+pub use result::{JobResultFeature, JobResultFinalizer, JobResultSink, SUBMIT_JOB_RESULT_TOOL};
 
 use std::borrow::Cow;
 use std::path::Path;
@@ -72,6 +72,28 @@ pub struct PreparedInternalJob {
     manifest: WorkerManifest,
     client: Box<dyn LlmClient>,
     system_prompt: String,
+    features: FeatureRegistryBuilder,
+    finalizer: Option<Arc<dyn JobResultFinalizer>>,
+}
+
+/// Explicit Host-supplied domain modules and the Profile requirements they satisfy.
+/// Profile names never issue this grant. The Host must bind all module operations to
+/// the immutable request/attempt and fence them on cancellation, timeout and shutdown.
+#[derive(Default)]
+pub struct JobFeatureGrant {
+    pub features: FeatureRegistryBuilder,
+    pub satisfied_requirements: Vec<&'static str>,
+    pub finalizer: Option<Arc<dyn JobResultFinalizer>>,
+}
+
+pub fn prepare_job_with_grant(
+    request: &::job::JobRequest,
+    cwd: &Path,
+    client: Option<Box<dyn LlmClient>>,
+    grant: JobFeatureGrant,
+) -> Result<PreparedInternalJob, JobExecutionError> {
+    let registry = ProfileDiscovery::user_settings().discover()?;
+    prepare_job_from_registry_with_grant(request, cwd, client, &registry, grant)
 }
 
 /// Resolve exactly the requested registry Profile using the existing user settings registry.
@@ -91,6 +113,16 @@ fn prepare_job_from_registry(
     cwd: &Path,
     client: Option<Box<dyn LlmClient>>,
     registry: &manifest::ProfileRegistry,
+) -> Result<PreparedInternalJob, JobExecutionError> {
+    prepare_job_from_registry_with_grant(request, cwd, client, registry, JobFeatureGrant::default())
+}
+
+fn prepare_job_from_registry_with_grant(
+    request: &::job::JobRequest,
+    cwd: &Path,
+    client: Option<Box<dyn LlmClient>>,
+    registry: &manifest::ProfileRegistry,
+    grant: JobFeatureGrant,
 ) -> Result<PreparedInternalJob, JobExecutionError> {
     request.validate()?;
     let resolved = ProfileResolver::new()
@@ -101,7 +133,16 @@ fn prepare_job_from_registry(
             ProfileResolveOptions::with_worker_name("internal-job"),
         )?;
     let mut manifest = resolved.manifest;
-    validate_result_only_profile(&manifest)?;
+    validate_profile_with_grant(&manifest, &grant.satisfied_requirements)?;
+    if let Some(finalizer) = &grant.finalizer {
+        finalizer
+            .validate_profile(&manifest)
+            .map_err(JobExecutionError::UnsupportedProfile)?;
+    }
+    let mut tool_names = vec![SUBMIT_JOB_RESULT_TOOL.to_string()];
+    for descriptor in grant.features.descriptors() {
+        tool_names.extend(descriptor.tools.iter().map(|tool| tool.name.clone()));
+    }
     let scope = Scope::empty();
     let prompts = PromptCatalog::builtins_only()?;
     let template = SystemPromptTemplate::parse(
@@ -113,7 +154,7 @@ fn prepare_job_from_registry(
         cwd: Cow::Borrowed("(no filesystem)"),
         language: &manifest.engine.language,
         scope: &scope,
-        tool_names: vec![SUBMIT_JOB_RESULT_TOOL.into()],
+        tool_names,
         feature_instructions: &[],
         agents_md: None,
         resident_summary: None,
@@ -148,6 +189,8 @@ fn prepare_job_from_registry(
         manifest,
         client,
         system_prompt,
+        features: grant.features,
+        finalizer: grant.finalizer,
     })
 }
 
@@ -165,7 +208,11 @@ impl PreparedInternalJob {
     where
         F: FnOnce(tokio::sync::mpsc::Sender<()>),
     {
-        let feature = JobResultFeature::new(&self.request, attempt_id.clone(), submit)?;
+        let mut feature = JobResultFeature::new(&self.request, attempt_id.clone(), submit)?;
+        if let Some(finalizer) = self.finalizer {
+            feature =
+                feature.with_finalizer(finalizer, self.manifest.clone(), self.client.clone_boxed());
+        }
         let accepted = feature.accepted();
         let engine_policy = self.manifest.engine.clone();
         let spec = InternalWorkerSpec {
@@ -182,7 +229,7 @@ impl PreparedInternalJob {
             system_prompt: self.system_prompt,
             input: self.request.worker_input(&attempt_id)?,
             cache_key: None,
-            features: FeatureRegistryBuilder::new().with_module(feature.clone()),
+            features: self.features.with_module(feature.clone()),
             required_tools: &[SUBMIT_JOB_RESULT_TOOL],
             authority: InternalWorkerAuthority {
                 workspace: WorkerWorkspaceContext::no_workspace(),
@@ -218,7 +265,11 @@ impl PreparedInternalJob {
 }
 
 /// Validate requirements, not Profile names. Nothing configured as active is silently stripped.
-fn validate_result_only_profile(manifest: &WorkerManifest) -> Result<(), JobExecutionError> {
+
+fn validate_profile_with_grant(
+    manifest: &WorkerManifest,
+    satisfied: &[&str],
+) -> Result<(), JobExecutionError> {
     let manifest::FeatureConfig {
         task,
         memory,
@@ -306,7 +357,7 @@ fn validate_result_only_profile(manifest: &WorkerManifest) -> Result<(), JobExec
             manifest.session.record_event_trace,
         ),
     ] {
-        if enabled {
+        if enabled && !satisfied.contains(&name) {
             unavailable.push(name);
         }
     }

crates/worker/src/job/result.rs

@@ -23,6 +23,29 @@ pub trait JobResultSink: Send + Sync {
     fn submit(&self, submission: ::job::JobResultSubmission) -> Result<(), String>;
 }
 
+/// Consumer-owned asynchronous completion before durable result acceptance.
+/// The generic runner owns no domain algorithms. This runs in the result tool,
+/// never an orphaned background task, with the selected policy/model and a
+/// cancellation token that must propagate into any child resource cleanup.
+#[async_trait]
+pub trait JobResultFinalizer: Send + Sync {
+    fn validate_profile(&self, manifest: &manifest::WorkerManifest) -> Result<(), String>;
+    async fn finalize(
+        &self,
+        result: Value,
+        manifest: manifest::WorkerManifest,
+        client: Box<dyn agen::llm_client::LlmClient>,
+        cancellation: crate::feature::background::BackgroundTaskCancellation,
+    ) -> Result<Value, ToolError>;
+}
+
+#[derive(Clone)]
+struct FinalizerContext {
+    finalizer: Arc<dyn JobResultFinalizer>,
+    manifest: manifest::WorkerManifest,
+    client: Box<dyn agen::llm_client::LlmClient>,
+}
+
 #[derive(Deserialize, JsonSchema)]
 #[serde(deny_unknown_fields)]
 struct SubmitJobResultInput {
@@ -32,6 +55,7 @@ struct SubmitJobResultInput {
 #[derive(Default)]
 struct SubmissionState {
     sealed: Option<Value>,
+    prepared: Option<Value>,
     unconfirmed: Option<String>,
 }
 
@@ -46,6 +70,9 @@ pub struct JobResultFeature {
     state: Arc<Mutex<SubmissionState>>,
     accepted: Arc<Mutex<Option<::job::JobResultSubmission>>>,
     cancelled: Arc<Mutex<HashSet<String>>>,
+    serial: Arc<tokio::sync::Mutex<()>>,
+    cancellation: crate::feature::background::BackgroundTaskCancellation,
+    finalizer: Option<FinalizerContext>,
 }
 
 impl JobResultFeature {
@@ -65,9 +92,26 @@ impl JobResultFeature {
             state: Arc::default(),
             accepted: Arc::default(),
             cancelled: Arc::default(),
+            serial: Arc::default(),
+            cancellation: Default::default(),
+            finalizer: None,
         })
     }
 
+    pub fn with_finalizer(
+        mut self,
+        finalizer: Arc<dyn JobResultFinalizer>,
+        manifest: manifest::WorkerManifest,
+        client: Box<dyn agen::llm_client::LlmClient>,
+    ) -> Self {
+        self.finalizer = Some(FinalizerContext {
+            finalizer,
+            manifest,
+            client,
+        });
+        self
+    }
+
     pub(super) fn accepted(&self) -> Arc<Mutex<Option<::job::JobResultSubmission>>> {
         self.accepted.clone()
     }
@@ -107,6 +151,44 @@ impl Tool for JobResultFeature {
             .map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
         ::job::result_digest(&input.result, self.max_result_bytes)
             .map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
+        let _serial = self.serial.lock().await;
+        if self.cancellation.is_cancelled() {
+            return Err(ToolError::Cancelled(ToolOutput {
+                summary: "Job result submission cancelled".into(),
+                content: None,
+                attachments: vec![],
+            }));
+        }
+        let cached = {
+            let state = self.state.lock().expect("Job submission state poisoned");
+            if state
+                .sealed
+                .as_ref()
+                .is_some_and(|sealed| sealed != &input.result)
+            {
+                return Err(ToolError::InvalidArgument(
+                    "Job result is sealed; retry only the identical result".into(),
+                ));
+            }
+            state.prepared.clone()
+        };
+        let prepared = if let Some(value) = cached {
+            value
+        } else if let Some(finalizer) = &self.finalizer {
+            finalizer
+                .finalizer
+                .finalize(
+                    input.result.clone(),
+                    finalizer.manifest.clone(),
+                    finalizer.client.clone_boxed(),
+                    self.cancellation.clone(),
+                )
+                .await?
+        } else {
+            input.result.clone()
+        };
+        ::job::result_digest(&prepared, self.max_result_bytes)
+            .map_err(|e| ToolError::InvalidArgument(e.to_string()))?;
         let mut state = self.state.lock().expect("Job submission state poisoned");
         if self
             .cancelled
@@ -129,12 +211,13 @@ impl Tool for JobResultFeature {
         } else {
             // Seal before dispatch: a sink error can follow durable Host acceptance.
             state.sealed = Some(input.result.clone());
+            state.prepared = Some(prepared.clone());
         }
         let submission = ::job::JobResultSubmission {
             job_id: self.job_id.clone(),
             attempt_id: self.attempt_id.clone(),
             input_revision: self.input_revision.clone(),
-            result: input.result,
+            result: prepared,
         };
         let mut accepted = self.accepted.lock().expect("Job result state poisoned");
         if accepted.is_none() {
@@ -156,6 +239,7 @@ impl Tool for JobResultFeature {
     }
 
     async fn cancel_execution(&self, context: &ToolExecutionContext) -> Result<(), ToolError> {
+        self.cancellation.cancel();
         self.cancelled
             .lock()
             .expect("Job cancellation state poisoned")

crates/worker/src/job/tests.rs

@@ -1,14 +1,21 @@
 //! Scripted in-process model executions, not provider/process tests.
 use super::*;
+use crate::feature::background::BackgroundTaskCancellation;
+use crate::feature::{
+    FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution,
+    ToolDeclaration,
+};
 use agen::llm_client::event::{Event, ResponseStatus, StatusEvent};
 use agen::llm_client::{ClientError, Request};
 use agen::tool::{Tool, ToolExecutionContext};
+use agen::tool::{ToolError, ToolOutput};
 use async_trait::async_trait;
 use futures::Stream;
 use serde_json::{Value, json};
 use std::collections::VecDeque;
 use std::pin::Pin;
 use std::sync::Mutex;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 
 #[derive(Clone)]
 struct ScriptedClient {
@@ -592,6 +599,458 @@ fn unknown_prompt_and_model_are_not_substituted_even_with_injected_client() {
     }
 }
 
+// A domain capability is issued by the Host, never by the selected Profile.
+#[derive(Clone, Default)]
+struct GrantedTool(Arc<AtomicUsize>);
+#[async_trait]
+impl Tool for GrantedTool {
+    async fn execute(&self, _: &str, _: ToolExecutionContext) -> Result<ToolOutput, ToolError> {
+        self.0.fetch_add(1, Ordering::SeqCst);
+        Ok(ToolOutput {
+            summary: "granted".into(),
+            content: None,
+            attachments: vec![],
+        })
+    }
+}
+impl FeatureModule for GrantedTool {
+    fn descriptor(&self) -> FeatureDescriptor {
+        FeatureDescriptor::builtin("test-job-domain", "Test domain").with_tool(
+            ToolDeclaration::new("GrantedDomainTool", "Explicit Host capability"),
+        )
+    }
+    fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> {
+        let tool = self.clone();
+        context.tools().register(ToolContribution::new(
+            "GrantedDomainTool",
+            Arc::new(move || {
+                (
+                    agen::tool::ToolMeta::new("GrantedDomainTool"),
+                    Arc::new(tool.clone()) as Arc<dyn Tool>,
+                )
+            }),
+        ))
+    }
+}
+
+#[tokio::test]
+async fn domain_grant_is_explicit_not_profile_derived_and_preserves_selected_policy() {
+    let root = tempfile::tempdir().unwrap();
+    let mut profile = artifact("internal.flow_verifier_system", 100000, 1000);
+    profile["feature"] = json!({"task":{"enabled":true}});
+    let registry = registry(root.path(), profile.clone(), profile);
+    let marker = GrantedTool::default();
+    for selector in ["user:a", "user:b", "default"] {
+        assert!(matches!(
+            prepare_job_from_registry(
+                &request(selector),
+                root.path(),
+                Some(Box::new(ScriptedClient::new(vec![]))),
+                &registry,
+            ),
+            Err(JobExecutionError::UnsupportedProfile(_))
+        ));
+        let client = ScriptedClient::new(vec![
+            call("GrantedDomainTool", json!({})),
+            call(SUBMIT_JOB_RESULT_TOOL, json!({"result":true})),
+            prose(),
+        ]);
+        let job = prepare_job_from_registry_with_grant(
+            &request(selector),
+            root.path(),
+            Some(Box::new(client.clone())),
+            &registry,
+            JobFeatureGrant {
+                features: FeatureRegistryBuilder::new().with_module(marker.clone()),
+                satisfied_requirements: vec!["feature.task"],
+                finalizer: None,
+            },
+        )
+        .unwrap();
+        assert_eq!(
+            job.manifest.model.model_id.as_deref(),
+            Some("scripted-model")
+        );
+        assert_eq!(
+            job.manifest.engine.instruction,
+            "internal.flow_verifier_system"
+        );
+        assert_eq!(job.manifest.engine.max_turns.unwrap().get(), MAX_JOB_TURNS);
+        assert_eq!(job.manifest.engine.max_tokens, Some(MAX_JOB_TOKENS));
+        assert!(job.manifest.scope.allow.is_empty());
+        assert!(job.manifest.delegation_scope.allow.is_empty());
+        job.run("attempt".into(), Arc::new(RecordingSink::default()), |_| {})
+            .await
+            .unwrap();
+        for sent in client.requests.lock().unwrap().iter() {
+            let mut names = sent
+                .tools
+                .iter()
+                .map(|tool| tool.name.as_str())
+                .collect::<Vec<_>>();
+            names.sort();
+            assert_eq!(names, ["GrantedDomainTool", SUBMIT_JOB_RESULT_TOOL]);
+            assert_eq!(sent.config.max_tokens, Some(MAX_JOB_TOKENS));
+        }
+    }
+    assert_eq!(marker.0.load(Ordering::SeqCst), 3);
+}
+
+#[test]
+fn domain_grant_does_not_satisfy_unrelated_profile_requirements() {
+    let root = tempfile::tempdir().unwrap();
+    for (key, value, needle) in [
+        (
+            "feature",
+            json!({"task":{"enabled":true},"web":{"enabled":true}}),
+            "feature.web",
+        ),
+        ("scope", json!("workspace_read"), "scope"),
+        (
+            "delegation_scope",
+            json!("workspace_write"),
+            "delegation_scope",
+        ),
+    ] {
+        let mut profile = artifact("default", 512, 3);
+        profile[key] = value;
+        let registry = registry(root.path(), profile.clone(), profile);
+        let error = prepare_job_from_registry_with_grant(
+            &request("user:a"),
+            root.path(),
+            Some(Box::new(ScriptedClient::new(vec![]))),
+            &registry,
+            JobFeatureGrant {
+                features: FeatureRegistryBuilder::new().with_module(GrantedTool::default()),
+                satisfied_requirements: vec!["feature.task"],
+                finalizer: None,
+            },
+        )
+        .err()
+        .unwrap();
+        assert!(
+            matches!(error, JobExecutionError::UnsupportedProfile(_)),
+            "{error}"
+        );
+        assert!(error.to_string().contains(needle), "{error}");
+    }
+}
+
+#[derive(Default)]
+struct ControlledFinalizer {
+    calls: AtomicUsize,
+    active: Arc<AtomicUsize>,
+    manifests: Mutex<Vec<manifest::WorkerManifest>>,
+    entered: tokio::sync::Notify,
+    release: tokio::sync::Notify,
+    cancellation_seen: tokio::sync::Notify,
+    cleanup: tokio::sync::Notify,
+    cancelled: AtomicBool,
+    gated: bool,
+    // Reusing a captured request with the supplied client proves transport identity.
+    probe: Option<Arc<Mutex<Vec<Request>>>>,
+    reject_profile: bool,
+}
+struct FinalizerResource(Arc<AtomicUsize>);
+impl Drop for FinalizerResource {
+    fn drop(&mut self) {
+        self.0.fetch_sub(1, Ordering::SeqCst);
+    }
+}
+#[async_trait]
+impl JobResultFinalizer for ControlledFinalizer {
+    fn validate_profile(&self, _: &manifest::WorkerManifest) -> Result<(), String> {
+        if self.reject_profile {
+            Err("test finalizer refuses selected policy".into())
+        } else {
+            Ok(())
+        }
+    }
+    async fn finalize(
+        &self,
+        result: Value,
+        manifest: manifest::WorkerManifest,
+        client: Box<dyn LlmClient>,
+        cancellation: BackgroundTaskCancellation,
+    ) -> Result<Value, ToolError> {
+        self.calls.fetch_add(1, Ordering::SeqCst);
+        self.active.fetch_add(1, Ordering::SeqCst);
+        let _resource = FinalizerResource(self.active.clone());
+        self.manifests.lock().unwrap().push(manifest);
+        if let Some(requests) = &self.probe {
+            let request = requests.lock().unwrap()[0].clone();
+            let events = client
+                .stream(request)
+                .await
+                .map_err(|e| ToolError::ExecutionFailed(e.to_string()))?;
+            let _: Vec<_> = futures::StreamExt::collect(events).await;
+        }
+        self.entered.notify_one();
+        if self.gated {
+            tokio::select! {
+                _ = self.release.notified() => {},
+                _ = cancellation.cancelled() => {
+                    self.cancelled.store(true, Ordering::SeqCst);
+                    self.cancellation_seen.notify_one();
+                    self.cleanup.notified().await;
+                    return Err(ToolError::Cancelled(ToolOutput {
+                        summary: "finalizer cleaned up".into(), content: None, attachments: vec![],
+                    }));
+                }
+            }
+        }
+        Ok(json!({"prepared":result,"completion":self.calls.load(Ordering::SeqCst)}))
+    }
+}
+async fn signalled(notify: &tokio::sync::Notify) {
+    tokio::time::timeout(std::time::Duration::from_secs(2), notify.notified())
+        .await
+        .unwrap();
+}
+
+#[tokio::test]
+async fn result_finalization_is_awaited_with_selected_model_policy_and_no_orphan_resource() {
+    let root = tempfile::tempdir().unwrap();
+    let registry = registry(
+        root.path(),
+        artifact("default", 777, 4),
+        artifact("default", 512, 3),
+    );
+    let client = ScriptedClient::new(vec![
+        call(SUBMIT_JOB_RESULT_TOOL, json!({"result":{"draft":true}})),
+        prose(),
+        prose(),
+    ]);
+    let finalizer = Arc::new(ControlledFinalizer {
+        gated: true,
+        probe: Some(client.requests.clone()),
+        ..Default::default()
+    });
+    let job = prepare_job_from_registry_with_grant(
+        &request("user:a"),
+        root.path(),
+        Some(Box::new(client.clone())),
+        &registry,
+        JobFeatureGrant {
+            finalizer: Some(finalizer.clone()),
+            ..Default::default()
+        },
+    )
+    .unwrap();
+    let expected = serde_json::to_value(&job.manifest).unwrap();
+    let sink = Arc::new(RecordingSink::default());
+    let recording = sink.clone();
+    let (tx, rx) = tokio::sync::oneshot::channel();
+    let run = tokio::spawn(async move {
+        job.run("attempt".into(), recording, move |sender| {
+            let _ = tx.send(sender);
+        })
+        .await
+    });
+    let sender = rx.await.unwrap();
+    signalled(&finalizer.entered).await;
+    assert!(!run.is_finished());
+    assert!(
+        sink.0.lock().unwrap().is_empty(),
+        "sink cannot accept an unfinished result"
+    );
+    assert_eq!(finalizer.active.load(Ordering::SeqCst), 1);
+    assert_eq!(
+        serde_json::to_value(&finalizer.manifests.lock().unwrap()[0]).unwrap(),
+        expected
+    );
+    assert_eq!(
+        client.requests.lock().unwrap().len(),
+        2,
+        "finalizer uses the injected selected transport"
+    );
+    finalizer.release.notify_one();
+    let accepted = tokio::time::timeout(std::time::Duration::from_secs(2), run)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    assert_eq!(
+        accepted.submission.result,
+        json!({"prepared":{"draft":true},"completion":1})
+    );
+    assert_eq!(sink.0.lock().unwrap().as_slice(), &[accepted.submission]);
+    assert_eq!(finalizer.active.load(Ordering::SeqCst), 0);
+    assert_eq!(finalizer.calls.load(Ordering::SeqCst), 1);
+    assert!(sender.is_closed());
+}
+
+#[tokio::test]
+async fn cancellation_reaches_finalizer_and_runner_waits_for_cleanup_without_acceptance() {
+    let root = tempfile::tempdir().unwrap();
+    let registry = registry(
+        root.path(),
+        artifact("default", 512, 3),
+        artifact("default", 512, 3),
+    );
+    let finalizer = Arc::new(ControlledFinalizer {
+        gated: true,
+        ..Default::default()
+    });
+    let job = prepare_job_from_registry_with_grant(
+        &request("user:a"),
+        root.path(),
+        Some(Box::new(ScriptedClient::new(vec![
+            call(SUBMIT_JOB_RESULT_TOOL, json!({"result":true})),
+            prose(),
+        ]))),
+        &registry,
+        JobFeatureGrant {
+            finalizer: Some(finalizer.clone()),
+            ..Default::default()
+        },
+    )
+    .unwrap();
+    let sink = Arc::new(RecordingSink::default());
+    let recording = sink.clone();
+    let (tx, rx) = tokio::sync::oneshot::channel();
+    let run = tokio::spawn(async move {
+        job.run("attempt".into(), recording, move |sender| {
+            let _ = tx.send(sender);
+        })
+        .await
+    });
+    let sender = rx.await.unwrap();
+    signalled(&finalizer.entered).await;
+    assert!(sink.0.lock().unwrap().is_empty());
+    sender.send(()).await.unwrap();
+    signalled(&finalizer.cancellation_seen).await;
+    assert!(finalizer.cancelled.load(Ordering::SeqCst));
+    assert!(
+        !run.is_finished(),
+        "cancellation must await consumer resource cleanup"
+    );
+    assert_eq!(finalizer.active.load(Ordering::SeqCst), 1);
+    assert!(sink.0.lock().unwrap().is_empty());
+    finalizer.cleanup.notify_one();
+    let outcome = tokio::time::timeout(std::time::Duration::from_secs(2), run)
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(
+        matches!(outcome, Err(JobExecutionError::Cancelled)),
+        "{outcome:?}"
+    );
+    assert!(sink.0.lock().unwrap().is_empty());
+    assert_eq!(finalizer.active.load(Ordering::SeqCst), 0);
+    assert!(sender.is_closed());
+}
+
+#[test]
+fn finalizer_can_refuse_selected_profile_before_execution() {
+    let root = tempfile::tempdir().unwrap();
+    let registry = registry(
+        root.path(),
+        artifact("default", 512, 3),
+        artifact("default", 512, 3),
+    );
+    let finalizer = Arc::new(ControlledFinalizer {
+        reject_profile: true,
+        ..Default::default()
+    });
+    let client = ScriptedClient::new(vec![]);
+    let error = prepare_job_from_registry_with_grant(
+        &request("user:a"),
+        root.path(),
+        Some(Box::new(client.clone())),
+        &registry,
+        JobFeatureGrant {
+            finalizer: Some(finalizer.clone()),
+            ..Default::default()
+        },
+    )
+    .err()
+    .unwrap();
+    assert!(matches!(error, JobExecutionError::UnsupportedProfile(_)));
+    assert!(
+        error
+            .to_string()
+            .contains("test finalizer refuses selected policy")
+    );
+    assert!(client.requests.lock().unwrap().is_empty());
+    assert_eq!(finalizer.calls.load(Ordering::SeqCst), 0);
+}
+
+#[derive(Default)]
+struct UncertainThenAcceptedSink(Mutex<Vec<::job::JobResultSubmission>>);
+impl JobResultSink for UncertainThenAcceptedSink {
+    fn submit(&self, submission: ::job::JobResultSubmission) -> Result<(), String> {
+        let mut submissions = self.0.lock().unwrap();
+        submissions.push(submission);
+        if submissions.len() == 1 {
+            Err("durable acceptance is uncertain".into())
+        } else {
+            Ok(())
+        }
+    }
+}
+
+#[tokio::test]
+async fn exact_uncertain_sink_retry_reuses_prepared_outcome_without_rerunning_finalizer() {
+    let root = tempfile::tempdir().unwrap();
+    let job = prepared(root.path(), &request("user:a"), ScriptedClient::new(vec![]));
+    let finalizer = Arc::new(ControlledFinalizer::default());
+    let sink = Arc::new(UncertainThenAcceptedSink::default());
+    let feature = JobResultFeature::new(&request("user:a"), "attempt".into(), sink.clone())
+        .unwrap()
+        .with_finalizer(finalizer.clone(), job.manifest, job.client);
+    assert!(
+        feature
+            .execute(
+                r#"{"result":{"draft":true}}"#,
+                ToolExecutionContext::direct()
+            )
+            .await
+            .is_err()
+    );
+    assert!(feature.unconfirmed().is_some());
+    assert!(feature.accepted().lock().unwrap().is_none());
+    assert_eq!(finalizer.calls.load(Ordering::SeqCst), 1);
+    assert!(
+        feature
+            .execute(
+                r#"{"result":{"draft":false}}"#,
+                ToolExecutionContext::direct()
+            )
+            .await
+            .is_err()
+    );
+    assert_eq!(sink.0.lock().unwrap().len(), 1);
+    feature
+        .execute(
+            r#"{"result":{"draft":true}}"#,
+            ToolExecutionContext::direct(),
+        )
+        .await
+        .unwrap();
+    feature
+        .execute(
+            r#"{"result":{"draft":true}}"#,
+            ToolExecutionContext::direct(),
+        )
+        .await
+        .unwrap();
+    assert!(feature.unconfirmed().is_none());
+    let submissions = sink.0.lock().unwrap();
+    assert_eq!(submissions.len(), 2);
+    assert_eq!(submissions[0], submissions[1]);
+    assert_eq!(
+        submissions[0].result,
+        json!({"prepared":{"draft":true},"completion":1})
+    );
+    assert_eq!(
+        feature.accepted().lock().unwrap().as_ref(),
+        Some(&submissions[1])
+    );
+    assert_eq!(finalizer.calls.load(Ordering::SeqCst), 1);
+    assert_eq!(finalizer.active.load(Ordering::SeqCst), 0);
+}
+
 struct UnknownSink(Mutex<Vec<::job::JobResultSubmission>>);
 impl JobResultSink for UnknownSink {
     fn submit(&self, value: ::job::JobResultSubmission) -> Result<(), String> {

crates/worker/src/lib.rs

@@ -25,6 +25,7 @@ pub mod shared_state;
 mod shutdown_after_idle;
 pub mod skill;
 pub mod spawn;
+pub mod subjektiv;
 pub mod wip;
 
 mod internal_worker;

crates/worker/src/session_capture.rs

@@ -712,7 +712,9 @@ impl SessionCapture {
 
 fn public_provenance(origin: &WorkerHistoryProvenance) -> SessionEntryProvenance {
     match origin {
-        WorkerHistoryProvenance::HumanInput { .. } => SessionEntryProvenance::HumanInput,
+        WorkerHistoryProvenance::HumanInput { .. } | WorkerHistoryProvenance::LocalHumanInput => {
+            SessionEntryProvenance::HumanInput
+        }
         WorkerHistoryProvenance::WorkerInput { .. } => SessionEntryProvenance::WorkerInput,
         WorkerHistoryProvenance::FlowInstruction { .. } => SessionEntryProvenance::FlowInstruction,
         WorkerHistoryProvenance::BackendInstruction { .. } => {

crates/worker/src/subjektiv.rs

@@ -0,0 +1,700 @@
+//! Explicit Host capability for subject-scoped Memory and committed Sessions.
+//!
+//! A manifest is policy, not a connection. Hosts bind Subject/storage authority
+//! in their implementation; the Worker supplies its own Worker/Session context.
+//! Neither Subject identity nor filesystem/Workspace authority is model input.
+
+pub use crate::feature::builtin::memory::SubjektivConsolidationToolsFeature;
+pub use crate::feature::builtin::memory_surface_lifecycle::SubjektivSurfaceLifecycleFeature;
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use agen::tool::ToolError;
+use memory::backend::{
+    MemoryConsolidateStagingOperation, MemoryConsolidationOutput, MemoryStageCandidateOperation,
+};
+use server_api::*;
+
+use crate::{WorkspaceClient, WorkspaceRequest, WorkspaceRequestMethod, WorkspaceServerOperation};
+
+/// Host-issued execution context, never deserialized from model arguments.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SubjektivHostContext {
+    pub worker_id: String,
+    pub session_id: String,
+}
+
+/// Explicit local configuration. This is not a trusted Workspace snapshot and
+/// is not persisted in the reusable Worker manifest.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct SubjektivHostSettings {
+    pub language: String,
+}
+
+#[derive(Debug, thiserror::Error)]
+pub enum SubjektivHostError {
+    #[error("{message}")]
+    Conflict { code: String, message: String },
+    #[error("{0}")]
+    InvalidArgument(String),
+    #[error("{0}")]
+    Unavailable(String),
+}
+
+impl From<SubjektivHostError> for ToolError {
+    fn from(error: SubjektivHostError) -> Self {
+        match error {
+            SubjektivHostError::Conflict { code, message } => {
+                Self::StructuredConflict { code, message }
+            }
+            SubjektivHostError::InvalidArgument(message) => Self::InvalidArgument(message),
+            SubjektivHostError::Unavailable(message) => Self::ExecutionFailed(message),
+        }
+    }
+}
+
+/// Object-safe capability implemented by Backend and standalone Hosts.
+/// Implementations must enforce their bound Subject, Worker/Session authority,
+/// committed evidence and (for Jobs) immutable batch attenuation. Knowing an
+/// identifier is not a grant. No method grants ambient filesystem access.
+pub trait SubjektivHost: Send + Sync + 'static {
+    fn settings(&self) -> SubjektivHostSettings;
+    /// Separate explicit grant for decision/surface execution; Profile policy
+    /// alone must never authorize consolidation tools.
+    fn consolidation_granted(&self) -> bool {
+        false
+    }
+    /// Optional real Backend identity, used only for prompt provenance.
+    fn workspace_id(&self) -> Option<String> {
+        None
+    }
+    fn memory(
+        &self,
+        context: &SubjektivHostContext,
+        operation: SubjektivMemoryBackendOperation,
+    ) -> Result<SubjektivMemoryBackendResponse, SubjektivHostError>;
+    fn session(
+        &self,
+        context: &SubjektivHostContext,
+        operation: SubjektivSessionBackendOperation,
+    ) -> Result<SubjektivSessionBackendResponse, SubjektivHostError>;
+    fn stage_candidate(
+        &self,
+        context: &SubjektivHostContext,
+        operation: MemoryStageCandidateOperation,
+    ) -> Result<SubjektivStageCandidateResponse, SubjektivHostError>;
+    fn record_session(
+        &self,
+        context: &SubjektivHostContext,
+        create_if_missing: bool,
+    ) -> Result<SubjektivRecordSessionResponse, SubjektivHostError>;
+    fn request_consolidation(
+        &self,
+        context: &SubjektivHostContext,
+        operation: MemoryConsolidateStagingOperation,
+    ) -> Result<MemoryConsolidationOutput, SubjektivHostError>;
+}
+
+/// Backend adapter. Existing authenticated Workspace operations and snapshot
+/// validation remain the authority; no local Workspace is synthesized.
+pub struct BackendSubjektivHost {
+    client: Arc<dyn WorkspaceClient>,
+    settings: SubjektivHostSettings,
+    workspace_id: String,
+    consolidation_granted: bool,
+}
+
+impl BackendSubjektivHost {
+    pub fn from_resolved_config(
+        config: &manifest::ResolvedSubjektivFeatureConfig,
+        client: Arc<dyn WorkspaceClient>,
+    ) -> std::io::Result<Option<Arc<dyn SubjektivHost>>> {
+        config.validate_execution().map_err(std::io::Error::other)?;
+        if !config.execution_enabled() {
+            return Ok(None);
+        }
+        let settings = config
+            .workspace_settings()
+            .expect("execution requires trusted settings");
+        if !client.is_available() || client.workspace_id() != Some(settings.workspace_id.as_str()) {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "subjektiv requires available Backend authority matching trusted Workspace settings",
+            ));
+        }
+        Ok(Some(Arc::new(Self {
+            client,
+            workspace_id: settings.workspace_id,
+            settings: SubjektivHostSettings {
+                language: settings.language,
+            },
+            consolidation_granted: config.profile.consolidation_tools,
+        })))
+    }
+
+    fn execute<T: serde::de::DeserializeOwned>(
+        &self,
+        operation: WorkspaceServerOperation,
+    ) -> Result<T, SubjektivHostError> {
+        let response = self
+            .client
+            .execute_server_operation(operation)
+            .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?;
+        decode(response)
+    }
+}
+
+fn decode<T: serde::de::DeserializeOwned>(
+    response: crate::WorkspaceResponse,
+) -> Result<T, SubjektivHostError> {
+    if !response.is_success() {
+        let parsed = serde_json::from_str::<RepositoryApiError>(&response.body).ok();
+        let message = parsed
+            .as_ref()
+            .map(|error| error.message.clone())
+            .unwrap_or(response.body);
+        if let Some(code) = parsed.as_ref().and_then(|error| {
+            error
+                .diagnostics
+                .iter()
+                .find(|diagnostic| {
+                    matches!(
+                        diagnostic.code.as_str(),
+                        "revision_conflict"
+                            | "stale_cursor"
+                            | "candidate_decision_conflict"
+                            | "subject_scope_mismatch"
+                    )
+                })
+                .map(|diagnostic| diagnostic.code.clone())
+        }) {
+            return Err(SubjektivHostError::Conflict { code, message });
+        }
+        return Err(if response.status == 409 {
+            SubjektivHostError::Conflict {
+                code: "conflict".into(),
+                message,
+            }
+        } else if matches!(response.status, 400 | 404 | 422) {
+            SubjektivHostError::InvalidArgument(message)
+        } else {
+            SubjektivHostError::Unavailable(message)
+        });
+    }
+    serde_json::from_str(&response.body).map_err(|error| {
+        SubjektivHostError::Unavailable(format!("decode subjektiv response: {error}"))
+    })
+}
+
+impl SubjektivHost for BackendSubjektivHost {
+    fn settings(&self) -> SubjektivHostSettings {
+        SubjektivHostSettings {
+            language: self.settings.language.clone(),
+        }
+    }
+    fn workspace_id(&self) -> Option<String> {
+        Some(self.workspace_id.clone())
+    }
+    fn consolidation_granted(&self) -> bool {
+        self.consolidation_granted
+    }
+    fn memory(
+        &self,
+        _context: &SubjektivHostContext,
+        operation: SubjektivMemoryBackendOperation,
+    ) -> Result<SubjektivMemoryBackendResponse, SubjektivHostError> {
+        // Preserve the bounded resident-context request timeout.
+        if matches!(
+            operation,
+            SubjektivMemoryBackendOperation::ResidentContext(_)
+        ) {
+            let request = WorkspaceRequest::json(
+                WorkspaceRequestMethod::Post,
+                format!("/api/w/{}/subjektiv/memory", self.workspace_id),
+                serde_json::to_string(&SubjektivMemoryBackendRequest { operation })
+                    .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
+            );
+            return decode(
+                self.client
+                    .execute_with_timeout(request, Duration::from_secs(5))
+                    .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
+            );
+        }
+        self.execute(WorkspaceServerOperation::SubjektivMemory(
+            SubjektivMemoryBackendRequest { operation },
+        ))
+    }
+    fn session(
+        &self,
+        _context: &SubjektivHostContext,
+        operation: SubjektivSessionBackendOperation,
+    ) -> Result<SubjektivSessionBackendResponse, SubjektivHostError> {
+        let response = self
+            .client
+            .execute_server_operation(WorkspaceServerOperation::SubjektivSession(
+                SubjektivSessionBackendRequest { operation },
+            ))
+            .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?;
+        // Session domain diagnostics are typed even on non-2xx responses.
+        if let Ok(value) = serde_json::from_str(&response.body) {
+            return Ok(value);
+        }
+        decode(response)
+    }
+    fn stage_candidate(
+        &self,
+        context: &SubjektivHostContext,
+        operation: MemoryStageCandidateOperation,
+    ) -> Result<SubjektivStageCandidateResponse, SubjektivHostError> {
+        self.execute(WorkspaceServerOperation::SubjektivStageCandidate(
+            SubjektivStageCandidateRequest {
+                session_id: context.session_id.clone(),
+                operation,
+            },
+        ))
+    }
+    fn record_session(
+        &self,
+        context: &SubjektivHostContext,
+        create_if_missing: bool,
+    ) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> {
+        self.execute(WorkspaceServerOperation::SubjektivRecordSession(
+            SubjektivRecordSessionRequest {
+                session_id: context.session_id.clone(),
+                create_if_missing,
+            },
+        ))
+    }
+    fn request_consolidation(
+        &self,
+        _context: &SubjektivHostContext,
+        operation: MemoryConsolidateStagingOperation,
+    ) -> Result<MemoryConsolidationOutput, SubjektivHostError> {
+        let response = self
+            .client
+            .execute(WorkspaceRequest::json(
+                WorkspaceRequestMethod::Post,
+                format!("/api/w/{}/subjektiv/consolidation", self.workspace_id),
+                serde_json::to_string(&MemoryConsolidateStagingRequest {
+                    force: operation.force,
+                })
+                .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?,
+            ))
+            .map_err(|error| SubjektivHostError::Unavailable(error.to_string()))?;
+        let response: MemoryConsolidationResponse = decode(response)?;
+        Ok(MemoryConsolidationOutput {
+            status: response.status,
+            summary: response.summary,
+            candidate_count: response.candidate_count,
+            total_bytes: response.total_bytes,
+        })
+    }
+}
+
+/// Live connection with Worker-supplied authority. Not serializable and never
+/// inherited by an ordinary SubWorker.
+#[derive(Clone)]
+pub struct SubjektivHostConnection {
+    pub host: Arc<dyn SubjektivHost>,
+    pub context: SubjektivHostContext,
+}
+
+impl SubjektivHostConnection {
+    pub fn new(
+        host: Arc<dyn SubjektivHost>,
+        context: SubjektivHostContext,
+    ) -> std::io::Result<Self> {
+        let language = host.settings().language;
+        if context.worker_id.trim().is_empty()
+            || context.session_id.trim().is_empty()
+            || context.worker_id.chars().any(char::is_control)
+            || context.session_id.chars().any(char::is_control)
+            || !manifest::is_normalized_workspace_memory_language(&language)
+        {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "invalid subjektiv Host context/settings",
+            ));
+        }
+        Ok(Self { host, context })
+    }
+    pub fn memory(
+        &self,
+        operation: SubjektivMemoryBackendOperation,
+    ) -> Result<SubjektivMemoryBackendResponse, SubjektivHostError> {
+        if let SubjektivMemoryBackendOperation::StageExplicit(input) = &operation
+            && input.session_id != self.context.session_id
+        {
+            return Err(SubjektivHostError::InvalidArgument(
+                "explicit staging targets a foreign Session".into(),
+            ));
+        }
+        if matches!(
+            operation,
+            SubjektivMemoryBackendOperation::DecideCandidate(_)
+                | SubjektivMemoryBackendOperation::PrepareSurface(_)
+                | SubjektivMemoryBackendOperation::PublishSurface(_)
+                | SubjektivMemoryBackendOperation::FailSurface(_)
+        ) && !self.host.consolidation_granted()
+        {
+            return Err(SubjektivHostError::InvalidArgument(
+                "Host did not grant consolidation".into(),
+            ));
+        }
+        self.host.memory(&self.context, operation)
+    }
+    pub fn session(
+        &self,
+        operation: SubjektivSessionBackendOperation,
+    ) -> Result<SubjektivSessionBackendResponse, SubjektivHostError> {
+        self.host.session(&self.context, operation)
+    }
+    pub async fn prepare_subjektiv_memory_surface(
+        &self,
+    ) -> Result<SubjektivSurfacePrepareResponse, SubjektivHostError> {
+        match self.memory(SubjektivMemoryBackendOperation::PrepareSurface(
+            SubjektivSurfacePrepareRequest {},
+        ))? {
+            SubjektivMemoryBackendResponse::SurfacePrepared(value) => Ok(value),
+            _ => Err(SubjektivHostError::Unavailable(
+                "unexpected surface preparation response".into(),
+            )),
+        }
+    }
+    pub async fn publish_subjektiv_memory_surface(
+        &self,
+        input: SubjektivSurfacePublishRequest,
+    ) -> Result<SubjektivSurfacePublishResponse, SubjektivHostError> {
+        match self.memory(SubjektivMemoryBackendOperation::PublishSurface(input))? {
+            SubjektivMemoryBackendResponse::SurfacePublished(value) => Ok(value),
+            _ => Err(SubjektivHostError::Unavailable(
+                "unexpected surface publication response".into(),
+            )),
+        }
+    }
+    pub async fn fail_subjektiv_memory_surface(
+        &self,
+        input: SubjektivSurfaceFailureRequest,
+    ) -> Result<SubjektivSurfaceFailureResponse, SubjektivHostError> {
+        match self.memory(SubjektivMemoryBackendOperation::FailSurface(input))? {
+            SubjektivMemoryBackendResponse::SurfaceFailed(value) => Ok(value),
+            _ => Err(SubjektivHostError::Unavailable(
+                "unexpected surface failure response".into(),
+            )),
+        }
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn test_connection(client: Arc<dyn WorkspaceClient>) -> SubjektivHostConnection {
+    let mut config = manifest::ResolvedSubjektivFeatureConfig::default();
+    config.profile.enabled = true;
+    config.profile.consolidation_tools = true;
+    config
+        .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot {
+            workspace_id: client.workspace_id().unwrap().to_string(),
+            settings_revision: 1,
+            language: "English".into(),
+        })
+        .unwrap();
+    SubjektivHostConnection::new(
+        BackendSubjektivHost::from_resolved_config(&config, client)
+            .unwrap()
+            .unwrap(),
+        SubjektivHostContext {
+            worker_id: "worker-test".into(),
+            session_id: "session-1".into(),
+        },
+    )
+    .unwrap()
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::{
+        Method, WorkerBootstrap, WorkerBootstrapLayout, WorkerControllerTransport,
+        WorkerFilesystemAuthority, WorkerManifest, WorkerWorkspaceContext,
+    };
+    use agen::llm_client::event::{Event as LlmEvent, ResponseStatus, StatusEvent};
+    use agen::llm_client::{ClientError, LlmClient, Request};
+    use async_trait::async_trait;
+    use futures_util::Stream;
+    use session_store::{CombinedStore, FsStore, FsWorkerStore, WorkerMetadataStore};
+    use std::pin::Pin;
+    use std::sync::Mutex;
+
+    #[derive(Default)]
+    struct LocalHost {
+        contexts: Mutex<Vec<SubjektivHostContext>>,
+        attributed: Mutex<Vec<(SubjektivHostContext, bool)>>,
+    }
+    impl SubjektivHost for LocalHost {
+        fn settings(&self) -> SubjektivHostSettings {
+            SubjektivHostSettings {
+                language: "English".into(),
+            }
+        }
+        fn memory(
+            &self,
+            context: &SubjektivHostContext,
+            operation: SubjektivMemoryBackendOperation,
+        ) -> Result<SubjektivMemoryBackendResponse, SubjektivHostError> {
+            self.contexts.lock().unwrap().push(context.clone());
+            match operation {
+                SubjektivMemoryBackendOperation::ResidentContext(_) => {
+                    Ok(SubjektivMemoryBackendResponse::ResidentContext(
+                        SubjektivResidentContextOutput {
+                            behavior_md: "Explicit local subject behavior.".into(),
+                            behavior_revision: 1,
+                            memory_surface: memory::backend::MemoryResidentSummaryOutput {
+                                availability:
+                                    memory::backend::MemoryResidentSummaryAvailability::Ungenerated,
+                                content: None,
+                            },
+                        },
+                    ))
+                }
+                _ => Err(SubjektivHostError::Unavailable(
+                    "not granted by test Host".into(),
+                )),
+            }
+        }
+        fn session(
+            &self,
+            _: &SubjektivHostContext,
+            _: SubjektivSessionBackendOperation,
+        ) -> Result<SubjektivSessionBackendResponse, SubjektivHostError> {
+            Err(SubjektivHostError::Unavailable("not used".into()))
+        }
+        fn stage_candidate(
+            &self,
+            _: &SubjektivHostContext,
+            _: MemoryStageCandidateOperation,
+        ) -> Result<SubjektivStageCandidateResponse, SubjektivHostError> {
+            Err(SubjektivHostError::Unavailable("not used".into()))
+        }
+        fn record_session(
+            &self,
+            context: &SubjektivHostContext,
+            create_if_missing: bool,
+        ) -> Result<SubjektivRecordSessionResponse, SubjektivHostError> {
+            self.attributed
+                .lock()
+                .unwrap()
+                .push((context.clone(), create_if_missing));
+            Ok(SubjektivRecordSessionResponse {
+                subject_id: "test-owned-local-subject".into(),
+                session_id: context.session_id.clone(),
+            })
+        }
+        fn request_consolidation(
+            &self,
+            _: &SubjektivHostContext,
+            _: MemoryConsolidateStagingOperation,
+        ) -> Result<MemoryConsolidationOutput, SubjektivHostError> {
+            Err(SubjektivHostError::Unavailable("no candidates".into()))
+        }
+    }
+
+    #[derive(Clone)]
+    struct ScriptedClient(tokio::sync::mpsc::UnboundedSender<Request>);
+    #[async_trait]
+    impl LlmClient for ScriptedClient {
+        fn clone_boxed(&self) -> Box<dyn LlmClient> {
+            Box::new(self.clone())
+        }
+        async fn stream(
+            &self,
+            request: Request,
+        ) -> Result<Pin<Box<dyn Stream<Item = Result<LlmEvent, ClientError>> + Send>>, ClientError>
+        {
+            self.0.send(request).unwrap();
+            Ok(Box::pin(futures_util::stream::iter(vec![
+                Ok(LlmEvent::text_block_start(0)),
+                Ok(LlmEvent::text_delta(0, "Done.")),
+                Ok(LlmEvent::text_block_stop(0, None)),
+                Ok(LlmEvent::Status(StatusEvent {
+                    status: ResponseStatus::Completed,
+                })),
+            ])))
+        }
+    }
+
+    async fn launch_case(enabled: bool, attach: bool) {
+        let temp = tempfile::tempdir().unwrap();
+        let _runtime_sandbox =
+            crate::runtime::worker_allocation::test_util::RuntimeDirSandbox::new(temp.path());
+        let name = format!("subjektiv-host-{}", uuid::Uuid::now_v7());
+        let mut manifest = WorkerManifest::from_toml(&format!(
+            r#"
+[worker]
+name = "{name}"
+[model]
+scheme = "anthropic"
+model_id = "scripted-model"
+[engine]
+[scope]
+allow = []
+[feature.task]
+enabled = true
+"#
+        ))
+        .unwrap();
+        manifest.feature.subjektiv.profile.enabled = enabled;
+        manifest.feature.subjektiv.profile.extraction.enabled = true;
+        manifest.feature.subjektiv.profile.extraction.threshold = Some(u64::MAX);
+        assert!(manifest.feature.subjektiv.workspace_settings.is_none());
+        let store = CombinedStore::new(
+            FsStore::new(temp.path().join("sessions")).unwrap(),
+            FsWorkerStore::new(temp.path().join("workers")).unwrap(),
+        );
+        let host = Arc::new(LocalHost::default());
+        let (tx, mut rx) = tokio::sync::mpsc::unbounded_channel();
+        let bootstrap = WorkerBootstrap::new(
+            manifest,
+            store.clone(),
+            crate::PromptCatalogSource::builtins_only(),
+            WorkerWorkspaceContext::local_filesystem(None),
+            WorkerFilesystemAuthority::None,
+            WorkerBootstrapLayout::Direct {
+                runtime_base: temp.path().join("runtime"),
+                bash_output_dir: temp.path().join("bash-output"),
+            },
+            WorkerControllerTransport::InProcess,
+        )
+        .with_model_client(ScriptedClient(tx));
+        let bootstrap = if attach {
+            bootstrap.with_subjektiv_host(host.clone())
+        } else {
+            bootstrap
+        };
+        let prepared = bootstrap.prepare().await.unwrap();
+        let expected = SubjektivHostContext {
+            worker_id: name.clone(),
+            session_id: prepared.worker().session_id().to_string(),
+        };
+        assert!(
+            prepared
+                .worker()
+                .manifest()
+                .feature
+                .subjektiv
+                .workspace_settings
+                .is_none()
+        );
+        assert!(prepared.worker().workspace_id().is_none());
+        assert!(
+            prepared
+                .worker()
+                .workspace_client()
+                .workspace_id()
+                .is_none()
+        );
+        let controller = prepared.start().await.unwrap();
+        controller
+            .handle
+            .send(Method::submit_text(
+                protocol::new_submission_request_id(),
+                "Hello",
+            ))
+            .await
+            .unwrap();
+        let request = tokio::time::timeout(Duration::from_secs(5), rx.recv())
+            .await
+            .unwrap()
+            .unwrap();
+        let tools = request
+            .tools
+            .iter()
+            .map(|tool| tool.name.as_str())
+            .collect::<Vec<_>>();
+        assert!(
+            tools.contains(&"TaskCreate"),
+            "ordinary Features must survive"
+        );
+        for name in [
+            "SubjektivMemoryQuery",
+            "SubjektivMemoryRead",
+            "SubjektivMemoryRemember",
+            "SubjektivSessionList",
+            "SubjektivSessionRead",
+        ] {
+            assert_eq!(tools.contains(&name), enabled && attach, "{name}");
+        }
+        assert!(!tools.contains(&"MemoryApplyCandidate"));
+        let rendered = format!(
+            "{}\n{}",
+            request.system_prompt.as_deref().unwrap_or_default(),
+            request
+                .items
+                .iter()
+                .filter_map(|item| item.as_text())
+                .collect::<Vec<_>>()
+                .join("\n")
+        );
+        assert_eq!(
+            rendered.contains("Explicit local subject behavior."),
+            enabled && attach
+        );
+        if enabled && attach {
+            assert_eq!(
+                *host.attributed.lock().unwrap(),
+                vec![(expected.clone(), true)]
+            );
+            assert!(
+                host.contexts
+                    .lock()
+                    .unwrap()
+                    .iter()
+                    .all(|context| context == &expected)
+            );
+            let metadata = store.read_by_name(&name).unwrap().unwrap();
+            assert!(matches!(
+                metadata.subjektiv_session_attribution,
+                Some(session_store::SubjektivSessionAttributionState::Confirmed { .. })
+            ));
+        } else {
+            assert!(host.contexts.lock().unwrap().is_empty());
+            assert!(host.attributed.lock().unwrap().is_empty());
+        }
+        controller
+            .handle
+            .send(Method::Shutdown {
+                command: protocol::WorkerCommandEnvelope::new(1),
+            })
+            .await
+            .unwrap();
+        tokio::time::timeout(Duration::from_secs(5), controller.shutdown)
+            .await
+            .unwrap()
+            .unwrap();
+        controller.controller_task.await.unwrap();
+    }
+
+    #[tokio::test]
+    async fn explicit_local_host_installs_ordinary_features_without_workspace_snapshot() {
+        launch_case(true, true).await;
+    }
+    #[tokio::test]
+    async fn enabled_policy_without_host_is_inert() {
+        launch_case(true, false).await;
+    }
+    #[tokio::test]
+    async fn disabled_policy_with_host_has_no_subjektiv_side_effects() {
+        launch_case(false, true).await;
+    }
+    #[test]
+    fn consolidation_installation_requires_separate_explicit_host_grant() {
+        let connection = SubjektivHostConnection::new(
+            Arc::new(LocalHost::default()),
+            SubjektivHostContext {
+                worker_id: "test-worker".into(),
+                session_id: "test-session".into(),
+            },
+        )
+        .unwrap();
+        assert!(SubjektivConsolidationToolsFeature::new(connection).is_err());
+    }
+}

crates/worker/src/worker.rs

@@ -357,6 +357,9 @@ pub(crate) fn authenticated_input_provenance(
 ) -> WorkerHistoryProvenance {
     match source {
         protocol::AuthenticatedInputSource::UntrustedWire => WorkerHistoryProvenance::LegacyUnknown,
+        protocol::AuthenticatedInputSource::LocalOperator => {
+            WorkerHistoryProvenance::LocalHumanInput
+        }
         protocol::AuthenticatedInputSource::Account { account_id } => {
             WorkerHistoryProvenance::HumanInput {
                 account_id: account_id.clone(),
@@ -2752,6 +2755,7 @@ pub struct Worker<C: LlmClient, St: Store> {
     inject_resident_summary: bool,
     /// Deferred resident prompt source installed by an enabled Feature.
     feature_resident_summary_source: Option<Arc<dyn SystemPromptContributionSource>>,
+    subjektiv_host: Option<Arc<dyn crate::subjektiv::SubjektivHost>>,
     /// One-shot generic Feature lifecycle notification required after restore.
     restore_feature_lifecycle_pending: bool,
     /// Complete system prompt replacement installed by an enabled Feature.
@@ -3014,6 +3018,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
             prompts,
             inject_resident_summary: true,
             feature_resident_summary_source: None,
+            subjektiv_host: None,
             restore_feature_lifecycle_pending: false,
             feature_system_prompt_override: None,
             user_segments: Vec::new(),
@@ -3233,6 +3238,49 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         self.workspace_context.client_handle()
     }
 
+    /// Bind a live capability before Feature installation. This is deliberately
+    /// separate from Workspace authority and is not inherited by SubWorkers.
+    pub fn bind_subjektiv_host(&mut self, host: Option<Arc<dyn crate::subjektiv::SubjektivHost>>) {
+        self.subjektiv_host = host;
+    }
+
+    pub fn subjektiv_host_connection(
+        &self,
+    ) -> std::io::Result<Option<crate::subjektiv::SubjektivHostConnection>> {
+        use crate::subjektiv::{
+            BackendSubjektivHost, SubjektivHostConnection, SubjektivHostContext,
+        };
+        let config = &self.manifest.feature.subjektiv;
+        config.validate_execution().map_err(std::io::Error::other)?;
+        if !config.profile.enabled {
+            return Ok(None);
+        }
+        let host = if let Some(host) = &self.subjektiv_host {
+            if config.workspace_settings.is_some()
+                || self.workspace_id().is_some()
+                || self.workspace_client().workspace_id().is_some()
+            {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::InvalidInput,
+                    "explicit local subjektiv Host cannot replace trusted Backend Workspace authority",
+                ));
+            }
+            Some(host.clone())
+        } else {
+            BackendSubjektivHost::from_resolved_config(config, self.workspace_client_handle())?
+        };
+        host.map(|host| {
+            SubjektivHostConnection::new(
+                host,
+                SubjektivHostContext {
+                    worker_id: self.manifest.worker.name.clone(),
+                    session_id: self.session_id().to_string(),
+                },
+            )
+        })
+        .transpose()
+    }
+
     /// Bind the host-owned Worker-session observation projection. The provider
     /// is responsible for workspace authorization and per-capture revalidation.
     pub fn bind_worker_observation_provider(
@@ -7324,6 +7372,7 @@ where
             prompts: common.prompts,
             inject_resident_summary: true,
             feature_resident_summary_source: None,
+            subjektiv_host: None,
             restore_feature_lifecycle_pending: false,
             feature_system_prompt_override: None,
             user_segments: Vec::new(),
@@ -7433,64 +7482,62 @@ where
             create_if_missing,
             initial_diagnostic.to_string(),
         ))?;
-        let response = match self.workspace_client().execute_server_operation(
-            WorkspaceServerOperation::SubjektivRecordSession(
-                server_api::SubjektivRecordSessionRequest {
-                    session_id: session_id.to_string(),
-                    create_if_missing,
-                },
-            ),
-        ) {
-            Ok(response) => response,
-            Err(error) => {
-                let message = bounded_attribution_diagnostic(format!(
-                    "Session attribution transport outcome is unknown: {error}"
-                ));
-                self.persist_subjektiv_session_attribution(
-                    incomplete_subjektiv_attribution_state(
-                        session_id,
-                        create_if_missing,
-                        message.clone(),
-                    ),
-                )?;
-                return Err(WorkerError::SubjektivSessionAttribution {
-                    state: incomplete_subjektiv_attribution_label(create_if_missing),
-                    message,
-                });
-            }
-        };
-        if !response.is_success() {
-            let message = bounded_attribution_diagnostic(format!(
-                "Session attribution was rejected with HTTP {}: {}",
-                response.status, response.body
-            ));
-            let state = if create_if_missing {
-                SubjektivSessionAttributionState::Failed {
-                    session_id,
-                    diagnostic: message.clone(),
-                }
-            } else {
-                SubjektivSessionAttributionState::LegacyUnknown {
-                    session_id,
-                    diagnostic: message.clone(),
-                }
-            };
-            self.persist_subjektiv_session_attribution(state)?;
-            return Err(WorkerError::SubjektivSessionAttribution {
-                state: if create_if_missing {
-                    "failed"
-                } else {
-                    "legacy_unknown"
-                },
-                message,
-            });
-        }
-        let output: server_api::SubjektivRecordSessionResponse =
-            match serde_json::from_str(&response.body) {
+        let output = if self.subjektiv_host.is_some() {
+            let connection = self
+                .subjektiv_host_connection()
+                .map_err(|error| WorkerError::SubjektivSessionAttribution {
+                    state: "failed",
+                    message: error.to_string(),
+                })?
+                .ok_or_else(|| WorkerError::SubjektivSessionAttribution {
+                    state: "failed",
+                    message: "Session attribution requires enabled explicit Host capability".into(),
+                })?;
+            match connection
+                .host
+                .record_session(&connection.context, create_if_missing)
+            {
                 Ok(output) => output,
+                Err(error) => {
+                    let message = bounded_attribution_diagnostic(error.to_string());
+                    let rejected =
+                        !matches!(error, crate::subjektiv::SubjektivHostError::Unavailable(_));
+                    let state = if create_if_missing && rejected {
+                        SubjektivSessionAttributionState::Failed {
+                            session_id,
+                            diagnostic: message.clone(),
+                        }
+                    } else {
+                        incomplete_subjektiv_attribution_state(
+                            session_id,
+                            create_if_missing,
+                            message.clone(),
+                        )
+                    };
+                    self.persist_subjektiv_session_attribution(state)?;
+                    return Err(WorkerError::SubjektivSessionAttribution {
+                        state: if create_if_missing && rejected {
+                            "failed"
+                        } else {
+                            incomplete_subjektiv_attribution_label(create_if_missing)
+                        },
+                        message,
+                    });
+                }
+            }
+        } else {
+            let response = match self.workspace_client().execute_server_operation(
+                WorkspaceServerOperation::SubjektivRecordSession(
+                    server_api::SubjektivRecordSessionRequest {
+                        session_id: session_id.to_string(),
+                        create_if_missing,
+                    },
+                ),
+            ) {
+                Ok(response) => response,
                 Err(error) => {
                     let message = bounded_attribution_diagnostic(format!(
-                        "Session attribution response outcome is unknown: {error}"
+                        "Session attribution transport outcome is unknown: {error}"
                     ));
                     self.persist_subjektiv_session_attribution(
                         incomplete_subjektiv_attribution_state(
@@ -7505,6 +7552,54 @@ where
                     });
                 }
             };
+            if !response.is_success() {
+                let message = bounded_attribution_diagnostic(format!(
+                    "Session attribution was rejected with HTTP {}: {}",
+                    response.status, response.body
+                ));
+                let state = if create_if_missing {
+                    SubjektivSessionAttributionState::Failed {
+                        session_id,
+                        diagnostic: message.clone(),
+                    }
+                } else {
+                    SubjektivSessionAttributionState::LegacyUnknown {
+                        session_id,
+                        diagnostic: message.clone(),
+                    }
+                };
+                self.persist_subjektiv_session_attribution(state)?;
+                return Err(WorkerError::SubjektivSessionAttribution {
+                    state: if create_if_missing {
+                        "failed"
+                    } else {
+                        "legacy_unknown"
+                    },
+                    message,
+                });
+            }
+            let output: server_api::SubjektivRecordSessionResponse =
+                match serde_json::from_str(&response.body) {
+                    Ok(output) => output,
+                    Err(error) => {
+                        let message = bounded_attribution_diagnostic(format!(
+                            "Session attribution response outcome is unknown: {error}"
+                        ));
+                        self.persist_subjektiv_session_attribution(
+                            incomplete_subjektiv_attribution_state(
+                                session_id,
+                                create_if_missing,
+                                message.clone(),
+                            ),
+                        )?;
+                        return Err(WorkerError::SubjektivSessionAttribution {
+                            state: incomplete_subjektiv_attribution_label(create_if_missing),
+                            message,
+                        });
+                    }
+                };
+            output
+        };
         if output.session_id != session_id.to_string()
             || output.subject_id.trim().is_empty()
             || output.subject_id.chars().any(char::is_control)
@@ -7616,6 +7711,7 @@ where
             prompts: common.prompts,
             inject_resident_summary: true,
             feature_resident_summary_source: None,
+            subjektiv_host: None,
             restore_feature_lifecycle_pending: false,
             feature_system_prompt_override: None,
             user_segments: Vec::new(),
@@ -7744,6 +7840,7 @@ where
             prompts: common.prompts,
             inject_resident_summary: true,
             feature_resident_summary_source: None,
+            subjektiv_host: None,
             restore_feature_lifecycle_pending: false,
             feature_system_prompt_override: None,
             user_segments: Vec::new(),
@@ -8130,6 +8227,7 @@ where
             prompts: common.prompts,
             inject_resident_summary: true,
             feature_resident_summary_source: None,
+            subjektiv_host: None,
             restore_feature_lifecycle_pending: true,
             feature_system_prompt_override: None,
             user_segments: state.user_segments,
@@ -9852,7 +9950,7 @@ mod build_summary_prompt_tests {
         });
         let source = crate::feature::builtin::memory::ordinary_subjektiv_resident_summary_source(
             manifest,
-            client,
+            Some(crate::subjektiv::test_connection(client)),
             Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())),
         )
         .unwrap()
@@ -14731,12 +14829,14 @@ permission = "write"
         let resident_source = || {
             crate::feature::builtin::memory::ordinary_subjektiv_resident_summary_source(
                 &manifest,
-                Arc::new(RestoreSubjectResidentClient {
-                    availability: memory::backend::MemoryResidentSummaryAvailability::Ready,
-                    content: Some("literal /prepare()".into()),
-                    load_count: Arc::default(),
-                    behavior_revision: Some(behavior_revision.clone()),
-                }),
+                Some(crate::subjektiv::test_connection(Arc::new(
+                    RestoreSubjectResidentClient {
+                        availability: memory::backend::MemoryResidentSummaryAvailability::Ready,
+                        content: Some("literal /prepare()".into()),
+                        load_count: Arc::default(),
+                        behavior_revision: Some(behavior_revision.clone()),
+                    },
+                ))),
                 Arc::new(ArcSwap::from(PromptCatalog::builtins_only().unwrap())),
             )
             .unwrap()