diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index b0096c7df..87e84f9c0 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -787,9 +787,9 @@ impl WorkerController { // Runtime-owned Workers persist their current model-visible head before // controller exposure. Feature installation above must happen first so a // fresh subject Worker captures the current resident surface in its durable - // prompt. Restored Workers defer their one-shot append-only resident refresh - // until the next model-visible context materialization, after the Runtime has - // committed restoration of the existing durable head. + // prompt. Restored Workers defer generic Feature restore notification until + // the next model-visible context materialization; only an installed + // Feature may then request a typed durable history append. if runtime_managed && worker.needs_initial_session_head_materialization() { worker .materialize_durable_session_head() @@ -1342,6 +1342,21 @@ where worker.workspace_client_handle(), )? { + let workspace_id = worker + .workspace_client() + .workspace_id() + .expect("validated subject resident source has Workspace identity") + .to_string(); + if let Some(refresh_feature) = + crate::feature::builtin::memory::SubjektivResidentRestoreRefreshFeature::for_host( + worker.manifest_lifecycle_features_enabled(), + Arc::clone(&resident_summary_source), + worker.prompts(), + workspace_id, + ) + { + feature_registry.add_module(refresh_feature); + } feature_prompt_contribution = Some((Some(resident_summary_source), None)); } if ordinary_subjektiv_features_enabled diff --git a/crates/worker/src/feature.rs b/crates/worker/src/feature.rs index 62665c09d..41dfdcc56 100644 --- a/crates/worker/src/feature.rs +++ b/crates/worker/src/feature.rs @@ -25,7 +25,8 @@ use thiserror::Error; use crate::hook::{ BeforeSessionRewrite, Hook, HookExecutionPolicy, HookRegistryBuilder, OnPromptSubmit, - OnTurnEnd, PostToolCall, PreLlmRequest, PreToolCall, RunCommitted, RunExit, WorkerStopping, + OnTurnEnd, PostToolCall, PreLlmRequest, PreToolCall, RunCommitted, RunExit, WorkerRestored, + WorkerStopping, }; use background::{ BackgroundTaskSpec, FeatureBackgroundTask, FeatureBackgroundTaskRegistry, @@ -268,6 +269,7 @@ pub enum FeatureHookPoint { RunExit, RunCommitted, BeforeSessionRewrite, + WorkerRestored, WorkerStopping, } @@ -1214,6 +1216,25 @@ impl HookContributionRegistrar<'_> { Ok(()) } + pub fn add_worker_restored( + &mut self, + name: impl Into, + policy: HookExecutionPolicy, + hook: impl Hook + 'static, + ) -> Result<(), FeatureInstallError> { + let declaration = HookDeclaration::new(name, FeatureHookPoint::WorkerRestored); + self.require_declared(&declaration)?; + self.hook_builder + .add_named_worker_restored( + format!("{}:{}", self.feature_id, declaration.name), + policy, + hook, + ) + .map_err(|error| FeatureInstallError::InvalidDescriptor(error.to_string()))?; + self.record(declaration); + Ok(()) + } + pub fn add_worker_stopping( &mut self, name: impl Into, diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index f4199d0d5..62411f540 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -7,6 +7,7 @@ use std::sync::Arc; use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput}; +use arc_swap::ArcSwap; use async_trait::async_trait; use memory::backend::{ MemoryBackendHttpResponse, MemoryBackendOperation, MemoryBackendOperationResult, @@ -19,9 +20,11 @@ use serde::de::DeserializeOwned; use serde_json::json; use crate::feature::{ - FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution, - ToolDeclaration, + FeatureDescriptor, FeatureHookPoint, FeatureInstallContext, FeatureInstallError, FeatureModule, + HookDeclaration, ToolContribution, ToolDeclaration, }; +use crate::hook::{Hook, HookError, HookErrorCategory, HookExecutionPolicy, WorkerRestored}; +use crate::prompt::catalog::{PromptCatalog, WorkerPrompt}; use crate::worker::{ SystemPromptContribution, SystemPromptContributionSource, WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod, @@ -722,6 +725,102 @@ impl SystemPromptContributionSource for WorkspaceSubjektivResidentSummarySource } } +const SUBJEKTIV_RESTORE_REFRESH_HOOK: &str = "refresh-resident-surface-after-restore"; + +/// subjektiv-owned restore behavior. Initial prompt contribution remains a +/// shared host facility, while loading and rendering a post-restore refresh is +/// installed only with an execution-enabled, subject-attached subjektiv Feature. +#[derive(Clone)] +pub(crate) struct SubjektivResidentRestoreRefreshFeature { + source: Arc, + prompts: Arc>, + workspace_id: String, +} + +impl SubjektivResidentRestoreRefreshFeature { + pub(crate) fn for_host( + lifecycle_enabled: bool, + source: Arc, + prompts: Arc>, + workspace_id: impl Into, + ) -> Option { + lifecycle_enabled.then(|| Self::new(source, prompts, workspace_id)) + } + + pub(crate) fn new( + source: Arc, + prompts: Arc>, + workspace_id: impl Into, + ) -> Self { + Self { + source, + prompts, + workspace_id: workspace_id.into(), + } + } +} + +impl FeatureModule for SubjektivResidentRestoreRefreshFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin( + "subjektiv-resident-restore-refresh", + "subjektiv Resident Restore Refresh", + ) + .with_description( + "Refreshes the current subject resident Memory surface at a Worker restore boundary.", + ) + .with_hook(HookDeclaration::new( + SUBJEKTIV_RESTORE_REFRESH_HOOK, + FeatureHookPoint::WorkerRestored, + )) + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + context.hooks().add_worker_restored( + SUBJEKTIV_RESTORE_REFRESH_HOOK, + HookExecutionPolicy::fail_closed(), + SubjektivResidentRestoreRefreshHook { + source: Arc::clone(&self.source), + prompts: Arc::clone(&self.prompts), + workspace_id: self.workspace_id.clone(), + }, + ) + } +} + +struct SubjektivResidentRestoreRefreshHook { + source: Arc, + prompts: Arc>, + workspace_id: String, +} + +#[async_trait] +impl Hook for SubjektivResidentRestoreRefreshHook { + async fn call(&self, context: &crate::hook::WorkerRestoredContext) -> Result<(), HookError> { + let contribution = self.source.load().await; + let (surface_ready, summary) = match &contribution { + SystemPromptContribution::Ready(summary) => (true, Some(summary.as_str())), + SystemPromptContribution::Unavailable => (false, None), + }; + let prompts = self.prompts.load_full(); + let body = prompts + .resident_memory_restore_section(surface_ready, summary) + .map_err(|error| HookError::new(HookErrorCategory::Internal, error.to_string()))?; + let projection = prompts.projection(); + context.system_items().append_resident_summary_refresh( + body, + session_store::PromptRenderProvenance { + workspace_id: Some(self.workspace_id.clone()), + config_revision: projection.config_revision, + source_digest: projection.source_digest.clone(), + projection_digest: projection.catalog_digest.clone(), + logical_name: WorkerPrompt::ResidentMemoryRestoreSection.key().to_string(), + }, + ); + Ok(()) + } +} + pub(crate) struct MemoryFeatureInstallPlan { pub(crate) module: MemoryToolsFeature, pub(crate) resident_summary_source: Option>, @@ -1463,10 +1562,59 @@ permission = "write" .is_none() ); + let mut policy_only = disabled.clone(); + policy_only.feature.subjektiv.profile.enabled = true; + let unattached_client = Arc::new(SubjectResidentClient::new( + memory::backend::MemoryResidentSummaryAvailability::Ready, + Some("policy-only surface must not load"), + )); + assert!( + ordinary_subjektiv_resident_summary_source(&policy_only, unattached_client.clone()) + .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()).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()) + .unwrap() + .unwrap(); + let prompts = Arc::new(ArcSwap::from( + crate::prompt::catalog::PromptCatalog::builtins_only().unwrap(), + )); + + assert!( + SubjektivResidentRestoreRefreshFeature::for_host(false, source, prompts, "workspace",) + .is_none() + ); + } + + #[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()) + .unwrap() + .unwrap(); + let feature = SubjektivResidentRestoreRefreshFeature::new( + source, + Arc::new(ArcSwap::from( + crate::prompt::catalog::PromptCatalog::builtins_only().unwrap(), + )), + "workspace", + ); + let descriptor = feature.descriptor(); + + assert!(descriptor.tools.is_empty()); + assert_eq!(descriptor.hooks.len(), 1); + assert_eq!(descriptor.hooks[0].point, FeatureHookPoint::WorkerRestored); + } + #[test] fn memory_feature_owns_normal_and_staging_tool_surfaces() { let normal = MemoryToolsFeature::new(test_client(), false); diff --git a/crates/worker/src/hook.rs b/crates/worker/src/hook.rs index 1bebebfb2..6e0445d87 100644 --- a/crates/worker/src/hook.rs +++ b/crates/worker/src/hook.rs @@ -283,6 +283,34 @@ impl SystemItemAppendHandle { } } +/// Host-created narrow append authority exposed only at a restore lifecycle +/// boundary. It cannot emit arbitrary system items. +pub struct RestoreSystemItemAppendHandle { + pending: Arc>>, +} + +impl RestoreSystemItemAppendHandle { + pub(crate) fn new(pending: Arc>>) -> Self { + Self { pending } + } + + /// Queue a Feature-rendered resident-summary refresh for durable append at + /// the restore lifecycle boundary. + pub fn append_resident_summary_refresh( + &self, + body: impl Into, + prompt_provenance: session_store::PromptRenderProvenance, + ) { + self.pending + .lock() + .expect("restore system-item append queue poisoned") + .push(SystemItem::ResidentSummaryRefresh { + body: body.into(), + prompt_provenance: Some(prompt_provenance), + }); + } +} + // ============================================================================= // Hook input summary/context types (read-only) // ============================================================================= @@ -524,9 +552,34 @@ pub struct WorkerStoppingContext { pub reason: String, } +/// Context delivered once after a persisted Worker has been restored and all +/// enabled Features have been installed, before the next model-visible turn. +pub struct WorkerRestoredContext { + pub invocation: HookInvocationContext, + system_items: RestoreSystemItemAppendHandle, +} + +impl WorkerRestoredContext { + pub(crate) fn new( + invocation: HookInvocationContext, + system_items: RestoreSystemItemAppendHandle, + ) -> Self { + Self { + invocation, + system_items, + } + } + + /// Narrow durable append authority for restore-owned typed system items. + pub fn system_items(&self) -> &RestoreSystemItemAppendHandle { + &self.system_items + } +} + pub struct RunExit; pub struct RunCommitted; pub struct BeforeSessionRewrite; +pub struct WorkerRestored; pub struct WorkerStopping; impl HookEventKind for RunExit { @@ -544,6 +597,11 @@ impl HookEventKind for BeforeSessionRewrite { type Output = BeforeSessionRewriteAction; } +impl HookEventKind for WorkerRestored { + type Input = WorkerRestoredContext; + type Output = (); +} + impl HookEventKind for WorkerStopping { type Input = WorkerStoppingContext; type Output = (); @@ -609,6 +667,7 @@ pub struct HookRegistryBuilder { run_exit: Vec>, run_committed: Vec>, before_session_rewrite: Vec>, + worker_restored: Vec>, worker_stopping: Vec>, } @@ -684,6 +743,12 @@ impl HookRegistryBuilder { before_session_rewrite, BeforeSessionRewrite ); + add_hook_methods!( + add_worker_restored, + add_named_worker_restored, + worker_restored, + WorkerRestored + ); add_hook_methods!( add_worker_stopping, add_named_worker_stopping, @@ -691,7 +756,7 @@ impl HookRegistryBuilder { WorkerStopping ); - pub(crate) fn checkpoint(&self) -> [usize; 9] { + pub(crate) fn checkpoint(&self) -> [usize; 10] { [ self.on_prompt_submit.len(), self.pre_llm_request.len(), @@ -701,11 +766,12 @@ impl HookRegistryBuilder { self.run_exit.len(), self.run_committed.len(), self.before_session_rewrite.len(), + self.worker_restored.len(), self.worker_stopping.len(), ] } - pub(crate) fn rollback_to(&mut self, checkpoint: [usize; 9]) { + pub(crate) fn rollback_to(&mut self, checkpoint: [usize; 10]) { self.on_prompt_submit.truncate(checkpoint[0]); self.pre_llm_request.truncate(checkpoint[1]); self.pre_tool_call.truncate(checkpoint[2]); @@ -714,7 +780,8 @@ impl HookRegistryBuilder { self.run_exit.truncate(checkpoint[5]); self.run_committed.truncate(checkpoint[6]); self.before_session_rewrite.truncate(checkpoint[7]); - self.worker_stopping.truncate(checkpoint[8]); + self.worker_restored.truncate(checkpoint[8]); + self.worker_stopping.truncate(checkpoint[9]); } pub fn build(self) -> HookRegistry { @@ -727,6 +794,7 @@ impl HookRegistryBuilder { run_exit: self.run_exit, run_committed: self.run_committed, before_session_rewrite: self.before_session_rewrite, + worker_restored: self.worker_restored, worker_stopping: self.worker_stopping, diagnostics: std::sync::Mutex::new(Vec::new()), } @@ -743,6 +811,7 @@ pub struct HookRegistry { run_exit: Vec>, run_committed: Vec>, before_session_rewrite: Vec>, + worker_restored: Vec>, worker_stopping: Vec>, diagnostics: std::sync::Mutex>, } @@ -840,6 +909,26 @@ impl HookRegistry { .unwrap_or(BeforeSessionRewriteAction::Continue)) } + pub async fn on_worker_restored( + &self, + context: &WorkerRestoredContext, + ) -> Result<(), HookExecutionError> { + for registration in &self.worker_restored { + if let Err(error) = registration.call(context).await { + self.record_diagnostic(error.clone()); + match error.policy { + HookFailurePolicy::FailOpenWithDiagnostic => { + tracing::warn!(owner = %error.owner, error = %error.source, "worker-restored hook failed open"); + } + HookFailurePolicy::FailClosed | HookFailurePolicy::AttentionRequired => { + return Err(error); + } + } + } + } + Ok(()) + } + pub async fn on_worker_stopping( &self, context: &WorkerStoppingContext, @@ -881,6 +970,29 @@ mod tests { } } + #[test] + fn restore_append_handle_queues_only_resident_refresh_items() { + let pending = Arc::new(Mutex::new(Vec::new())); + let handle = RestoreSystemItemAppendHandle::new(Arc::clone(&pending)); + let provenance = session_store::PromptRenderProvenance { + workspace_id: Some("workspace".into()), + config_revision: 1, + source_digest: "source".into(), + projection_digest: "projection".into(), + logical_name: "internal.resident_memory_restore_section".into(), + }; + + handle.append_resident_summary_refresh("current surface", provenance.clone()); + + assert!(matches!( + pending.lock().unwrap().as_slice(), + [SystemItem::ResidentSummaryRefresh { + body, + prompt_provenance: Some(actual), + }] if body == "current surface" && actual == &provenance + )); + } + #[test] fn pre_request_context_exposes_handle_only_when_host_supplies_one() { let info = PreRequestInfo { diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 1e0441297..1c4c04752 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -62,8 +62,9 @@ use crate::feature::{ use crate::hook::{ BeforeSessionRewriteAction, BeforeSessionRewriteContext, Hook, HookHistoryRange, HookInvocationContext, HookRegistry, HookRegistryBuilder, OnPromptSubmit, OnTurnEnd, - PostToolCall, PreLlmRequest, PreToolCall, RunCommittedContext, RunCommittedExit, - RunExitContext, SessionRewriteKind, WorkerStoppingContext, + PostToolCall, PreLlmRequest, PreToolCall, RestoreSystemItemAppendHandle, RunCommittedContext, + RunCommittedExit, RunExitContext, SessionRewriteKind, WorkerRestoredContext, + WorkerStoppingContext, }; use crate::in_flight::InFlightEvents; use crate::internal_worker::{ @@ -411,9 +412,7 @@ use crate::ipc::alerter::Alerter; use crate::ipc::interceptor::WorkerInterceptor; use crate::ipc::notify_buffer::NotifyBuffer; use crate::prompt::agents_md::read_agents_md; -use crate::prompt::catalog::{ - CatalogError, PromptCatalog, WorkerPrompt, WorkspacePromptProjection, -}; +use crate::prompt::catalog::{CatalogError, PromptCatalog, WorkspacePromptProjection}; use crate::prompt::source::PromptCatalogSource; use crate::prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTemplate}; use crate::runtime::dir; @@ -2513,8 +2512,8 @@ pub struct Worker { inject_resident_summary: bool, /// Deferred resident prompt source installed by an enabled Feature. feature_resident_summary_source: Option>, - /// One-shot append-only resident refresh required at a restore boundary. - restore_resident_summary_refresh_pending: bool, + /// One-shot generic Feature lifecycle notification required after restore. + restore_feature_lifecycle_pending: bool, /// Complete system prompt replacement installed by an enabled Feature. feature_system_prompt_override: Option, /// Typed user submissions in submit order. K-th entry corresponds to @@ -2768,7 +2767,7 @@ impl Worker { prompts, inject_resident_summary: true, feature_resident_summary_source: None, - restore_resident_summary_refresh_pending: false, + restore_feature_lifecycle_pending: false, feature_system_prompt_override: None, user_segments: Vec::new(), sink: SegmentLogSink::new(), @@ -3843,63 +3842,71 @@ impl Worker { } } - async fn ensure_restore_resident_summary_refreshed(&mut self) -> Result<(), WorkerError> + async fn notify_features_of_restore(&mut self) -> Result<(), WorkerError> where St: Clone + 'static, { - if !self.restore_resident_summary_refresh_pending { + if !self.restore_feature_lifecycle_pending { return Ok(()); } - let Some(source) = self.feature_resident_summary_source.clone() else { - self.restore_resident_summary_refresh_pending = false; - return Ok(()); - }; - let contribution = if self.inject_resident_summary { - source.load().await - } else { - SystemPromptContribution::Unavailable - }; - let (surface_ready, summary) = match &contribution { - SystemPromptContribution::Ready(summary) => (true, Some(summary.as_str())), - SystemPromptContribution::Unavailable => (false, None), - }; - let prompt_catalog = self.prompts.load_full(); - let body = prompt_catalog.resident_memory_restore_section(surface_ready, summary)?; - let prompt_provenance = - Some(self.prompt_render_provenance(WorkerPrompt::ResidentMemoryRestoreSection.key())); - let item = SystemItem::ResidentSummaryRefresh { - body: body.clone(), - prompt_provenance, - }; - let metadata = new_history_metadata( - WorkerHistoryProvenance::BackendInstruction { operation_id: None }, - None, + self.ensure_interceptor_installed(); + let pending_system_items = Arc::new(Mutex::new(Vec::new())); + if let Some(hooks) = self.hook_registry.clone() { + let context = WorkerRestoredContext::new( + self.hook_invocation_context(None), + RestoreSystemItemAppendHandle::new(Arc::clone(&pending_system_items)), + ); + tokio::time::timeout( + FEATURE_HOOK_CHAIN_TIMEOUT, + hooks.on_worker_restored(&context), + ) + .await + .map_err(|_| { + hooks.record_chain_timeout("worker-restored"); + WorkerError::FeatureLifecycle( + "worker-restored hook chain exceeded its host deadline".to_string(), + ) + })? + .map_err(|error| WorkerError::FeatureLifecycle(error.to_string()))?; + } + + let system_items = std::mem::take( + &mut *pending_system_items + .lock() + .expect("restore system-item queue poisoned"), ); - self.commit_entry(LogEntry::AnnotatedSystemItem { - ts: segment_log::now_millis(), - entry: session_store::LoggedSystemHistoryEntry { - item, - metadata: metadata.clone(), - }, - extensions: Vec::new(), - })?; - let history_entry = HistoryEntry::new(Item::system_message(body), metadata); - let mut annotate = history_annotator( - self.log_writer_handle(), - vec![history_entry.clone()], - self.pending_committed_history.clone(), - ); - let (engine, session) = ( - self.engine.as_mut().expect("worker present"), - &mut self.session, - ); - engine.append_history_with( - session.history_mut(), - std::iter::once(history_entry.item), - &mut annotate, - )?; - session.note_mutation(); - self.restore_resident_summary_refresh_pending = false; + for item in system_items { + let history_item = item.to_history_item(); + let metadata = new_history_metadata( + WorkerHistoryProvenance::BackendInstruction { operation_id: None }, + None, + ); + self.commit_entry(LogEntry::AnnotatedSystemItem { + ts: segment_log::now_millis(), + entry: session_store::LoggedSystemHistoryEntry { + item, + metadata: metadata.clone(), + }, + extensions: Vec::new(), + })?; + let history_entry = HistoryEntry::new(history_item, metadata); + let mut annotate = history_annotator( + self.log_writer_handle(), + vec![history_entry.clone()], + self.pending_committed_history.clone(), + ); + let (engine, session) = ( + self.engine.as_mut().expect("worker present"), + &mut self.session, + ); + engine.append_history_with( + session.history_mut(), + std::iter::once(history_entry.item), + &mut annotate, + )?; + session.note_mutation(); + } + self.restore_feature_lifecycle_pending = false; Ok(()) } @@ -3915,7 +3922,7 @@ impl Worker { St: Clone + 'static, { let Some(template) = self.system_prompt_template.take() else { - return self.ensure_restore_resident_summary_refreshed().await; + return self.notify_features_of_restore().await; }; if let Some(rendered) = self.feature_system_prompt_override.take() { self.engine @@ -6653,7 +6660,7 @@ where prompts: common.prompts, inject_resident_summary: true, feature_resident_summary_source: None, - restore_resident_summary_refresh_pending: false, + restore_feature_lifecycle_pending: false, feature_system_prompt_override: None, user_segments: Vec::new(), sink: SegmentLogSink::new(), @@ -6743,7 +6750,7 @@ where prompts: common.prompts, inject_resident_summary: true, feature_resident_summary_source: None, - restore_resident_summary_refresh_pending: false, + restore_feature_lifecycle_pending: false, feature_system_prompt_override: None, user_segments: Vec::new(), sink: SegmentLogSink::new(), @@ -6868,7 +6875,7 @@ where prompts: common.prompts, inject_resident_summary: true, feature_resident_summary_source: None, - restore_resident_summary_refresh_pending: false, + restore_feature_lifecycle_pending: false, feature_system_prompt_override: None, user_segments: Vec::new(), sink: SegmentLogSink::new(), @@ -7250,7 +7257,7 @@ where prompts: common.prompts, inject_resident_summary: true, feature_resident_summary_source: None, - restore_resident_summary_refresh_pending: true, + restore_feature_lifecycle_pending: true, feature_system_prompt_override: None, user_segments: state.user_segments, // Seed the mirror with the entries we just replayed so a @@ -8838,6 +8845,130 @@ mod build_summary_prompt_tests { ) } + #[derive(Debug)] + struct RestoreLegacyResidentClient { + content: String, + load_count: Arc, + } + + impl WorkspaceClient for RestoreLegacyResidentClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-test") + } + + fn kind(&self) -> &str { + "restore-legacy-resident-test" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &self, + request: WorkspaceRequest, + ) -> Result { + let request: server_api::MemoryBackendRequest = + serde_json::from_str(request.body.as_deref().unwrap_or_default()).unwrap(); + assert!(matches!( + request.0, + memory::backend::MemoryBackendOperation::ResidentSummary(_) + )); + self.load_count.fetch_add(1, Ordering::SeqCst); + Ok(WorkspaceResponse { + status: 200, + body: serde_json::to_string(&server_api::MemoryBackendResponse( + memory::backend::MemoryBackendHttpResponse::Ok { + result: memory::backend::MemoryBackendOperationResult::ResidentSummary( + memory::backend::MemoryResidentSummaryOutput { + availability: + memory::backend::MemoryResidentSummaryAvailability::Ready, + content: Some(self.content.clone()), + }, + ), + }, + )) + .unwrap(), + }) + } + } + + fn legacy_resident_client(content: &str) -> (Arc, Arc) { + let load_count = Arc::new(AtomicUsize::new(0)); + ( + Arc::new(RestoreLegacyResidentClient { + content: content.to_string(), + load_count: Arc::clone(&load_count), + }), + load_count, + ) + } + + #[derive(Debug)] + struct RestoreSubjectResidentClient { + availability: memory::backend::MemoryResidentSummaryAvailability, + content: Option, + load_count: Arc, + } + + impl WorkspaceClient for RestoreSubjectResidentClient { + fn workspace_id(&self) -> Option<&str> { + Some("workspace-test") + } + + fn kind(&self) -> &str { + "restore-subject-resident-test" + } + + fn is_available(&self) -> bool { + true + } + + fn execute( + &self, + request: WorkspaceRequest, + ) -> Result { + let request: server_api::SubjektivMemoryBackendRequest = + serde_json::from_str(request.body.as_deref().unwrap_or_default()).unwrap(); + assert!(matches!( + request.operation, + server_api::SubjektivMemoryBackendOperation::ResidentSummary(_) + )); + self.load_count.fetch_add(1, Ordering::SeqCst); + Ok(WorkspaceResponse { + status: 200, + body: serde_json::to_string( + &server_api::SubjektivMemoryBackendResponse::ResidentSummary( + memory::backend::MemoryResidentSummaryOutput { + availability: self.availability, + content: self.content.clone(), + }, + ), + ) + .unwrap(), + }) + } + } + + fn subject_resident_source_for_restore( + manifest: &WorkerManifest, + availability: memory::backend::MemoryResidentSummaryAvailability, + content: Option<&str>, + ) -> (Arc, Arc) { + let load_count = Arc::new(AtomicUsize::new(0)); + let client: Arc = Arc::new(RestoreSubjectResidentClient { + availability, + content: content.map(str::to_string), + load_count: Arc::clone(&load_count), + }); + let source = crate::feature::builtin::memory::ordinary_subjektiv_resident_summary_source( + manifest, client, + ) + .unwrap() + .unwrap(); + (source, load_count) + } + fn test_summary_input(items: &[Item]) -> String { build_summary_input( items, @@ -11199,7 +11330,7 @@ mod build_summary_prompt_tests { session_store::FsStore::new(dir.path().join("sessions")).unwrap(), session_store::FsWorkerStore::new(dir.path().join("workers")).unwrap(), ); - let manifest = WorkerManifest::from_toml(&format!( + let mut manifest = WorkerManifest::from_toml(&format!( r#" [worker] name = "restore-resident-{}" @@ -11219,6 +11350,16 @@ permission = "write" cwd.display() )) .unwrap(); + manifest.feature.memory.profile.enabled = true; + manifest + .feature + .memory + .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: "workspace-test".to_string(), + settings_revision: 1, + language: "English".to_string(), + }) + .unwrap(); let authority = WorkerFilesystemAuthority::local(cwd.clone(), cwd.clone()); let scope = Scope::writable(&cwd).unwrap(); let mut original = Worker::new( @@ -11238,10 +11379,22 @@ permission = "write" ) .unwrap(), ); - let (old_source, _) = - test_system_prompt_contribution_source(Some("old subject surface".into())); - original.install_system_prompt_contribution(Some(old_source), None); + let (legacy_client, initial_legacy_loads) = + legacy_resident_client("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH"); + let legacy_plan = crate::feature::builtin::memory::MemoryFeatureInstallPlan::prepare( + &manifest, + legacy_client, + original.prompts().load_full(), + ) + .unwrap() + .unwrap(); + let legacy_source = legacy_plan.resident_summary_source.clone(); + let report = original + .install_features(FeatureRegistryBuilder::new().with_module(legacy_plan.module)); + assert!(!report.has_errors()); + original.install_system_prompt_contribution(legacy_source, None); original.materialize_durable_session_head().await.unwrap(); + assert_eq!(initial_legacy_loads.load(Ordering::SeqCst), 1); let session_id = original.session_id(); let segment_id = original.segment_id(); assert!( @@ -11249,11 +11402,14 @@ permission = "write" .engine() .get_system_prompt() .unwrap() - .contains("old subject surface") + .contains("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH") ); drop(original); - let mut restored = + // A legacy Memory resident source remains an initial-prompt contribution + // only. The generic restore lifecycle does not infer subjektiv behavior + // from the presence of a shared resident source. + let mut legacy_only_restore = Worker::, _>::restore_from_manifest_with_context_and_model_client( session_id, segment_id, @@ -11266,10 +11422,69 @@ permission = "write" ) .await .unwrap(); + let legacy_history = legacy_only_restore.history().to_vec(); + let (legacy_client, legacy_loads) = + legacy_resident_client("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH"); + let legacy_plan = crate::feature::builtin::memory::MemoryFeatureInstallPlan::prepare( + &manifest, + legacy_client, + legacy_only_restore.prompts().load_full(), + ) + .unwrap() + .unwrap(); + let legacy_source = legacy_plan.resident_summary_source.clone(); + let report = legacy_only_restore + .install_features(FeatureRegistryBuilder::new().with_module(legacy_plan.module)); + assert!(!report.has_errors()); + legacy_only_restore.install_system_prompt_contribution(legacy_source, None); + legacy_only_restore + .materialize_durable_session_head() + .await + .unwrap(); + assert_eq!(legacy_loads.load(Ordering::SeqCst), 0); + assert_eq!(legacy_only_restore.history(), legacy_history); + drop(legacy_only_restore); + + let mut subject_manifest = manifest.clone(); + subject_manifest.feature.memory = Default::default(); + subject_manifest.feature.subjektiv.profile.enabled = true; + subject_manifest + .feature + .subjektiv + .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot { + workspace_id: "workspace-test".to_string(), + settings_revision: 1, + language: "English".to_string(), + }) + .unwrap(); + + let mut restored = + Worker::, _>::restore_from_manifest_with_context_and_model_client( + session_id, + segment_id, + subject_manifest.clone(), + store.clone(), + crate::prompt::source::PromptCatalogSource::builtins_only(), + WorkerWorkspaceContext::no_workspace(), + authority.clone(), + Some(Box::new(NoopClient)), + ) + .await + .unwrap(); let history_before = restored.history().to_vec(); - let (current_source, current_loads) = - test_system_prompt_contribution_source(Some("latest subject surface".into())); - restored.install_system_prompt_contribution(Some(current_source), None); + let (current_source, current_loads) = subject_resident_source_for_restore( + &subject_manifest, + memory::backend::MemoryResidentSummaryAvailability::Ready, + Some("latest subject surface"), + ); + let report = restored.install_features(FeatureRegistryBuilder::new().with_module( + crate::feature::builtin::memory::SubjektivResidentRestoreRefreshFeature::new( + current_source, + restored.prompts(), + "workspace-test", + ), + )); + assert!(!report.has_errors()); restored.materialize_durable_session_head().await.unwrap(); assert_eq!(current_loads.load(Ordering::SeqCst), 1); @@ -11283,7 +11498,7 @@ permission = "write" .to_string(); assert!(refresh.contains("latest subject surface")); assert!(refresh.contains("supersedes every resident memory summary")); - assert!(!refresh.contains("old subject surface")); + assert!(!refresh.contains("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH")); let refreshed_len = restored.history().len(); restored.materialize_durable_session_head().await.unwrap(); assert_eq!(restored.history().len(), refreshed_len); @@ -11293,7 +11508,7 @@ permission = "write" Worker::, _>::restore_from_manifest_with_context_and_model_client( session_id, segment_id, - manifest.clone(), + subject_manifest.clone(), store.clone(), crate::prompt::source::PromptCatalogSource::builtins_only(), WorkerWorkspaceContext::no_workspace(), @@ -11302,9 +11517,19 @@ permission = "write" ) .await .unwrap(); - let (empty_source, empty_loads) = - test_system_prompt_contribution_source(Some(String::new())); - empty_restore.install_system_prompt_contribution(Some(empty_source), None); + let (empty_source, empty_loads) = subject_resident_source_for_restore( + &subject_manifest, + memory::backend::MemoryResidentSummaryAvailability::Ready, + None, + ); + let report = empty_restore.install_features(FeatureRegistryBuilder::new().with_module( + crate::feature::builtin::memory::SubjektivResidentRestoreRefreshFeature::new( + empty_source, + empty_restore.prompts(), + "workspace-test", + ), + )); + assert!(!report.has_errors()); empty_restore .materialize_durable_session_head() .await @@ -11323,52 +11548,70 @@ permission = "write" assert!(!empty_boundary.contains("No current ready resident memory surface")); drop(empty_restore); - let mut stale_restore = - Worker::, _>::restore_from_manifest_with_context_and_model_client( + for availability in [ + memory::backend::MemoryResidentSummaryAvailability::Ungenerated, + memory::backend::MemoryResidentSummaryAvailability::Stale, + memory::backend::MemoryResidentSummaryAvailability::Failed, + ] { + let mut unavailable_restore = Worker::, _>::restore_from_manifest_with_context_and_model_client( session_id, segment_id, - manifest, + subject_manifest.clone(), store.clone(), crate::prompt::source::PromptCatalogSource::builtins_only(), WorkerWorkspaceContext::no_workspace(), - authority, + authority.clone(), Some(Box::new(NoopClient)), ) .await .unwrap(); - let stale_history_before = stale_restore.history().to_vec(); - let (stale_source, stale_loads) = test_system_prompt_contribution_source(None); - stale_restore.install_system_prompt_contribution(Some(stale_source), None); - stale_restore - .materialize_durable_session_head() - .await - .unwrap(); + let history_before = unavailable_restore.history().to_vec(); + let (source, loads) = subject_resident_source_for_restore( + &subject_manifest, + availability, + Some("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH"), + ); + let report = + unavailable_restore.install_features(FeatureRegistryBuilder::new().with_module( + crate::feature::builtin::memory::SubjektivResidentRestoreRefreshFeature::new( + source, + unavailable_restore.prompts(), + "workspace-test", + ), + )); + assert!(!report.has_errors()); + unavailable_restore + .materialize_durable_session_head() + .await + .unwrap(); - assert_eq!(stale_loads.load(Ordering::SeqCst), 1); - assert_eq!( - &stale_restore.history()[..stale_history_before.len()], - stale_history_before - ); - let tombstone = stale_restore - .history() - .last() - .unwrap() - .as_text() - .unwrap() - .to_string(); - assert!(tombstone.contains("No current ready resident memory surface")); - assert!(!tombstone.contains("latest subject surface")); - let entries = store.read_all(session_id, segment_id).unwrap(); - assert!(matches!( - entries.last(), - Some(LogEntry::AnnotatedSystemItem { - entry: session_store::LoggedSystemHistoryEntry { - item: SystemItem::ResidentSummaryRefresh { .. }, + assert_eq!(loads.load(Ordering::SeqCst), 1); + assert_eq!( + &unavailable_restore.history()[..history_before.len()], + history_before + ); + let tombstone = unavailable_restore + .history() + .last() + .unwrap() + .as_text() + .unwrap() + .to_string(); + assert!(tombstone.contains("No current ready resident memory surface")); + assert!(!tombstone.contains("latest subject surface")); + assert!(!tombstone.contains("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH")); + let entries = store.read_all(session_id, segment_id).unwrap(); + assert!(matches!( + entries.last(), + Some(LogEntry::AnnotatedSystemItem { + entry: session_store::LoggedSystemHistoryEntry { + item: SystemItem::ResidentSummaryRefresh { .. }, + .. + }, .. - }, - .. - }) - )); + }) + )); + } } #[tokio::test] diff --git a/docs/design/subjektiv-surface.md b/docs/design/subjektiv-surface.md index 76f957018..9a4ecf828 100644 --- a/docs/design/subjektiv-surface.md +++ b/docs/design/subjektiv-surface.md @@ -42,14 +42,24 @@ and exact references remain immutable history, while current state points at the applicable result. Runtime-owned new Workers install Features before durably materializing the -initial Session head, so `ready` content or the explicit `ready-empty` result is -represented at initial prompt construction. Restored Workers preserve their -persisted prompt and all prior history verbatim. They append one durable -restore-boundary system item that supersedes earlier resident summaries with the -latest `ready` surface, an explicit `ready-empty` state, or a -no-current-surface tombstone for `ungenerated`, `stale`, or `failed`; subsequent -turns in that restored process do not append the boundary again. No regeneration, -correction, failure, or cutover rewrites a committed Session entry. +initial Session head. Initial resident prompt construction uses the host's shared +prompt-contribution source contract, so an enabled Feature can contribute its +current surface without owning Session persistence. Restore refresh is different: +it is a subjektiv Feature lifecycle hook installed only for an execution-enabled, +subject-attached Worker. The generic Worker host announces the restore boundary +and durably commits typed system items requested by installed Features; it does +not infer refresh behavior from the presence of a resident source. Consequently, +legacy Memory-only Workers, policy-only unattached Workers, and lifecycle-suppressed +Internal Workers/Reviewers do not read a subjektiv surface, render the restore +template, or append a refresh. + +The subjektiv restore hook preserves the persisted prompt and all prior history +verbatim. It requests one durable restore-boundary system item that supersedes +earlier resident summaries with the latest `ready` surface, an explicit +`ready-empty` state, or a no-current-surface tombstone for `ungenerated`, `stale`, +or `failed`; subsequent turns in that restored process do not append the boundary +again. No regeneration, correction, failure, or cutover rewrites a committed +Session entry. ## Semantic fixture and verification