From dfe837655a6d2e8c0cfd5b234b956925f2b30d20 Mon Sep 17 00:00:00 2001 From: Hare Date: Fri, 2 Oct 2026 15:01:09 +0900 Subject: [PATCH] fix: preserve resident surface startup states --- crates/memory/src/backend.rs | 18 ++++ crates/worker-runtime/src/worker_backend.rs | 4 - crates/worker/src/controller.rs | 35 ++++++++ crates/worker/src/feature/builtin/memory.rs | 50 +++++------ crates/worker/src/prompt/catalog.rs | 6 +- crates/worker/src/worker.rs | 74 ++++++++++++++-- crates/workspace-server/src/memory_backend.rs | 19 +++-- crates/workspace-server/src/server.rs | 85 +++++++++++++++---- docs/design/subjektiv-surface.md | 2 +- .../resident_memory_restore_section.md | 6 +- 10 files changed, 230 insertions(+), 69 deletions(-) diff --git a/crates/memory/src/backend.rs b/crates/memory/src/backend.rs index 8b7b8a546..c8186bf97 100644 --- a/crates/memory/src/backend.rs +++ b/crates/memory/src/backend.rs @@ -41,6 +41,7 @@ pub enum MemoryBackendHttpResponse { #[serde(tag = "kind", rename_all = "snake_case")] pub enum MemoryBackendOperationResult { ToolOutput(MemoryToolOutput), + ResidentSummary(MemoryResidentSummaryOutput), Acknowledged(MemoryBackendAckOutput), StagingWritten(MemoryStagingWriteOutput), } @@ -52,6 +53,23 @@ pub struct MemoryToolOutput { pub content: Option, } +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, JsonSchema)] +#[serde(rename_all = "snake_case")] +pub enum MemoryResidentSummaryAvailability { + Ready, + Ungenerated, + Stale, + Failed, + Unavailable, +} + +#[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] +pub struct MemoryResidentSummaryOutput { + pub availability: MemoryResidentSummaryAvailability, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub content: Option, +} + #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] pub struct MemoryQueryOperation { #[serde(default, skip_serializing_if = "Option::is_none")] diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index c1ffeca2d..c4bf58844 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -1266,10 +1266,6 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { } } - worker - .materialize_durable_session_head() - .await - .map_err(|error| format!("materialize durable Worker session head: {error}"))?; let workspace_client = worker.workspace_client_handle(); let started = prepared.start().await.map_err(|error| match error { WorkerBootstrapError::Worker(source) => { diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 0597a926d..4a0d99bbe 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -767,6 +767,17 @@ impl WorkerController { // the actual registered set instead of a hand-maintained mirror. worker.engine().tool_server_handle().flush_pending(); + // Runtime-owned fresh Workers persist their initial model-visible head + // before controller exposure. Feature installation above must happen first + // so the resident contribution is loaded into that durable prompt. Restored + // Workers retain their established first-turn refresh lifecycle. + if runtime_managed && worker.needs_initial_session_head_materialization() { + worker + .materialize_durable_session_head() + .await + .map_err(|error| std::io::Error::other(error.to_string()))?; + } + // === 4. Initial runtime files + WorkerSharedState + WorkerHandle + // SocketServer === let manifest_toml = toml::to_string_pretty(worker.manifest()).unwrap_or_default(); @@ -3465,6 +3476,30 @@ mod tests { use tempfile::TempDir; use tokio::net::UnixListener; + #[test] + fn runtime_session_head_is_materialized_after_feature_installation_before_exposure() { + let production = include_str!("controller.rs") + .split_once("#[cfg(test)]\nmod tests") + .map(|(production, _)| production) + .expect("controller test module marker"); + let startup = production + .split_once("async fn spawn_initialized") + .map(|(_, startup)| startup) + .expect("controller startup implementation"); + let feature_install = startup + .find("let fs_for_view = register_worker_tools(") + .expect("feature installation call"); + let materialize = startup + .find(".materialize_durable_session_head()") + .expect("runtime session-head materialization"); + let exposure = startup + .find("let handle = WorkerHandle") + .expect("controller handle exposure"); + + assert!(feature_install < materialize); + assert!(materialize < exposure); + } + #[test] fn automatic_memory_extraction_destinations_are_mutually_exclusive() { let mut memory = manifest::MemoryFeatureProfileConfig::default(); diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index 3aa2f50db..85f8936da 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -23,8 +23,8 @@ use crate::feature::{ ToolDeclaration, }; use crate::worker::{ - SystemPromptContributionSource, WorkspaceClient, WorkspaceClientError, WorkspaceRequest, - WorkspaceRequestMethod, + SystemPromptContribution, SystemPromptContributionSource, WorkspaceClient, + WorkspaceClientError, WorkspaceRequest, WorkspaceRequestMethod, }; #[derive(Clone, Debug)] @@ -652,7 +652,7 @@ struct WorkspaceResidentSummarySource { #[async_trait] impl SystemPromptContributionSource for WorkspaceResidentSummarySource { - async fn load(&self) -> Option { + async fn load(&self) -> SystemPromptContribution { match self .client .execute_memory_backend_operation( @@ -662,14 +662,22 @@ impl SystemPromptContributionSource for WorkspaceResidentSummarySource { ) .await { - Ok(memory::backend::MemoryBackendOperationResult::ToolOutput(output)) => output.content, + Ok(memory::backend::MemoryBackendOperationResult::ResidentSummary(output)) + if output.availability + == memory::backend::MemoryResidentSummaryAvailability::Ready => + { + SystemPromptContribution::Ready(output.content.unwrap_or_default()) + } + Ok(memory::backend::MemoryBackendOperationResult::ResidentSummary(_)) => { + SystemPromptContribution::Unavailable + } Ok(other) => { tracing::debug!(?other, "unexpected resident Memory Backend result"); - None + SystemPromptContribution::Unavailable } Err(error) => { tracing::debug!(%error, "resident Memory summary unavailable"); - None + SystemPromptContribution::Unavailable } } } @@ -843,8 +851,8 @@ mod tests { let body = serde_json::json!({ "status": "ok", "result": { - "kind": "tool_output", - "summary": "resident Memory summary collected", + "kind": "resident_summary", + "availability": "ready", "content": content, } }) @@ -976,12 +984,8 @@ mod tests { .unwrap() .unwrap(); assert_eq!( - plan.resident_summary_source - .unwrap() - .load() - .await - .as_deref(), - Some("# Durable Memory") + plan.resident_summary_source.unwrap().load().await, + SystemPromptContribution::Ready("# Durable Memory".into()) ); } @@ -1016,22 +1020,12 @@ mod tests { .unwrap(); assert_eq!( - first - .resident_summary_source - .unwrap() - .load() - .await - .as_deref(), - Some("first resident summary") + first.resident_summary_source.unwrap().load().await, + SystemPromptContribution::Ready("first resident summary".into()) ); assert_eq!( - restored - .resident_summary_source - .unwrap() - .load() - .await - .as_deref(), - Some("updated resident summary") + restored.resident_summary_source.unwrap().load().await, + SystemPromptContribution::Ready("updated resident summary".into()) ); } diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index b76ff37e5..f8c52cc47 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -507,11 +507,15 @@ impl PromptCatalog { } pub fn resident_memory_restore_section( &self, + surface_ready: bool, summary: Option<&str>, ) -> Result { self.render( WorkerPrompt::ResidentMemoryRestoreSection, - single("summary", summary.unwrap_or_default()), + minijinja::context! { + surface_ready => surface_ready, + summary => summary.unwrap_or_default(), + }, ) } pub fn worker_orchestration_guidance_section(&self) -> Result { diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 9df0ff6d2..3b24407a5 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -2336,9 +2336,15 @@ impl WorkerSession { } } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) enum SystemPromptContribution { + Ready(String), + Unavailable, +} + #[async_trait::async_trait] pub(crate) trait SystemPromptContributionSource: Send + Sync { - async fn load(&self) -> Option; + async fn load(&self) -> SystemPromptContribution; } #[derive(Debug, Clone)] @@ -3835,13 +3841,17 @@ impl Worker { self.restore_resident_summary_refresh_pending = false; return Ok(()); }; - let summary = if self.inject_resident_summary { + let contribution = if self.inject_resident_summary { source.load().await } else { - None + 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(summary.as_deref())?; + 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 { @@ -3925,7 +3935,14 @@ impl Worker { } let resident_summary = if self.inject_resident_summary { match &self.feature_resident_summary_source { - Some(source) => source.load().await, + Some(source) => match source.load().await { + SystemPromptContribution::Ready(summary) if !summary.is_empty() => { + Some(summary) + } + SystemPromptContribution::Ready(_) | SystemPromptContribution::Unavailable => { + None + } + }, None => None, } } else { @@ -3981,6 +3998,10 @@ impl Worker { .is_some_and(|state| state.pre_run_eligible(self.total_tokens().tokens)) } + pub(crate) fn needs_initial_session_head_materialization(&self) -> bool { + self.system_prompt_template.is_some() + } + /// Materialize the initial durable session head without running the model. /// /// Runtime-managed Workers call this before creation is accepted so even a @@ -8770,13 +8791,13 @@ mod build_summary_prompt_tests { use super::*; struct TestSystemPromptContributionSource { - value: Option, + value: SystemPromptContribution, load_count: Arc, } #[async_trait::async_trait] impl SystemPromptContributionSource for TestSystemPromptContributionSource { - async fn load(&self) -> Option { + async fn load(&self) -> SystemPromptContribution { self.load_count.fetch_add(1, Ordering::SeqCst); self.value.clone() } @@ -8788,7 +8809,10 @@ mod build_summary_prompt_tests { let load_count = Arc::new(AtomicUsize::new(0)); ( Arc::new(TestSystemPromptContributionSource { - value, + value: value.map_or( + SystemPromptContribution::Unavailable, + SystemPromptContribution::Ready, + ), load_count: Arc::clone(&load_count), }), load_count, @@ -11246,6 +11270,40 @@ permission = "write" assert_eq!(restored.history().len(), refreshed_len); drop(restored); + let mut empty_restore = + Worker::, _>::restore_from_manifest_with_context_and_model_client( + session_id, + segment_id, + manifest.clone(), + store.clone(), + crate::prompt::source::PromptCatalogSource::builtins_only(), + WorkerWorkspaceContext::no_workspace(), + authority.clone(), + Some(Box::new(NoopClient)), + ) + .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); + empty_restore + .materialize_durable_session_head() + .await + .unwrap(); + assert_eq!(empty_loads.load(Ordering::SeqCst), 1); + let empty_boundary = empty_restore + .history() + .last() + .unwrap() + .as_text() + .unwrap() + .to_string(); + assert!( + empty_boundary.contains("current ready resident memory surface is intentionally empty") + ); + 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( session_id, diff --git a/crates/workspace-server/src/memory_backend.rs b/crates/workspace-server/src/memory_backend.rs index a242c099f..394755b6a 100644 --- a/crates/workspace-server/src/memory_backend.rs +++ b/crates/workspace-server/src/memory_backend.rs @@ -1,8 +1,9 @@ use memory::backend::{ MemoryBackendAckOutput, MemoryBackendOperation, MemoryBackendOperationResult, MemoryDocumentReadOperation, MemoryDocumentUpdateOperation, MemoryQueryOperation, - MemoryStagingCloseAction, MemoryStagingCloseOperation, MemoryStagingListOperation, - MemoryStagingReadOperation, MemoryStagingWriteOutput, MemoryToolOutput, + MemoryResidentSummaryAvailability, MemoryResidentSummaryOutput, MemoryStagingCloseAction, + MemoryStagingCloseOperation, MemoryStagingListOperation, MemoryStagingReadOperation, + MemoryStagingWriteOutput, MemoryToolOutput, }; use memory::extract::{ExtractedCandidate, StagingEvidence, StagingRecord}; use memory::schema::{SourceEvidenceRef, SourceRef}; @@ -32,7 +33,7 @@ pub fn execute_memory_backend_operation_with_authority( .map(MemoryBackendOperationResult::ToolOutput) } MemoryBackendOperation::ResidentSummary(_operation) => Ok( - MemoryBackendOperationResult::ToolOutput(execute_resident_summary(authority)?), + MemoryBackendOperationResult::ResidentSummary(execute_resident_summary(authority)?), ), MemoryBackendOperation::AppendAudit(_operation) => Ok( MemoryBackendOperationResult::Acknowledged(MemoryBackendAckOutput { @@ -88,16 +89,18 @@ pub fn execute_memory_backend_operation_with_authority( } } -fn execute_resident_summary(authority: &A) -> Result { +fn execute_resident_summary( + authority: &A, +) -> Result { let document = authority.ensure_memory_document()?; if document.body_md.trim().is_empty() { - Ok(MemoryToolOutput { - summary: "resident memory summary unavailable".to_string(), + Ok(MemoryResidentSummaryOutput { + availability: MemoryResidentSummaryAvailability::Unavailable, content: None, }) } else { - Ok(MemoryToolOutput { - summary: "resident memory document collected".to_string(), + Ok(MemoryResidentSummaryOutput { + availability: MemoryResidentSummaryAvailability::Ready, content: Some(document.body_md), }) } diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 6fea2c5e5..3314c3798 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -21,7 +21,7 @@ use flow::{FlowSourceKind, FlowSourceResolveRequest, ResolvedFlowSource}; use futures::{SinkExt, StreamExt}; use memory::backend::{ MemoryBackendHttpResponse, MemoryBackendOperation, MemoryBackendOperationResult, - MemoryConsolidateStagingOperation, MemoryConsolidationOutput, MemoryToolOutput, + MemoryConsolidateStagingOperation, MemoryConsolidationOutput, }; use protocol::Segment; use protocol::stream::{decode_method, encode_event}; @@ -16590,34 +16590,39 @@ fn execute_scoped_resident_summary( .map_err(|error| Error::Store(error.to_string()))? .resident_surface(subject_id) .map_err(|error| Error::Store(error.to_string()))?; - let (summary, content) = match resident.availability { + Ok(MemoryBackendOperationResult::ResidentSummary( + resident_summary_output(resident), + )) +} + +fn resident_summary_output( + resident: crate::subjektiv::ResidentSurface, +) -> memory::backend::MemoryResidentSummaryOutput { + let (availability, content) = match resident.availability { crate::subjektiv::SurfaceAvailability::Ready => { let snapshot = resident.snapshot.expect("ready surface has a snapshot"); - if snapshot.body_md.is_empty() { - ("subject resident Memory surface is normally empty", None) - } else { - ( - "subject resident Memory surface collected", - Some(snapshot.body_md), - ) - } - } - crate::subjektiv::SurfaceAvailability::Ungenerated => { - ("subject resident Memory surface is not generated", None) + ( + memory::backend::MemoryResidentSummaryAvailability::Ready, + (!snapshot.body_md.is_empty()).then_some(snapshot.body_md), + ) } + crate::subjektiv::SurfaceAvailability::Ungenerated => ( + memory::backend::MemoryResidentSummaryAvailability::Ungenerated, + None, + ), crate::subjektiv::SurfaceAvailability::Stale => ( - "subject resident Memory surface is stale and was omitted", + memory::backend::MemoryResidentSummaryAvailability::Stale, None, ), crate::subjektiv::SurfaceAvailability::Failed => ( - "subject resident Memory surface generation failed and was omitted", + memory::backend::MemoryResidentSummaryAvailability::Failed, None, ), }; - Ok(MemoryBackendOperationResult::ToolOutput(MemoryToolOutput { - summary: summary.into(), + memory::backend::MemoryResidentSummaryOutput { + availability, content, - })) + } } const SUBJEKTIV_SINGLETON_PREFIX: &str = "subjektiv:"; @@ -30146,6 +30151,50 @@ mod tests { bytes } + #[test] + fn resident_summary_transport_preserves_ready_empty_state() { + let output = resident_summary_output(crate::subjektiv::ResidentSurface { + availability: crate::subjektiv::SurfaceAvailability::Ready, + snapshot: Some(crate::subjektiv::SurfaceSnapshot { + schema_version: 1, + id: "surface-empty".into(), + subject_id: "subject-a".into(), + body_md: String::new(), + memory_refs: Vec::new(), + built_from_store_revision: 7, + created_at: "2026-10-02T00:00:00Z".into(), + }), + }); + + assert_eq!( + output.availability, + memory::backend::MemoryResidentSummaryAvailability::Ready + ); + assert_eq!(output.content, None); + + for (surface, expected) in [ + ( + crate::subjektiv::SurfaceAvailability::Ungenerated, + memory::backend::MemoryResidentSummaryAvailability::Ungenerated, + ), + ( + crate::subjektiv::SurfaceAvailability::Stale, + memory::backend::MemoryResidentSummaryAvailability::Stale, + ), + ( + crate::subjektiv::SurfaceAvailability::Failed, + memory::backend::MemoryResidentSummaryAvailability::Failed, + ), + ] { + let output = resident_summary_output(crate::subjektiv::ResidentSurface { + availability: surface, + snapshot: None, + }); + assert_eq!(output.availability, expected); + assert_eq!(output.content, None); + } + } + #[test] fn external_permission_levels_map_to_existing_bits_and_reject_legacy_partial_grants() { let bits = |capabilities| { diff --git a/docs/design/subjektiv-surface.md b/docs/design/subjektiv-surface.md index ffabf0638..c76050af8 100644 --- a/docs/design/subjektiv-surface.md +++ b/docs/design/subjektiv-surface.md @@ -22,7 +22,7 @@ Every editor point carries one or more exact `{memory_id, revision}` references. Publication is idempotent for identical output at one store revision. A different concurrent output cannot replace the first published surface. Confirmed Memory writes atomically mark the prior surface stale. Failed surface work never rolls back confirmed Memory or candidate disposition. -Resident injection loads only a `ready` snapshot built from the subject's exact current store revision. `un-generated`, `failed`, and `stale` states are omitted rather than represented as "no Memory". A successful empty generation is represented by a current ready snapshot with empty Markdown. Snapshots created before this generation policy existed are retained as immutable history but migrate as `stale` and must be regenerated before injection. New Workers materialize the current surface in their initial prompt. Restored Workers preserve that persisted prompt and all prior history verbatim, then append one durable restore-boundary system item that supersedes earlier resident summaries with the latest ready surface or an explicit no-current-surface tombstone; subsequent turns in the same restored process do not append it again. +Resident injection loads only a `ready` snapshot built from the subject's exact current store revision. The resident Backend contract transports typed `ready`, `un-generated`, `failed`, and `stale` availability separately from optional Markdown, so a valid ready-empty snapshot cannot be conflated with omission or request failure. `un-generated`, `failed`, and `stale` states are omitted rather than represented as "no Memory". A successful empty generation is represented by a current ready snapshot with empty Markdown. Snapshots created before this generation policy existed are retained as immutable history but migrate as `stale` and must be regenerated before injection. Runtime-owned new Workers install Features before durably materializing the initial session head, ensuring the current contribution is present in their initial prompt. Restored Workers preserve that persisted prompt and all prior history verbatim, then 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; subsequent turns in the same restored process do not append it again. ## Semantic fixture and verification diff --git a/resources/prompts/internal/resident_memory_restore_section.md b/resources/prompts/internal/resident_memory_restore_section.md index 307c20c7b..fc8a22907 100644 --- a/resources/prompts/internal/resident_memory_restore_section.md +++ b/resources/prompts/internal/resident_memory_restore_section.md @@ -4,10 +4,14 @@ This restore-boundary record supersedes every resident memory summary embedded earlier in the system prompt or session history. Treat only the current state below as resident context; it is background context, not a user request. -{% if summary %} +{% if surface_ready %} ### Current ready surface +{% if summary %} {{ summary }} {% else %} +The current ready resident memory surface is intentionally empty. +{% endif %} +{% else %} No current ready resident memory surface is available. Ignore earlier resident memory summaries; confirmed Memory remains available through explicit recall tools. {% endif %}