fix: keep subjektiv restore refresh feature-owned

This commit is contained in:
2026-10-03 23:37:23 +09:00
parent 00c7e223b1
commit eea873e7dc
6 changed files with 679 additions and 130 deletions
+18 -3
View File
@@ -787,9 +787,9 @@ impl WorkerController {
// Runtime-owned Workers persist their current model-visible head before // Runtime-owned Workers persist their current model-visible head before
// controller exposure. Feature installation above must happen first so a // controller exposure. Feature installation above must happen first so a
// fresh subject Worker captures the current resident surface in its durable // fresh subject Worker captures the current resident surface in its durable
// prompt. Restored Workers defer their one-shot append-only resident refresh // prompt. Restored Workers defer generic Feature restore notification until
// until the next model-visible context materialization, after the Runtime has // the next model-visible context materialization; only an installed
// committed restoration of the existing durable head. // Feature may then request a typed durable history append.
if runtime_managed && worker.needs_initial_session_head_materialization() { if runtime_managed && worker.needs_initial_session_head_materialization() {
worker worker
.materialize_durable_session_head() .materialize_durable_session_head()
@@ -1342,6 +1342,21 @@ where
worker.workspace_client_handle(), 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)); feature_prompt_contribution = Some((Some(resident_summary_source), None));
} }
if ordinary_subjektiv_features_enabled if ordinary_subjektiv_features_enabled
+22 -1
View File
@@ -25,7 +25,8 @@ use thiserror::Error;
use crate::hook::{ use crate::hook::{
BeforeSessionRewrite, Hook, HookExecutionPolicy, HookRegistryBuilder, OnPromptSubmit, BeforeSessionRewrite, Hook, HookExecutionPolicy, HookRegistryBuilder, OnPromptSubmit,
OnTurnEnd, PostToolCall, PreLlmRequest, PreToolCall, RunCommitted, RunExit, WorkerStopping, OnTurnEnd, PostToolCall, PreLlmRequest, PreToolCall, RunCommitted, RunExit, WorkerRestored,
WorkerStopping,
}; };
use background::{ use background::{
BackgroundTaskSpec, FeatureBackgroundTask, FeatureBackgroundTaskRegistry, BackgroundTaskSpec, FeatureBackgroundTask, FeatureBackgroundTaskRegistry,
@@ -268,6 +269,7 @@ pub enum FeatureHookPoint {
RunExit, RunExit,
RunCommitted, RunCommitted,
BeforeSessionRewrite, BeforeSessionRewrite,
WorkerRestored,
WorkerStopping, WorkerStopping,
} }
@@ -1214,6 +1216,25 @@ impl HookContributionRegistrar<'_> {
Ok(()) Ok(())
} }
pub fn add_worker_restored(
&mut self,
name: impl Into<String>,
policy: HookExecutionPolicy,
hook: impl Hook<WorkerRestored> + '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( pub fn add_worker_stopping(
&mut self, &mut self,
name: impl Into<String>, name: impl Into<String>,
+150 -2
View File
@@ -7,6 +7,7 @@
use std::sync::Arc; use std::sync::Arc;
use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput}; use agen::tool::{Tool, ToolDefinition, ToolError, ToolExecutionContext, ToolMeta, ToolOutput};
use arc_swap::ArcSwap;
use async_trait::async_trait; use async_trait::async_trait;
use memory::backend::{ use memory::backend::{
MemoryBackendHttpResponse, MemoryBackendOperation, MemoryBackendOperationResult, MemoryBackendHttpResponse, MemoryBackendOperation, MemoryBackendOperationResult,
@@ -19,9 +20,11 @@ use serde::de::DeserializeOwned;
use serde_json::json; use serde_json::json;
use crate::feature::{ use crate::feature::{
FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution, FeatureDescriptor, FeatureHookPoint, FeatureInstallContext, FeatureInstallError, FeatureModule,
ToolDeclaration, HookDeclaration, ToolContribution, ToolDeclaration,
}; };
use crate::hook::{Hook, HookError, HookErrorCategory, HookExecutionPolicy, WorkerRestored};
use crate::prompt::catalog::{PromptCatalog, WorkerPrompt};
use crate::worker::{ use crate::worker::{
SystemPromptContribution, SystemPromptContributionSource, WorkspaceClient, SystemPromptContribution, SystemPromptContributionSource, WorkspaceClient,
WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod, 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<dyn SystemPromptContributionSource>,
prompts: Arc<ArcSwap<PromptCatalog>>,
workspace_id: String,
}
impl SubjektivResidentRestoreRefreshFeature {
pub(crate) fn for_host(
lifecycle_enabled: bool,
source: Arc<dyn SystemPromptContributionSource>,
prompts: Arc<ArcSwap<PromptCatalog>>,
workspace_id: impl Into<String>,
) -> Option<Self> {
lifecycle_enabled.then(|| Self::new(source, prompts, workspace_id))
}
pub(crate) fn new(
source: Arc<dyn SystemPromptContributionSource>,
prompts: Arc<ArcSwap<PromptCatalog>>,
workspace_id: impl Into<String>,
) -> 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<dyn SystemPromptContributionSource>,
prompts: Arc<ArcSwap<PromptCatalog>>,
workspace_id: String,
}
#[async_trait]
impl Hook<WorkerRestored> 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) struct MemoryFeatureInstallPlan {
pub(crate) module: MemoryToolsFeature, pub(crate) module: MemoryToolsFeature,
pub(crate) resident_summary_source: Option<Arc<dyn SystemPromptContributionSource>>, pub(crate) resident_summary_source: Option<Arc<dyn SystemPromptContributionSource>>,
@@ -1463,10 +1562,59 @@ permission = "write"
.is_none() .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"); let foreign = subject_manifest("other-workspace");
assert!(ordinary_subjektiv_resident_summary_source(&foreign, test_client()).is_err()); 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] #[test]
fn memory_feature_owns_normal_and_staging_tool_surfaces() { fn memory_feature_owns_normal_and_staging_tool_surfaces() {
let normal = MemoryToolsFeature::new(test_client(), false); let normal = MemoryToolsFeature::new(test_client(), false);
+115 -3
View File
@@ -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<Mutex<Vec<SystemItem>>>,
}
impl RestoreSystemItemAppendHandle {
pub(crate) fn new(pending: Arc<Mutex<Vec<SystemItem>>>) -> 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<String>,
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) // Hook input summary/context types (read-only)
// ============================================================================= // =============================================================================
@@ -524,9 +552,34 @@ pub struct WorkerStoppingContext {
pub reason: String, 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 RunExit;
pub struct RunCommitted; pub struct RunCommitted;
pub struct BeforeSessionRewrite; pub struct BeforeSessionRewrite;
pub struct WorkerRestored;
pub struct WorkerStopping; pub struct WorkerStopping;
impl HookEventKind for RunExit { impl HookEventKind for RunExit {
@@ -544,6 +597,11 @@ impl HookEventKind for BeforeSessionRewrite {
type Output = BeforeSessionRewriteAction; type Output = BeforeSessionRewriteAction;
} }
impl HookEventKind for WorkerRestored {
type Input = WorkerRestoredContext;
type Output = ();
}
impl HookEventKind for WorkerStopping { impl HookEventKind for WorkerStopping {
type Input = WorkerStoppingContext; type Input = WorkerStoppingContext;
type Output = (); type Output = ();
@@ -609,6 +667,7 @@ pub struct HookRegistryBuilder {
run_exit: Vec<RegisteredHook<RunExit>>, run_exit: Vec<RegisteredHook<RunExit>>,
run_committed: Vec<RegisteredHook<RunCommitted>>, run_committed: Vec<RegisteredHook<RunCommitted>>,
before_session_rewrite: Vec<RegisteredHook<BeforeSessionRewrite>>, before_session_rewrite: Vec<RegisteredHook<BeforeSessionRewrite>>,
worker_restored: Vec<RegisteredHook<WorkerRestored>>,
worker_stopping: Vec<RegisteredHook<WorkerStopping>>, worker_stopping: Vec<RegisteredHook<WorkerStopping>>,
} }
@@ -684,6 +743,12 @@ impl HookRegistryBuilder {
before_session_rewrite, before_session_rewrite,
BeforeSessionRewrite BeforeSessionRewrite
); );
add_hook_methods!(
add_worker_restored,
add_named_worker_restored,
worker_restored,
WorkerRestored
);
add_hook_methods!( add_hook_methods!(
add_worker_stopping, add_worker_stopping,
add_named_worker_stopping, add_named_worker_stopping,
@@ -691,7 +756,7 @@ impl HookRegistryBuilder {
WorkerStopping WorkerStopping
); );
pub(crate) fn checkpoint(&self) -> [usize; 9] { pub(crate) fn checkpoint(&self) -> [usize; 10] {
[ [
self.on_prompt_submit.len(), self.on_prompt_submit.len(),
self.pre_llm_request.len(), self.pre_llm_request.len(),
@@ -701,11 +766,12 @@ impl HookRegistryBuilder {
self.run_exit.len(), self.run_exit.len(),
self.run_committed.len(), self.run_committed.len(),
self.before_session_rewrite.len(), self.before_session_rewrite.len(),
self.worker_restored.len(),
self.worker_stopping.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.on_prompt_submit.truncate(checkpoint[0]);
self.pre_llm_request.truncate(checkpoint[1]); self.pre_llm_request.truncate(checkpoint[1]);
self.pre_tool_call.truncate(checkpoint[2]); self.pre_tool_call.truncate(checkpoint[2]);
@@ -714,7 +780,8 @@ impl HookRegistryBuilder {
self.run_exit.truncate(checkpoint[5]); self.run_exit.truncate(checkpoint[5]);
self.run_committed.truncate(checkpoint[6]); self.run_committed.truncate(checkpoint[6]);
self.before_session_rewrite.truncate(checkpoint[7]); 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 { pub fn build(self) -> HookRegistry {
@@ -727,6 +794,7 @@ impl HookRegistryBuilder {
run_exit: self.run_exit, run_exit: self.run_exit,
run_committed: self.run_committed, run_committed: self.run_committed,
before_session_rewrite: self.before_session_rewrite, before_session_rewrite: self.before_session_rewrite,
worker_restored: self.worker_restored,
worker_stopping: self.worker_stopping, worker_stopping: self.worker_stopping,
diagnostics: std::sync::Mutex::new(Vec::new()), diagnostics: std::sync::Mutex::new(Vec::new()),
} }
@@ -743,6 +811,7 @@ pub struct HookRegistry {
run_exit: Vec<RegisteredHook<RunExit>>, run_exit: Vec<RegisteredHook<RunExit>>,
run_committed: Vec<RegisteredHook<RunCommitted>>, run_committed: Vec<RegisteredHook<RunCommitted>>,
before_session_rewrite: Vec<RegisteredHook<BeforeSessionRewrite>>, before_session_rewrite: Vec<RegisteredHook<BeforeSessionRewrite>>,
worker_restored: Vec<RegisteredHook<WorkerRestored>>,
worker_stopping: Vec<RegisteredHook<WorkerStopping>>, worker_stopping: Vec<RegisteredHook<WorkerStopping>>,
diagnostics: std::sync::Mutex<Vec<HookExecutionError>>, diagnostics: std::sync::Mutex<Vec<HookExecutionError>>,
} }
@@ -840,6 +909,26 @@ impl HookRegistry {
.unwrap_or(BeforeSessionRewriteAction::Continue)) .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( pub async fn on_worker_stopping(
&self, &self,
context: &WorkerStoppingContext, 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] #[test]
fn pre_request_context_exposes_handle_only_when_host_supplies_one() { fn pre_request_context_exposes_handle_only_when_host_supplies_one() {
let info = PreRequestInfo { let info = PreRequestInfo {
+356 -113
View File
@@ -62,8 +62,9 @@ use crate::feature::{
use crate::hook::{ use crate::hook::{
BeforeSessionRewriteAction, BeforeSessionRewriteContext, Hook, HookHistoryRange, BeforeSessionRewriteAction, BeforeSessionRewriteContext, Hook, HookHistoryRange,
HookInvocationContext, HookRegistry, HookRegistryBuilder, OnPromptSubmit, OnTurnEnd, HookInvocationContext, HookRegistry, HookRegistryBuilder, OnPromptSubmit, OnTurnEnd,
PostToolCall, PreLlmRequest, PreToolCall, RunCommittedContext, RunCommittedExit, PostToolCall, PreLlmRequest, PreToolCall, RestoreSystemItemAppendHandle, RunCommittedContext,
RunExitContext, SessionRewriteKind, WorkerStoppingContext, RunCommittedExit, RunExitContext, SessionRewriteKind, WorkerRestoredContext,
WorkerStoppingContext,
}; };
use crate::in_flight::InFlightEvents; use crate::in_flight::InFlightEvents;
use crate::internal_worker::{ use crate::internal_worker::{
@@ -411,9 +412,7 @@ use crate::ipc::alerter::Alerter;
use crate::ipc::interceptor::WorkerInterceptor; use crate::ipc::interceptor::WorkerInterceptor;
use crate::ipc::notify_buffer::NotifyBuffer; use crate::ipc::notify_buffer::NotifyBuffer;
use crate::prompt::agents_md::read_agents_md; use crate::prompt::agents_md::read_agents_md;
use crate::prompt::catalog::{ use crate::prompt::catalog::{CatalogError, PromptCatalog, WorkspacePromptProjection};
CatalogError, PromptCatalog, WorkerPrompt, WorkspacePromptProjection,
};
use crate::prompt::source::PromptCatalogSource; use crate::prompt::source::PromptCatalogSource;
use crate::prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTemplate}; use crate::prompt::system::{SystemPromptContext, SystemPromptError, SystemPromptTemplate};
use crate::runtime::dir; use crate::runtime::dir;
@@ -2513,8 +2512,8 @@ pub struct Worker<C: LlmClient, St: Store> {
inject_resident_summary: bool, inject_resident_summary: bool,
/// Deferred resident prompt source installed by an enabled Feature. /// Deferred resident prompt source installed by an enabled Feature.
feature_resident_summary_source: Option<Arc<dyn SystemPromptContributionSource>>, feature_resident_summary_source: Option<Arc<dyn SystemPromptContributionSource>>,
/// One-shot append-only resident refresh required at a restore boundary. /// One-shot generic Feature lifecycle notification required after restore.
restore_resident_summary_refresh_pending: bool, restore_feature_lifecycle_pending: bool,
/// Complete system prompt replacement installed by an enabled Feature. /// Complete system prompt replacement installed by an enabled Feature.
feature_system_prompt_override: Option<String>, feature_system_prompt_override: Option<String>,
/// Typed user submissions in submit order. K-th entry corresponds to /// Typed user submissions in submit order. K-th entry corresponds to
@@ -2768,7 +2767,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
prompts, prompts,
inject_resident_summary: true, inject_resident_summary: true,
feature_resident_summary_source: None, feature_resident_summary_source: None,
restore_resident_summary_refresh_pending: false, restore_feature_lifecycle_pending: false,
feature_system_prompt_override: None, feature_system_prompt_override: None,
user_segments: Vec::new(), user_segments: Vec::new(),
sink: SegmentLogSink::new(), sink: SegmentLogSink::new(),
@@ -3843,63 +3842,71 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
} }
} }
async fn ensure_restore_resident_summary_refreshed(&mut self) -> Result<(), WorkerError> async fn notify_features_of_restore(&mut self) -> Result<(), WorkerError>
where where
St: Clone + 'static, St: Clone + 'static,
{ {
if !self.restore_resident_summary_refresh_pending { if !self.restore_feature_lifecycle_pending {
return Ok(()); return Ok(());
} }
let Some(source) = self.feature_resident_summary_source.clone() else { self.ensure_interceptor_installed();
self.restore_resident_summary_refresh_pending = false; let pending_system_items = Arc::new(Mutex::new(Vec::new()));
return Ok(()); if let Some(hooks) = self.hook_registry.clone() {
}; let context = WorkerRestoredContext::new(
let contribution = if self.inject_resident_summary { self.hook_invocation_context(None),
source.load().await RestoreSystemItemAppendHandle::new(Arc::clone(&pending_system_items)),
} else { );
SystemPromptContribution::Unavailable tokio::time::timeout(
}; FEATURE_HOOK_CHAIN_TIMEOUT,
let (surface_ready, summary) = match &contribution { hooks.on_worker_restored(&context),
SystemPromptContribution::Ready(summary) => (true, Some(summary.as_str())), )
SystemPromptContribution::Unavailable => (false, None), .await
}; .map_err(|_| {
let prompt_catalog = self.prompts.load_full(); hooks.record_chain_timeout("worker-restored");
let body = prompt_catalog.resident_memory_restore_section(surface_ready, summary)?; WorkerError::FeatureLifecycle(
let prompt_provenance = "worker-restored hook chain exceeded its host deadline".to_string(),
Some(self.prompt_render_provenance(WorkerPrompt::ResidentMemoryRestoreSection.key())); )
let item = SystemItem::ResidentSummaryRefresh { })?
body: body.clone(), .map_err(|error| WorkerError::FeatureLifecycle(error.to_string()))?;
prompt_provenance, }
};
let metadata = new_history_metadata( let system_items = std::mem::take(
WorkerHistoryProvenance::BackendInstruction { operation_id: None }, &mut *pending_system_items
None, .lock()
.expect("restore system-item queue poisoned"),
); );
self.commit_entry(LogEntry::AnnotatedSystemItem { for item in system_items {
ts: segment_log::now_millis(), let history_item = item.to_history_item();
entry: session_store::LoggedSystemHistoryEntry { let metadata = new_history_metadata(
item, WorkerHistoryProvenance::BackendInstruction { operation_id: None },
metadata: metadata.clone(), None,
}, );
extensions: Vec::new(), self.commit_entry(LogEntry::AnnotatedSystemItem {
})?; ts: segment_log::now_millis(),
let history_entry = HistoryEntry::new(Item::system_message(body), metadata); entry: session_store::LoggedSystemHistoryEntry {
let mut annotate = history_annotator( item,
self.log_writer_handle(), metadata: metadata.clone(),
vec![history_entry.clone()], },
self.pending_committed_history.clone(), extensions: Vec::new(),
); })?;
let (engine, session) = ( let history_entry = HistoryEntry::new(history_item, metadata);
self.engine.as_mut().expect("worker present"), let mut annotate = history_annotator(
&mut self.session, self.log_writer_handle(),
); vec![history_entry.clone()],
engine.append_history_with( self.pending_committed_history.clone(),
session.history_mut(), );
std::iter::once(history_entry.item), let (engine, session) = (
&mut annotate, self.engine.as_mut().expect("worker present"),
)?; &mut self.session,
session.note_mutation(); );
self.restore_resident_summary_refresh_pending = false; 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(()) Ok(())
} }
@@ -3915,7 +3922,7 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
St: Clone + 'static, St: Clone + 'static,
{ {
let Some(template) = self.system_prompt_template.take() else { 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() { if let Some(rendered) = self.feature_system_prompt_override.take() {
self.engine self.engine
@@ -6653,7 +6660,7 @@ where
prompts: common.prompts, prompts: common.prompts,
inject_resident_summary: true, inject_resident_summary: true,
feature_resident_summary_source: None, feature_resident_summary_source: None,
restore_resident_summary_refresh_pending: false, restore_feature_lifecycle_pending: false,
feature_system_prompt_override: None, feature_system_prompt_override: None,
user_segments: Vec::new(), user_segments: Vec::new(),
sink: SegmentLogSink::new(), sink: SegmentLogSink::new(),
@@ -6743,7 +6750,7 @@ where
prompts: common.prompts, prompts: common.prompts,
inject_resident_summary: true, inject_resident_summary: true,
feature_resident_summary_source: None, feature_resident_summary_source: None,
restore_resident_summary_refresh_pending: false, restore_feature_lifecycle_pending: false,
feature_system_prompt_override: None, feature_system_prompt_override: None,
user_segments: Vec::new(), user_segments: Vec::new(),
sink: SegmentLogSink::new(), sink: SegmentLogSink::new(),
@@ -6868,7 +6875,7 @@ where
prompts: common.prompts, prompts: common.prompts,
inject_resident_summary: true, inject_resident_summary: true,
feature_resident_summary_source: None, feature_resident_summary_source: None,
restore_resident_summary_refresh_pending: false, restore_feature_lifecycle_pending: false,
feature_system_prompt_override: None, feature_system_prompt_override: None,
user_segments: Vec::new(), user_segments: Vec::new(),
sink: SegmentLogSink::new(), sink: SegmentLogSink::new(),
@@ -7250,7 +7257,7 @@ where
prompts: common.prompts, prompts: common.prompts,
inject_resident_summary: true, inject_resident_summary: true,
feature_resident_summary_source: None, feature_resident_summary_source: None,
restore_resident_summary_refresh_pending: true, restore_feature_lifecycle_pending: true,
feature_system_prompt_override: None, feature_system_prompt_override: None,
user_segments: state.user_segments, user_segments: state.user_segments,
// Seed the mirror with the entries we just replayed so a // 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<AtomicUsize>,
}
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<WorkspaceResponse, WorkspaceClientError> {
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<dyn WorkspaceClient>, Arc<AtomicUsize>) {
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<String>,
load_count: Arc<AtomicUsize>,
}
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<WorkspaceResponse, WorkspaceClientError> {
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<dyn SystemPromptContributionSource>, Arc<AtomicUsize>) {
let load_count = Arc::new(AtomicUsize::new(0));
let client: Arc<dyn WorkspaceClient> = 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 { fn test_summary_input(items: &[Item]) -> String {
build_summary_input( build_summary_input(
items, items,
@@ -11199,7 +11330,7 @@ mod build_summary_prompt_tests {
session_store::FsStore::new(dir.path().join("sessions")).unwrap(), session_store::FsStore::new(dir.path().join("sessions")).unwrap(),
session_store::FsWorkerStore::new(dir.path().join("workers")).unwrap(), session_store::FsWorkerStore::new(dir.path().join("workers")).unwrap(),
); );
let manifest = WorkerManifest::from_toml(&format!( let mut manifest = WorkerManifest::from_toml(&format!(
r#" r#"
[worker] [worker]
name = "restore-resident-{}" name = "restore-resident-{}"
@@ -11219,6 +11350,16 @@ permission = "write"
cwd.display() cwd.display()
)) ))
.unwrap(); .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 authority = WorkerFilesystemAuthority::local(cwd.clone(), cwd.clone());
let scope = Scope::writable(&cwd).unwrap(); let scope = Scope::writable(&cwd).unwrap();
let mut original = Worker::new( let mut original = Worker::new(
@@ -11238,10 +11379,22 @@ permission = "write"
) )
.unwrap(), .unwrap(),
); );
let (old_source, _) = let (legacy_client, initial_legacy_loads) =
test_system_prompt_contribution_source(Some("old subject surface".into())); legacy_resident_client("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH");
original.install_system_prompt_contribution(Some(old_source), None); 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(); original.materialize_durable_session_head().await.unwrap();
assert_eq!(initial_legacy_loads.load(Ordering::SeqCst), 1);
let session_id = original.session_id(); let session_id = original.session_id();
let segment_id = original.segment_id(); let segment_id = original.segment_id();
assert!( assert!(
@@ -11249,11 +11402,14 @@ permission = "write"
.engine() .engine()
.get_system_prompt() .get_system_prompt()
.unwrap() .unwrap()
.contains("old subject surface") .contains("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH")
); );
drop(original); 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::<Box<dyn LlmClient>, _>::restore_from_manifest_with_context_and_model_client( Worker::<Box<dyn LlmClient>, _>::restore_from_manifest_with_context_and_model_client(
session_id, session_id,
segment_id, segment_id,
@@ -11266,10 +11422,69 @@ permission = "write"
) )
.await .await
.unwrap(); .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::<Box<dyn LlmClient>, _>::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 history_before = restored.history().to_vec();
let (current_source, current_loads) = let (current_source, current_loads) = subject_resident_source_for_restore(
test_system_prompt_contribution_source(Some("latest subject surface".into())); &subject_manifest,
restored.install_system_prompt_contribution(Some(current_source), None); 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(); restored.materialize_durable_session_head().await.unwrap();
assert_eq!(current_loads.load(Ordering::SeqCst), 1); assert_eq!(current_loads.load(Ordering::SeqCst), 1);
@@ -11283,7 +11498,7 @@ permission = "write"
.to_string(); .to_string();
assert!(refresh.contains("latest subject surface")); assert!(refresh.contains("latest subject surface"));
assert!(refresh.contains("supersedes every resident memory summary")); 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(); let refreshed_len = restored.history().len();
restored.materialize_durable_session_head().await.unwrap(); restored.materialize_durable_session_head().await.unwrap();
assert_eq!(restored.history().len(), refreshed_len); assert_eq!(restored.history().len(), refreshed_len);
@@ -11293,7 +11508,7 @@ permission = "write"
Worker::<Box<dyn LlmClient>, _>::restore_from_manifest_with_context_and_model_client( Worker::<Box<dyn LlmClient>, _>::restore_from_manifest_with_context_and_model_client(
session_id, session_id,
segment_id, segment_id,
manifest.clone(), subject_manifest.clone(),
store.clone(), store.clone(),
crate::prompt::source::PromptCatalogSource::builtins_only(), crate::prompt::source::PromptCatalogSource::builtins_only(),
WorkerWorkspaceContext::no_workspace(), WorkerWorkspaceContext::no_workspace(),
@@ -11302,9 +11517,19 @@ permission = "write"
) )
.await .await
.unwrap(); .unwrap();
let (empty_source, empty_loads) = let (empty_source, empty_loads) = subject_resident_source_for_restore(
test_system_prompt_contribution_source(Some(String::new())); &subject_manifest,
empty_restore.install_system_prompt_contribution(Some(empty_source), None); 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 empty_restore
.materialize_durable_session_head() .materialize_durable_session_head()
.await .await
@@ -11323,52 +11548,70 @@ permission = "write"
assert!(!empty_boundary.contains("No current ready resident memory surface")); assert!(!empty_boundary.contains("No current ready resident memory surface"));
drop(empty_restore); drop(empty_restore);
let mut stale_restore = for availability in [
Worker::<Box<dyn LlmClient>, _>::restore_from_manifest_with_context_and_model_client( memory::backend::MemoryResidentSummaryAvailability::Ungenerated,
memory::backend::MemoryResidentSummaryAvailability::Stale,
memory::backend::MemoryResidentSummaryAvailability::Failed,
] {
let mut unavailable_restore = Worker::<Box<dyn LlmClient>, _>::restore_from_manifest_with_context_and_model_client(
session_id, session_id,
segment_id, segment_id,
manifest, subject_manifest.clone(),
store.clone(), store.clone(),
crate::prompt::source::PromptCatalogSource::builtins_only(), crate::prompt::source::PromptCatalogSource::builtins_only(),
WorkerWorkspaceContext::no_workspace(), WorkerWorkspaceContext::no_workspace(),
authority, authority.clone(),
Some(Box::new(NoopClient)), Some(Box::new(NoopClient)),
) )
.await .await
.unwrap(); .unwrap();
let stale_history_before = stale_restore.history().to_vec(); let history_before = unavailable_restore.history().to_vec();
let (stale_source, stale_loads) = test_system_prompt_contribution_source(None); let (source, loads) = subject_resident_source_for_restore(
stale_restore.install_system_prompt_contribution(Some(stale_source), None); &subject_manifest,
stale_restore availability,
.materialize_durable_session_head() Some("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH"),
.await );
.unwrap(); 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!(loads.load(Ordering::SeqCst), 1);
assert_eq!( assert_eq!(
&stale_restore.history()[..stale_history_before.len()], &unavailable_restore.history()[..history_before.len()],
stale_history_before history_before
); );
let tombstone = stale_restore let tombstone = unavailable_restore
.history() .history()
.last() .last()
.unwrap() .unwrap()
.as_text() .as_text()
.unwrap() .unwrap()
.to_string(); .to_string();
assert!(tombstone.contains("No current ready resident memory surface")); assert!(tombstone.contains("No current ready resident memory surface"));
assert!(!tombstone.contains("latest subject surface")); assert!(!tombstone.contains("latest subject surface"));
let entries = store.read_all(session_id, segment_id).unwrap(); assert!(!tombstone.contains("LEGACY_MEMORY_SENTINEL_MUST_NOT_REFRESH"));
assert!(matches!( let entries = store.read_all(session_id, segment_id).unwrap();
entries.last(), assert!(matches!(
Some(LogEntry::AnnotatedSystemItem { entries.last(),
entry: session_store::LoggedSystemHistoryEntry { Some(LogEntry::AnnotatedSystemItem {
item: SystemItem::ResidentSummaryRefresh { .. }, entry: session_store::LoggedSystemHistoryEntry {
item: SystemItem::ResidentSummaryRefresh { .. },
..
},
.. ..
}, })
.. ));
}) }
));
} }
#[tokio::test] #[tokio::test]
+18 -8
View File
@@ -42,14 +42,24 @@ and exact references remain immutable history, while current state points at the
applicable result. applicable result.
Runtime-owned new Workers install Features before durably materializing the Runtime-owned new Workers install Features before durably materializing the
initial Session head, so `ready` content or the explicit `ready-empty` result is initial Session head. Initial resident prompt construction uses the host's shared
represented at initial prompt construction. Restored Workers preserve their prompt-contribution source contract, so an enabled Feature can contribute its
persisted prompt and all prior history verbatim. They append one durable current surface without owning Session persistence. Restore refresh is different:
restore-boundary system item that supersedes earlier resident summaries with the it is a subjektiv Feature lifecycle hook installed only for an execution-enabled,
latest `ready` surface, an explicit `ready-empty` state, or a subject-attached Worker. The generic Worker host announces the restore boundary
no-current-surface tombstone for `ungenerated`, `stale`, or `failed`; subsequent and durably commits typed system items requested by installed Features; it does
turns in that restored process do not append the boundary again. No regeneration, not infer refresh behavior from the presence of a resident source. Consequently,
correction, failure, or cutover rewrites a committed Session entry. 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 ## Semantic fixture and verification