diff --git a/crates/server-api/src/lib.rs b/crates/server-api/src/lib.rs index 089d7c65c..3ff4bca70 100644 --- a/crates/server-api/src/lib.rs +++ b/crates/server-api/src/lib.rs @@ -1079,6 +1079,7 @@ pub trait ServerApi { )] async fn memory_backend( &self, + #[extension] context: ServerRequestContext, #[path] workspace_id: String, #[body] request: MemoryBackendRequest, ) -> Result; @@ -4825,6 +4826,86 @@ pub struct SubjektivMemoryCandidateDecisionResponse { pub surface_dirty: bool, } +/// One bounded current Memory supplied to the surface editor. The Backend +/// selects these records deterministically; callers cannot name arbitrary +/// Memory revisions or broaden the generation input. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSurfaceMaterial { + pub memory_id: String, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_u64))] + pub revision: u64, + pub kind: memory::extract::CandidateKind, + pub body_md: String, + pub why_useful: String, + #[serde(default, skip_serializing_if = "Option::is_none")] + pub staleness: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSurfacePrepareRequest {} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSurfacePrepareResponse { + pub generation_id: String, + #[schemars(range(min = 0, max = 9_007_199_254_740_991_u64))] + pub store_revision: u64, + #[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))] + pub active_memory_count: usize, + pub materials: Vec, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_usize))] + pub body_token_budget: usize, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_usize))] + pub input_token_budget: usize, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_usize))] + pub per_kind_limit: usize, + #[schemars(range(min = 1, max = 9_007_199_254_740_991_usize))] + pub total_material_limit: usize, +} + +/// One model-edited Markdown point and its exact confirmed-Memory support. +/// Existence and scope are mechanically validated; semantic correctness is +/// deliberately not claimed by that validation. +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSurfacePoint { + pub body_md: String, + pub memory_refs: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSurfacePublishRequest { + pub generation_id: String, + pub points: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSurfaceFailureRequest { + pub generation_id: String, + pub reason_code: String, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSurfacePublishResponse { + pub snapshot_id: String, + #[schemars(range(min = 0, max = 9_007_199_254_740_991_u64))] + pub built_from_store_revision: u64, + pub empty: bool, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq, JsonSchema)] +#[serde(deny_unknown_fields)] +pub struct SubjektivSurfaceFailureResponse { + #[schemars(range(min = 0, max = 9_007_199_254_740_991_u64))] + pub store_revision: u64, + pub status: String, +} + #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] #[serde(tag = "operation", content = "input", rename_all = "snake_case")] pub enum SubjektivMemoryBackendOperation { @@ -4837,6 +4918,9 @@ pub enum SubjektivMemoryBackendOperation { ListCandidates(SubjektivMemoryCandidateListRequest), ReadCandidate(SubjektivMemoryCandidateReadRequest), DecideCandidate(SubjektivMemoryCandidateDecisionRequest), + PrepareSurface(SubjektivSurfacePrepareRequest), + PublishSurface(SubjektivSurfacePublishRequest), + FailSurface(SubjektivSurfaceFailureRequest), } #[derive(Debug, Clone, Serialize, Deserialize, JsonSchema)] @@ -5005,6 +5089,9 @@ pub enum SubjektivMemoryBackendResponse { Candidates(SubjektivMemoryCandidateListResponse), Candidate(SubjektivMemoryCandidate), CandidateDecided(SubjektivMemoryCandidateDecisionResponse), + SurfacePrepared(SubjektivSurfacePrepareResponse), + SurfacePublished(SubjektivSurfacePublishResponse), + SurfaceFailed(SubjektivSurfaceFailureResponse), } pub const WORKSPACE_DELETION_MAX_OPERATION_ID_BYTES: usize = 128; diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index 0fee205bb..0597a926d 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -1313,6 +1313,16 @@ where { feature_registry.add_module(memory_lifecycle); } + if let Some(surface_lifecycle) = crate::feature::builtin::memory_surface_lifecycle::SubjektivSurfaceLifecycleFeature::from_manifest( + worker.manifest_lifecycle_features_enabled(), + worker.workspace_client_handle(), + spawner_manifest.clone(), + worker.llm_client_handle(), + prompts.clone(), + spawner_workspace_context.clone(), + )? { + feature_registry.add_module(surface_lifecycle); + } if let Some(subjektiv_lifecycle) = crate::feature::builtin::memory_lifecycle::SubjektivLifecycleFeature::from_resolved_config( worker.manifest_lifecycle_features_enabled(), diff --git a/crates/worker/src/feature/builtin.rs b/crates/worker/src/feature/builtin.rs index 2469afd82..3adbc5987 100644 --- a/crates/worker/src/feature/builtin.rs +++ b/crates/worker/src/feature/builtin.rs @@ -11,6 +11,8 @@ pub mod manage_worker; pub mod memory; pub(crate) mod memory_lifecycle; pub mod memory_staging_output; +pub(crate) mod memory_surface_lifecycle; +pub(crate) mod memory_surface_output; pub mod merge_request; pub mod objective; pub mod orchestration; diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index cafe6545d..3aa2f50db 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -93,6 +93,87 @@ impl dyn WorkspaceClient + '_ { ) -> Result { execute_memory_consolidation(self, operation, true).await } + + pub async fn prepare_subjektiv_memory_surface( + &self, + ) -> Result { + match execute_subjektiv_memory_operation( + self, + server_api::SubjektivMemoryBackendOperation::PrepareSurface( + server_api::SubjektivSurfacePrepareRequest {}, + ), + ) + .await? + { + server_api::SubjektivMemoryBackendResponse::SurfacePrepared(output) => Ok(output), + other => Err(WorkspaceMemoryBackendError::Backend(format!( + "unexpected surface preparation response: {other:?}" + ))), + } + } + + pub async fn publish_subjektiv_memory_surface( + &self, + input: server_api::SubjektivSurfacePublishRequest, + ) -> Result { + match execute_subjektiv_memory_operation( + self, + server_api::SubjektivMemoryBackendOperation::PublishSurface(input), + ) + .await? + { + server_api::SubjektivMemoryBackendResponse::SurfacePublished(output) => Ok(output), + other => Err(WorkspaceMemoryBackendError::Backend(format!( + "unexpected surface publication response: {other:?}" + ))), + } + } + + pub async fn fail_subjektiv_memory_surface( + &self, + input: server_api::SubjektivSurfaceFailureRequest, + ) -> Result { + match execute_subjektiv_memory_operation( + self, + server_api::SubjektivMemoryBackendOperation::FailSurface(input), + ) + .await? + { + server_api::SubjektivMemoryBackendResponse::SurfaceFailed(output) => Ok(output), + other => Err(WorkspaceMemoryBackendError::Backend(format!( + "unexpected surface failure response: {other:?}" + ))), + } + } +} + +async fn execute_subjektiv_memory_operation( + client: &dyn WorkspaceClient, + operation: server_api::SubjektivMemoryBackendOperation, +) -> Result { + let workspace_id = + client + .workspace_id() + .ok_or_else(|| WorkspaceMemoryBackendError::Unavailable { + reason: format!( + "workspace client kind `{}` has no workspace id", + client.kind() + ), + })?; + let response = client.execute(WorkspaceRequest::json( + WorkspaceRequestMethod::Post, + format!("/api/w/{workspace_id}/subjektiv/memory"), + serde_json::to_string(&server_api::SubjektivMemoryBackendRequest { operation })?, + ))?; + let status = reqwest::StatusCode::from_u16(response.status) + .unwrap_or(reqwest::StatusCode::INTERNAL_SERVER_ERROR); + if !response.is_success() { + return Err(WorkspaceMemoryBackendError::Http { + status, + body: response.body, + }); + } + Ok(serde_json::from_str(&response.body)?) } async fn execute_memory_backend( diff --git a/crates/worker/src/feature/builtin/memory_surface_lifecycle.rs b/crates/worker/src/feature/builtin/memory_surface_lifecycle.rs new file mode 100644 index 000000000..db7b3f9be --- /dev/null +++ b/crates/worker/src/feature/builtin/memory_surface_lifecycle.rs @@ -0,0 +1,288 @@ +use std::sync::Arc; +use std::time::Duration; + +use agen::llm_client::LlmClient; +use arc_swap::ArcSwap; +use async_trait::async_trait; + +use crate::PromptCatalog; +use crate::Scope; +use crate::feature::background::{ + BackgroundTaskCancellation, BackgroundTaskContext, BackgroundTaskSpec, BackgroundTaskTrigger, + FeatureBackgroundTask, +}; +use crate::feature::builtin::memory::WorkspaceMemoryBackendError; +use crate::feature::builtin::memory_surface_output::{ + MemorySurfaceOutputFeature, MemorySurfaceOutputState, +}; +use crate::feature::{ + BackgroundTaskDeclaration, FeatureDescriptor, FeatureInstallContext, FeatureInstallError, + FeatureModule, FeatureRegistryBuilder, +}; +use crate::hook::{HookError, HookErrorCategory}; +use crate::internal_worker::{ + InternalWorkerAuthority, InternalWorkerIdentity, InternalWorkerSpec, + run_internal_worker_with_cancel_sender, +}; +use crate::worker::{WorkerFilesystemAuthority, WorkerWorkspaceContext, WorkspaceClient}; +use manifest::WorkerManifest; + +const TASK_NAME: &str = "subjektiv-memory-surface"; +const TASK_TIMEOUT: Duration = Duration::from_secs(300); +const MAX_GENERATION_ATTEMPTS: usize = 2; +const EDITOR_MAX_TURNS: u32 = 3; + +#[derive(Clone)] +pub(crate) struct SubjektivSurfaceLifecycleFeature { + task: SubjektivSurfaceLifecycleTask, +} + +#[derive(Clone)] +struct SubjektivSurfaceLifecycleTask { + workspace_client: Arc, + manifest: WorkerManifest, + client: Box, + prompts: Arc>, + workspace_context: WorkerWorkspaceContext, +} + +impl SubjektivSurfaceLifecycleFeature { + pub(crate) fn from_manifest( + lifecycle_enabled: bool, + workspace_client: Arc, + manifest: WorkerManifest, + client: Box, + prompts: Arc>, + workspace_context: WorkerWorkspaceContext, + ) -> std::io::Result> { + let dedicated = manifest.profile.as_ref().is_some_and(|snapshot| { + matches!( + &snapshot.source, + manifest::ProfileSource::Registry { + source: manifest::ProfileRegistrySource::Builtin, + name, + .. + } if name == "subjektiv-memory-consolidation" + ) + }); + if !lifecycle_enabled || !dedicated { + return Ok(None); + } + if !workspace_client.is_available() || workspace_client.workspace_id().is_none() { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "subjektiv surface generation requires Backend Workspace API authority", + )); + } + Ok(Some(Self { + task: SubjektivSurfaceLifecycleTask { + workspace_client, + manifest, + client, + prompts, + workspace_context, + }, + })) + } +} + +impl FeatureModule for SubjektivSurfaceLifecycleFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("subjektiv-memory-surface-lifecycle", "subjektiv Memory Surface Lifecycle") + .with_description("Rebuilds one bounded subject Memory surface after each committed consolidation run.") + .with_background_task(BackgroundTaskDeclaration::worker_managed( + TASK_NAME, + "Generate and publish a current subject Memory surface from bounded confirmed revisions.", + )) + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + let declaration = BackgroundTaskDeclaration::worker_managed( + TASK_NAME, + "Generate and publish a current subject Memory surface from bounded confirmed revisions.", + ); + let mut spec = BackgroundTaskSpec::single_flight(declaration, TASK_TIMEOUT); + spec.trigger = BackgroundTaskTrigger::RunCommitted; + context.background_tasks().register(spec, self.task.clone()) + } +} + +#[async_trait] +impl FeatureBackgroundTask for SubjektivSurfaceLifecycleTask { + async fn run( + &self, + context: BackgroundTaskContext, + cancellation: BackgroundTaskCancellation, + ) -> Result<(), HookError> { + for attempt in 0..MAX_GENERATION_ATTEMPTS { + context.generation_fence.ensure_current()?; + if cancellation.is_cancelled() { + return Ok(()); + } + let generation = self + .workspace_client + .prepare_subjektiv_memory_surface() + .await + .map_err(surface_hook_error)?; + if generation.materials.is_empty() && generation.active_memory_count > 0 { + self.record_failure(&generation.generation_id, "input_budget_exhausted") + .await; + tracing::warn!( + active_memory_count = generation.active_memory_count, + "subject Memory surface materials could not fit the bounded input" + ); + return Ok(()); + } + let points = if generation.materials.is_empty() { + Vec::new() + } else { + match self.edit_surface(&generation, cancellation.clone()).await { + Ok(points) => points, + Err(error) => { + self.record_failure(&generation.generation_id, "editor_failed") + .await; + tracing::warn!(%error, "subject Memory surface editor failed"); + return Ok(()); + } + } + }; + context.generation_fence.ensure_current()?; + if cancellation.is_cancelled() { + return Ok(()); + } + let publish = self + .workspace_client + .publish_subjektiv_memory_surface(server_api::SubjektivSurfacePublishRequest { + generation_id: generation.generation_id.clone(), + points, + }) + .await; + match publish { + Ok(output) => { + tracing::debug!( + snapshot_id = output.snapshot_id, + store_revision = output.built_from_store_revision, + empty = output.empty, + "published subject Memory surface" + ); + return Ok(()); + } + Err(WorkspaceMemoryBackendError::Http { status, .. }) + if status == reqwest::StatusCode::CONFLICT + && attempt + 1 < MAX_GENERATION_ATTEMPTS => + { + // Confirmed Memory moved while editing. Discard the stale + // output and rebuild once from a fresh bounded generation. + continue; + } + Err(error) => { + self.record_failure(&generation.generation_id, "publish_failed") + .await; + tracing::warn!(%error, "subject Memory surface publication failed"); + return Ok(()); + } + } + } + Ok(()) + } +} + +impl SubjektivSurfaceLifecycleTask { + async fn edit_surface( + &self, + generation: &server_api::SubjektivSurfacePrepareResponse, + cancellation: BackgroundTaskCancellation, + ) -> Result, HookError> { + let language = self + .manifest + .feature + .memory + .workspace_settings() + .map(|settings| settings.language) + .ok_or_else(|| { + HookError::new( + HookErrorCategory::Internal, + "surface editor requires bound Workspace Memory settings", + ) + })?; + let system_prompt = self + .prompts + .load_full() + .subjektiv_memory_surface_system(&language) + .map_err(|error| HookError::new(HookErrorCategory::Internal, error.to_string()))?; + let input = serde_json::to_string_pretty(&serde_json::json!({ + "question": "この主体が次の作業を始める際、毎回思い出しておくべきことは何か。継続的な制約、判断の前提、未解決事項、再発を避けたい教訓を、指定予算内でまとめる。", + "body_token_budget": generation.body_token_budget, + "input_token_budget": generation.input_token_budget, + "materials": generation.materials, + })) + .map_err(|error| HookError::new(HookErrorCategory::Internal, error.to_string()))?; + let output_state = MemorySurfaceOutputState::new( + generation.materials.clone(), + generation.body_token_budget, + ); + let features = FeatureRegistryBuilder::new() + .with_module(MemorySurfaceOutputFeature::new(output_state.clone())); + let cancellation_observer = cancellation.clone(); + let cancel_observer = move |sender: tokio::sync::mpsc::Sender<()>| { + tokio::spawn(async move { + cancellation_observer.cancelled().await; + let _ = sender.send(()).await; + }); + }; + let result = run_internal_worker_with_cancel_sender( + InternalWorkerSpec { + identity: InternalWorkerIdentity { + kind: "subjektiv-memory-surface", + run_id: uuid::Uuid::now_v7(), + }, + manifest: self.manifest.clone(), + client: self.client.clone_boxed(), + system_prompt, + input, + cache_key: None, + max_turns: Some(EDITOR_MAX_TURNS), + engine_configurator: None, + features, + required_tools: &["SubmitMemorySurface"], + authority: InternalWorkerAuthority { + workspace: self.workspace_context.clone(), + filesystem: WorkerFilesystemAuthority::None, + scope: Scope::empty(), + workdir_session: None, + }, + }, + cancel_observer, + ) + .await; + if let Err(error) = result { + return Err(HookError::new( + HookErrorCategory::Internal, + format!("surface editor Internal Worker failed: {}", error.source), + )); + } + output_state.submitted().ok_or_else(|| { + HookError::new( + HookErrorCategory::Internal, + "surface editor finished without valid structured output", + ) + }) + } + + async fn record_failure(&self, generation_id: &str, reason_code: &str) { + if let Err(error) = self + .workspace_client + .fail_subjektiv_memory_surface(server_api::SubjektivSurfaceFailureRequest { + generation_id: generation_id.to_string(), + reason_code: reason_code.to_string(), + }) + .await + { + tracing::debug!(%error, "could not record subject Memory surface failure"); + } + } +} + +fn surface_hook_error(error: WorkspaceMemoryBackendError) -> HookError { + HookError::new(HookErrorCategory::Internal, error.to_string()) +} diff --git a/crates/worker/src/feature/builtin/memory_surface_output.rs b/crates/worker/src/feature/builtin/memory_surface_output.rs new file mode 100644 index 000000000..4b3233364 --- /dev/null +++ b/crates/worker/src/feature/builtin/memory_surface_output.rs @@ -0,0 +1,242 @@ +use std::collections::HashSet; +use std::sync::{Arc, Mutex}; + +use agen::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; +use async_trait::async_trait; +use schemars::JsonSchema; +use serde::Deserialize; + +use crate::feature::{ + FeatureDescriptor, FeatureInstallContext, FeatureInstallError, FeatureModule, ToolContribution, + ToolDeclaration, +}; + +const SUBMIT_TOOL: &str = "SubmitMemorySurface"; +const SUBMIT_DESCRIPTION: &str = "Submit a bounded Markdown Memory surface whose every point cites exact refs from the supplied confirmed-Memory materials. Reference validation proves existence and scope, not semantic correctness."; + +#[derive(Clone)] +pub(crate) struct MemorySurfaceOutputState { + materials: Arc>, + body_token_budget: usize, + submitted: Arc>>>, +} + +impl MemorySurfaceOutputState { + pub(crate) fn new( + materials: Vec, + body_token_budget: usize, + ) -> Self { + Self { + materials: Arc::new(materials), + body_token_budget, + submitted: Arc::new(Mutex::new(None)), + } + } + + pub(crate) fn submitted(&self) -> Option> { + self.submitted + .lock() + .expect("Memory surface output state poisoned") + .clone() + } + + fn validate(&self, points: &[server_api::SubjektivSurfacePoint]) -> Result<(), ToolError> { + if self.materials.is_empty() { + return if points.is_empty() { + Ok(()) + } else { + Err(ToolError::InvalidArgument( + "empty materials require an empty points array".into(), + )) + }; + } + if points.is_empty() { + return Err(ToolError::InvalidArgument( + "non-empty materials require at least one grounded point".into(), + )); + } + if points.len() > 24 { + return Err(ToolError::InvalidArgument(format!( + "points count {} exceeds the configured limit 24", + points.len() + ))); + } + let allowed = self + .materials + .iter() + .map(|material| (material.memory_id.as_str(), material.revision)) + .collect::>(); + let mut bodies = Vec::with_capacity(points.len()); + for point in points { + let body = point.body_md.trim(); + if body.is_empty() { + return Err(ToolError::InvalidArgument( + "surface point body_md must not be empty".into(), + )); + } + if point.memory_refs.is_empty() { + return Err(ToolError::InvalidArgument( + "every surface point requires at least one memory_ref".into(), + )); + } + let mut point_refs = HashSet::new(); + for reference in &point.memory_refs { + if !point_refs.insert((reference.memory_id.as_str(), reference.revision)) { + return Err(ToolError::InvalidArgument(format!( + "duplicate surface point ref {}@{}", + reference.memory_id, reference.revision + ))); + } + if !allowed.contains(&(reference.memory_id.as_str(), reference.revision)) { + return Err(ToolError::InvalidArgument(format!( + "surface ref {}@{} is not in the generation materials", + reference.memory_id, reference.revision + ))); + } + } + bodies.push(body); + } + let bytes = bodies + .iter() + .map(|body| body.len()) + .sum::() + .saturating_add(points.len().saturating_sub(1) * 2); + let tokens = bytes.saturating_add(3) / 4; + if tokens > self.body_token_budget { + return Err(ToolError::InvalidArgument(format!( + "surface body token estimate {tokens} exceeds budget {}", + self.body_token_budget + ))); + } + Ok(()) + } +} + +#[derive(Clone)] +pub(crate) struct MemorySurfaceOutputFeature { + state: MemorySurfaceOutputState, +} + +impl MemorySurfaceOutputFeature { + pub(crate) fn new(state: MemorySurfaceOutputState) -> Self { + Self { state } + } +} + +impl FeatureModule for MemorySurfaceOutputFeature { + fn descriptor(&self) -> FeatureDescriptor { + FeatureDescriptor::builtin("memory-surface-output", "Memory Surface Output") + .with_description( + "Restricted structured output for one clean-context Memory surface editor.", + ) + .with_tool(ToolDeclaration::new(SUBMIT_TOOL, SUBMIT_DESCRIPTION)) + } + + fn install(&self, context: &mut FeatureInstallContext<'_>) -> Result<(), FeatureInstallError> { + context.tools().register(ToolContribution::new( + SUBMIT_TOOL, + submit_definition(self.state.clone()), + )) + } +} + +fn submit_definition(state: MemorySurfaceOutputState) -> ToolDefinition { + Arc::new(move || { + let schema = serde_json::to_value(schemars::schema_for!(SubmitMemorySurfaceParams)) + .unwrap_or_else(|_| serde_json::json!({})); + let meta = ToolMeta::new(SUBMIT_TOOL) + .description(SUBMIT_DESCRIPTION) + .input_schema(schema); + let tool: Arc = Arc::new(SubmitMemorySurfaceTool { + state: state.clone(), + }); + (meta, tool) + }) +} + +#[derive(Debug, Deserialize, JsonSchema)] +#[serde(deny_unknown_fields)] +struct SubmitMemorySurfaceParams { + points: Vec, +} + +struct SubmitMemorySurfaceTool { + state: MemorySurfaceOutputState, +} + +#[async_trait] +impl Tool for SubmitMemorySurfaceTool { + async fn execute( + &self, + input_json: &str, + _context: agen::tool::ToolExecutionContext, + ) -> Result { + let params: SubmitMemorySurfaceParams = + serde_json::from_str(input_json).map_err(|error| { + ToolError::InvalidArgument(format!("invalid SubmitMemorySurface input: {error}")) + })?; + self.state.validate(¶ms.points)?; + let mut submitted = self + .state + .submitted + .lock() + .expect("Memory surface output state poisoned"); + if submitted.is_some() { + return Err(ToolError::InvalidArgument( + "SubmitMemorySurface already accepted output for this generation".into(), + )); + } + let count = params.points.len(); + *submitted = Some(params.points); + Ok(ToolOutput { + summary: format!("Accepted {count} grounded Memory surface point(s)."), + content: None, + attachments: Vec::new(), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use memory::extract::CandidateKind; + + fn material(id: &str) -> server_api::SubjektivSurfaceMaterial { + server_api::SubjektivSurfaceMaterial { + memory_id: id.into(), + revision: 1, + kind: CandidateKind::Constraint, + body_md: "Keep the boundary.".into(), + why_useful: "Avoid regression.".into(), + staleness: Some("Until the boundary changes.".into()), + } + } + + fn point(id: &str) -> server_api::SubjektivSurfacePoint { + server_api::SubjektivSurfacePoint { + body_md: "- Keep the boundary unless it changes.".into(), + memory_refs: vec![server_api::SubjektivMemoryRevisionRef { + memory_id: id.into(), + revision: 1, + }], + } + } + + #[test] + fn rejects_missing_or_material_external_refs_and_budget_overflow() { + let state = MemorySurfaceOutputState::new(vec![material("allowed")], 32); + assert!(state.validate(&[]).is_err()); + assert!(state.validate(&[point("other")]).is_err()); + let mut oversized = point("allowed"); + oversized.body_md = "x".repeat(132); + assert!(state.validate(&[oversized]).is_err()); + assert!(state.validate(&[point("allowed")]).is_ok()); + } + + #[test] + fn empty_materials_accept_only_normal_empty_output() { + let state = MemorySurfaceOutputState::new(Vec::new(), 32); + assert!(state.validate(&[]).is_ok()); + assert!(state.validate(&[point("invented")]).is_err()); + } +} diff --git a/crates/worker/src/prompt/catalog.rs b/crates/worker/src/prompt/catalog.rs index 13b32e874..685476d3a 100644 --- a/crates/worker/src/prompt/catalog.rs +++ b/crates/worker/src/prompt/catalog.rs @@ -95,6 +95,7 @@ pub enum WorkerPrompt { MemoryExtractSystem, MemoryConsolidationSystem, SubjektivMemoryConsolidationSystem, + SubjektivMemorySurfaceSystem, FlowVerifierSystem, BackendJobSystem, NotifyWrapper, @@ -116,6 +117,7 @@ impl WorkerPrompt { Self::SubjektivMemoryConsolidationSystem => { "internal.subjektiv_memory_consolidation_system" } + Self::SubjektivMemorySurfaceSystem => "internal.subjektiv_memory_surface_system", Self::FlowVerifierSystem => "internal.flow_verifier_system", Self::BackendJobSystem => "internal.backend_job_system", Self::NotifyWrapper => "internal.notify_wrapper", @@ -136,6 +138,7 @@ impl WorkerPrompt { WorkerPrompt::MemoryExtractSystem, WorkerPrompt::MemoryConsolidationSystem, WorkerPrompt::SubjektivMemoryConsolidationSystem, + WorkerPrompt::SubjektivMemorySurfaceSystem, WorkerPrompt::FlowVerifierSystem, WorkerPrompt::BackendJobSystem, WorkerPrompt::NotifyWrapper, @@ -463,6 +466,12 @@ impl PromptCatalog { single("language", language), ) } + pub fn subjektiv_memory_surface_system(&self, language: &str) -> Result { + self.render( + WorkerPrompt::SubjektivMemorySurfaceSystem, + single("language", language), + ) + } pub fn flow_verifier_system(&self) -> Result { self.render(WorkerPrompt::FlowVerifierSystem, Value::UNDEFINED) } @@ -927,6 +936,23 @@ mod tests { .unwrap() .contains("LANGUAGE_MARKER") ); + let surface = catalog + .subjektiv_memory_surface_system("SURFACE_LANGUAGE") + .unwrap(); + assert!(surface.contains("SURFACE_LANGUAGE")); + for required in [ + "conditions", + "negation", + "contradictions", + "staleness", + "SubmitMemorySurface", + "not semantic verification", + ] { + assert!( + surface.contains(required), + "missing surface rule {required}" + ); + } assert!( catalog .notify_wrapper("NOTIFICATION_MARKER") diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 83e337b35..6fea2c5e5 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -20,8 +20,8 @@ use config_source::ConfigTreeSnapshot; use flow::{FlowSourceKind, FlowSourceResolveRequest, ResolvedFlowSource}; use futures::{SinkExt, StreamExt}; use memory::backend::{ - MemoryBackendHttpResponse, MemoryBackendOperation, MemoryConsolidateStagingOperation, - MemoryConsolidationOutput, + MemoryBackendHttpResponse, MemoryBackendOperation, MemoryBackendOperationResult, + MemoryConsolidateStagingOperation, MemoryConsolidationOutput, MemoryToolOutput, }; use protocol::Segment; use protocol::stream::{decode_method, encode_event}; @@ -6998,6 +6998,7 @@ impl server_api::ServerApi for ServerApiContractService { async fn memory_backend( &self, + context: server_api::ServerRequestContext, workspace_id: String, request: server_api::MemoryBackendRequest, ) -> std::result::Result @@ -7005,6 +7006,7 @@ impl server_api::ServerApi for ServerApiContractService { scoped_memory_backend_operation( State(self.workspace_api()?.clone()), AxumPath(ScopedWorkspacePath { workspace_id }), + context, Json(request.0), ) .await @@ -16532,11 +16534,17 @@ async fn scoped_list_memory_staging( async fn scoped_memory_backend_operation( State(api): State, AxumPath(path): AxumPath, + context: server_api::ServerRequestContext, Json(operation): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; - let response = match execute_memory_backend_operation_with_authority(&api.authority, operation) - { + let result = match operation { + MemoryBackendOperation::ResidentSummary(_) => { + execute_scoped_resident_summary(&api, &context) + } + operation => execute_memory_backend_operation_with_authority(&api.authority, operation), + }; + let response = match result { Ok(result) => MemoryBackendHttpResponse::Ok { result }, Err(error) => MemoryBackendHttpResponse::Error { message: sanitize_backend_error(&error.to_string()), @@ -16545,6 +16553,73 @@ async fn scoped_memory_backend_operation( Ok(Json(response)) } +fn execute_scoped_resident_summary( + api: &WorkspaceApi, + context: &server_api::ServerRequestContext, +) -> crate::Result { + let Some(source) = context.runtime_source.as_ref() else { + return execute_memory_backend_operation_with_authority( + &api.authority, + MemoryBackendOperation::ResidentSummary(Default::default()), + ); + }; + let Some(worker_id) = source.worker_id.as_deref() else { + return execute_memory_backend_operation_with_authority( + &api.authority, + MemoryBackendOperation::ResidentSummary(Default::default()), + ); + }; + let worker = RuntimeWorkerRef::new(&source.runtime_id, worker_id); + let Some(lease) = api + .store + .require_current_worker_singleton_owner(&api.config.workspace_id, &worker)? + else { + return execute_memory_backend_operation_with_authority( + &api.authority, + MemoryBackendOperation::ResidentSummary(Default::default()), + ); + }; + let Some(subject_id) = lease.key.strip_prefix(SUBJEKTIV_SINGLETON_PREFIX) else { + return execute_memory_backend_operation_with_authority( + &api.authority, + MemoryBackendOperation::ResidentSummary(Default::default()), + ); + }; + let resident = + crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration) + .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 { + 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) + } + crate::subjektiv::SurfaceAvailability::Stale => ( + "subject resident Memory surface is stale and was omitted", + None, + ), + crate::subjektiv::SurfaceAvailability::Failed => ( + "subject resident Memory surface generation failed and was omitted", + None, + ), + }; + Ok(MemoryBackendOperationResult::ToolOutput(MemoryToolOutput { + summary: summary.into(), + content, + })) +} + const SUBJEKTIV_SINGLETON_PREFIX: &str = "subjektiv:"; const SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX: &str = crate::hosts::SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX; @@ -17069,10 +17144,97 @@ async fn scoped_subjektiv_memory_backend( subjektiv_candidate_decision_response(receipt), ) } + server_api::SubjektivMemoryBackendOperation::PrepareSurface(_) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Consolidation, + "surface generation preparation", + )?; + let generation = store + .prepare_surface_generation(&subject_id) + .map_err(subjektiv_store_error)?; + server_api::SubjektivMemoryBackendResponse::SurfacePrepared( + subjektiv_surface_prepare_response(generation), + ) + } + server_api::SubjektivMemoryBackendOperation::PublishSurface(input) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Consolidation, + "surface publication", + )?; + let points = input + .points + .into_iter() + .map(|point| crate::subjektiv::SurfacePoint { + body_md: point.body_md, + memory_refs: point + .memory_refs + .into_iter() + .map(|reference| crate::subjektiv::MemoryRevisionRef { + memory_id: reference.memory_id, + revision: reference.revision, + }) + .collect(), + }) + .collect(); + let snapshot = store + .publish_surface_generation(&subject_id, &input.generation_id, points) + .map_err(subjektiv_store_error)?; + server_api::SubjektivMemoryBackendResponse::SurfacePublished( + server_api::SubjektivSurfacePublishResponse { + snapshot_id: snapshot.id, + built_from_store_revision: snapshot.built_from_store_revision, + empty: snapshot.body_md.is_empty(), + }, + ) + } + server_api::SubjektivMemoryBackendOperation::FailSurface(input) => { + require_subjektiv_worker_authority( + authority, + SubjektivWorkerAuthority::Consolidation, + "surface generation failure", + )?; + let revision = store + .fail_surface_generation(&subject_id, &input.generation_id, &input.reason_code) + .map_err(subjektiv_store_error)?; + server_api::SubjektivMemoryBackendResponse::SurfaceFailed( + server_api::SubjektivSurfaceFailureResponse { + store_revision: revision, + status: "failed".into(), + }, + ) + } }; Ok(Json(response)) } +fn subjektiv_surface_prepare_response( + generation: crate::subjektiv::SurfaceGeneration, +) -> server_api::SubjektivSurfacePrepareResponse { + server_api::SubjektivSurfacePrepareResponse { + generation_id: generation.id, + store_revision: generation.store_revision, + active_memory_count: generation.active_memory_count, + materials: generation + .materials + .into_iter() + .map(|material| server_api::SubjektivSurfaceMaterial { + memory_id: material.memory_id, + revision: material.revision, + kind: material.kind, + body_md: material.body_md, + why_useful: material.why_useful, + staleness: material.staleness, + }) + .collect(), + body_token_budget: crate::subjektiv::SURFACE_BODY_TOKEN_BUDGET, + input_token_budget: crate::subjektiv::SURFACE_INPUT_TOKEN_BUDGET, + per_kind_limit: crate::subjektiv::SURFACE_PER_KIND_LIMIT, + total_material_limit: crate::subjektiv::SURFACE_TOTAL_MATERIAL_LIMIT, + } +} + fn subjektiv_candidate_summary( candidate: crate::subjektiv::SubjectStagingRecord, ) -> server_api::SubjektivMemoryCandidateSummary { @@ -18215,7 +18377,8 @@ fn subjektiv_store_error(error: crate::subjektiv::SubjektivError) -> Error { crate::subjektiv::SubjektivError::SubjectScopeMismatch { .. } => { Error::WorkspacePermissionDenied(format!("subject_scope_mismatch: {error}")) } - crate::subjektiv::SubjektivError::RevisionConflict { .. } => { + crate::subjektiv::SubjektivError::RevisionConflict { .. } + | crate::subjektiv::SubjektivError::SurfaceGenerationConflict(_) => { Error::RepositoryConflict(format!("revision_conflict: {error}")) } crate::subjektiv::SubjektivError::CandidateResolved(_) @@ -18253,28 +18416,91 @@ async fn scoped_subjektiv_memory_consolidation( )?)) } +const SUBJECT_CONSOLIDATION_THRESHOLD_FILES: usize = 5; +const SUBJECT_CONSOLIDATION_THRESHOLD_BYTES: u64 = 50_000; + +fn subject_consolidation_required( + candidate_count: usize, + total_bytes: u64, + force: bool, + surface_availability: crate::subjektiv::SurfaceAvailability, +) -> bool { + let surface_needs_generation = + surface_availability != crate::subjektiv::SurfaceAvailability::Ready; + if candidate_count == 0 { + return surface_needs_generation; + } + force + || surface_needs_generation + || candidate_count >= SUBJECT_CONSOLIDATION_THRESHOLD_FILES + || total_bytes >= SUBJECT_CONSOLIDATION_THRESHOLD_BYTES +} + +#[cfg(test)] +mod subject_surface_consolidation_policy_tests { + use super::*; + use crate::subjektiv::SurfaceAvailability; + + #[test] + fn failed_stale_and_ungenerated_surfaces_bypass_candidate_thresholds() { + for availability in [ + SurfaceAvailability::Failed, + SurfaceAvailability::Stale, + SurfaceAvailability::Ungenerated, + ] { + assert!(subject_consolidation_required(0, 0, false, availability)); + assert!(subject_consolidation_required(1, 1, false, availability)); + } + assert!(!subject_consolidation_required( + 0, + 0, + true, + SurfaceAvailability::Ready + )); + assert!(!subject_consolidation_required( + 1, + 1, + false, + SurfaceAvailability::Ready + )); + assert!(subject_consolidation_required( + SUBJECT_CONSOLIDATION_THRESHOLD_FILES, + 1, + false, + SurfaceAvailability::Ready + )); + } +} + fn start_subjektiv_staging_consolidation( api: WorkspaceApi, subject_id: &str, operation: MemoryConsolidateStagingOperation, ) -> ApiResult { - let (candidate_count, total_bytes) = open_subjektiv_store(&api)? + let store = open_subjektiv_store(&api)?; + let (candidate_count, total_bytes) = store .pending_staging_backlog(subject_id) .map_err(subjektiv_store_error)?; - if candidate_count == 0 { - return Ok(MemoryConsolidationOutput { - status: "skipped_empty".into(), - summary: "No subject Memory staging candidates are pending.".into(), - candidate_count, - total_bytes, - }); - } - const CONSOLIDATION_THRESHOLD_FILES: usize = 5; - const CONSOLIDATION_THRESHOLD_BYTES: u64 = 50_000; - if !operation.force - && candidate_count < CONSOLIDATION_THRESHOLD_FILES - && total_bytes < CONSOLIDATION_THRESHOLD_BYTES - { + let surface_availability = store + .resident_surface(subject_id) + .map_err(subjektiv_store_error)? + .availability; + if !subject_consolidation_required( + candidate_count, + total_bytes, + operation.force, + surface_availability, + ) { + if candidate_count == 0 { + return Ok(MemoryConsolidationOutput { + status: "skipped_empty".into(), + summary: + "No subject Memory staging candidates are pending and the surface is current." + .into(), + candidate_count, + total_bytes, + }); + } return Ok(MemoryConsolidationOutput { status: "skipped_below_threshold".into(), summary: format!( @@ -18288,7 +18514,8 @@ fn start_subjektiv_staging_consolidation( let runtime_id = select_memory_consolidation_runtime(&api)?; let singleton_key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}"); let input_content = format!( - "Process the delegated subject's {candidate_count} pending Memory candidate(s) ({total_bytes} bytes) with MemoryStagingList, MemoryStagingRead, subject Memory reads, and MemoryApplyCandidate." + "Process the delegated subject's {candidate_count} pending Memory candidate(s) ({total_bytes} bytes) with MemoryStagingList, MemoryStagingRead, subject Memory reads, and MemoryApplyCandidate. After this committed consolidation turn, the Host will rebuild the {:?} resident surface from confirmed Memory in a separate clean context.", + surface_availability ); if let Some(output) = try_reuse_subject_consolidation_worker( &api, diff --git a/crates/workspace-server/src/subjektiv.rs b/crates/workspace-server/src/subjektiv.rs index a5b92a4f9..3b5a8708a 100644 --- a/crates/workspace-server/src/subjektiv.rs +++ b/crates/workspace-server/src/subjektiv.rs @@ -23,6 +23,15 @@ use crate::feature_storage::{ pub const SUBJEKTIV_SCHEMA_VERSION: u32 = 1; pub const SUBJEKTIV_FEATURE_ID: &str = "subjektiv"; pub const MAX_STAGING_ANCHORS: usize = 10; +/// Deterministic surface-generation policy. Input estimates use the repository's +/// provider-independent UTF-8 byte estimate (`ceil(bytes / 4)`). The material +/// payload is capped below the whole-input budget to reserve room for the fixed +/// prompt and JSON framing. +pub const SURFACE_BODY_TOKEN_BUDGET: usize = 1_024; +pub const SURFACE_INPUT_TOKEN_BUDGET: usize = 12_000; +pub const SURFACE_MATERIAL_TOKEN_BUDGET: usize = 10_000; +pub const SURFACE_PER_KIND_LIMIT: usize = 8; +pub const SURFACE_TOTAL_MATERIAL_LIMIT: usize = 24; pub type Result = std::result::Result; @@ -62,6 +71,8 @@ pub enum SubjektivError { expected: u64, actual: u64, }, + #[error("surface generation conflict: {0}")] + SurfaceGenerationConflict(String), #[error("memory `{memory_id}` cannot transition from {from:?} to {to:?}")] InvalidStateTransition { memory_id: String, @@ -401,7 +412,7 @@ pub struct StagingResolution { pub resolved_at: String, } -#[derive(Debug, Clone, Serialize, Deserialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct SurfaceSnapshot { pub schema_version: u32, @@ -414,6 +425,49 @@ pub struct SurfaceSnapshot { pub created_at: String, } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SurfaceMaterial { + pub memory_id: String, + pub revision: u64, + pub kind: CandidateKind, + pub body_md: String, + pub why_useful: String, + pub staleness: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SurfaceGeneration { + pub id: String, + pub subject_id: String, + pub store_revision: u64, + pub active_memory_count: usize, + pub materials: Vec, + pub created_at: String, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct SurfacePoint { + pub body_md: String, + pub memory_refs: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum SurfaceAvailability { + Ungenerated, + Stale, + Failed, + Ready, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ResidentSurface { + pub availability: SurfaceAvailability, + pub snapshot: Option, +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum MemoryRevisionTarget { @@ -841,6 +895,75 @@ END; Ok(()) } +fn add_surface_generation_state( + transaction: &Transaction<'_>, +) -> crate::feature_storage::Result<()> { + transaction.execute_batch( + r#" +CREATE TABLE surface_generation_state ( + subject_id TEXT PRIMARY KEY, + store_revision INTEGER NOT NULL CHECK (store_revision >= 0), + status TEXT NOT NULL CHECK (status IN ('dirty', 'failed', 'ready')), + snapshot_id TEXT, + reason_code TEXT, + updated_at TEXT NOT NULL, + FOREIGN KEY (subject_id) REFERENCES subjects(subject_id) ON DELETE RESTRICT, + FOREIGN KEY (subject_id, snapshot_id) + REFERENCES surface_snapshots(subject_id, snapshot_id) ON DELETE RESTRICT +); + +INSERT INTO surface_generation_state ( + subject_id, store_revision, status, snapshot_id, reason_code, updated_at +) +SELECT subjects.subject_id, + subjects.store_revision, + CASE WHEN EXISTS ( + SELECT 1 FROM surface_snapshots current + WHERE current.subject_id = subjects.subject_id + AND current.built_from_store_revision = subjects.store_revision + ) THEN 'ready' ELSE 'dirty' END, + ( + SELECT current.snapshot_id FROM surface_snapshots current + WHERE current.subject_id = subjects.subject_id + AND current.built_from_store_revision = subjects.store_revision + ORDER BY current.created_at ASC, current.snapshot_id ASC + LIMIT 1 + ), + NULL, + subjects.updated_at +FROM subjects +WHERE EXISTS ( + SELECT 1 FROM surface_snapshots any_snapshot + WHERE any_snapshot.subject_id = subjects.subject_id +); + +CREATE TABLE surface_generation_runs ( + subject_id TEXT NOT NULL, + generation_id TEXT NOT NULL, + store_revision INTEGER NOT NULL CHECK (store_revision >= 0), + active_memory_count INTEGER NOT NULL CHECK (active_memory_count >= 0), + materials_json TEXT NOT NULL, + created_at TEXT NOT NULL, + PRIMARY KEY (subject_id, generation_id), + FOREIGN KEY (subject_id) REFERENCES subjects(subject_id) ON DELETE RESTRICT +); + +CREATE INDEX surface_generation_runs_by_revision +ON surface_generation_runs(subject_id, store_revision, created_at, generation_id); + +CREATE TRIGGER surface_generation_runs_no_update +BEFORE UPDATE ON surface_generation_runs BEGIN + SELECT RAISE(ABORT, 'subjektiv surface generation inputs are immutable'); +END; +CREATE TRIGGER surface_generation_runs_no_delete +BEFORE DELETE ON surface_generation_runs BEGIN + SELECT RAISE(ABORT, 'subjektiv surface generation inputs are retained'); +END; +"#, + )?; + Ok(()) +} + static MIGRATIONS: &[FeatureMigration] = &[ FeatureMigration::new(1, "create subjektiv subject memory store", create_schema), FeatureMigration::new( @@ -853,6 +976,11 @@ static MIGRATIONS: &[FeatureMigration] = &[ "add idempotent atomic candidate decision receipts", add_candidate_decision_receipts, ), + FeatureMigration::new( + 4, + "add bounded surface generation state and runs", + add_surface_generation_state, + ), ]; pub const REGISTRATION: FeatureRegistration = @@ -1519,6 +1647,225 @@ impl SubjektivStore { }) } + /// Captures one deterministic, bounded generation input from only current + /// active Memory revisions. Kind order is fixed and selection is round-robin + /// across kinds; within each kind, updated_at desc then memory id asc. + pub fn prepare_surface_generation(&self, subject_id: &str) -> Result { + self.database.try_transaction(|transaction| { + let subject = require_active_subject(transaction, subject_id)?; + let (active_memory_count, materials) = + select_surface_materials(transaction, subject_id)?; + let generation = SurfaceGeneration { + id: issued_id("surface-generation"), + subject_id: subject_id.to_string(), + store_revision: subject.store_revision, + active_memory_count, + materials, + created_at: now(), + }; + transaction.execute( + "INSERT INTO surface_generation_runs ( + subject_id, generation_id, store_revision, active_memory_count, + materials_json, created_at + ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)", + params![ + subject_id, + generation.id, + to_i64(generation.store_revision)?, + to_i64(generation.active_memory_count as u64)?, + serde_json::to_string(&generation.materials)?, + generation.created_at + ], + )?; + Ok(generation) + }) + } + + /// Validates structured editor output against the exact material snapshot and + /// atomically publishes it only while the subject store generation is current. + pub fn publish_surface_generation( + &self, + subject_id: &str, + generation_id: &str, + points: Vec, + ) -> Result { + validate_label("surface generation id", generation_id)?; + self.database.try_transaction(|transaction| { + let subject = require_active_subject(transaction, subject_id)?; + let (store_revision, active_memory_count, materials_json) = transaction + .query_row( + "SELECT store_revision, active_memory_count, materials_json + FROM surface_generation_runs + WHERE subject_id = ?1 AND generation_id = ?2", + params![subject_id, generation_id], + |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, i64>(1)?, + row.get::<_, String>(2)?, + )) + }, + ) + .optional()? + .ok_or_else(|| { + SubjektivError::SurfaceGenerationConflict(format!( + "unknown generation `{generation_id}` for subject `{subject_id}`" + )) + })?; + let store_revision = u64::try_from(store_revision).map_err(|_| { + SubjektivError::InvalidRecord("surface generation revision is negative".into()) + })?; + if subject.store_revision != store_revision { + return Err(SubjektivError::SurfaceGenerationConflict(format!( + "generation {store_revision} is stale; current store revision is {}", + subject.store_revision + ))); + } + let materials: Vec = serde_json::from_str(&materials_json)?; + if active_memory_count > 0 && materials.is_empty() { + return Err(SubjektivError::InvalidRecord( + "active Memory exists but no record fits the surface input budget".into(), + )); + } + let (body_md, memory_refs) = validate_surface_points(&materials, &points)?; + if let Some(existing) = + surface_snapshot_for_revision(transaction, subject_id, store_revision)? + { + if existing.body_md == body_md && existing.memory_refs == memory_refs { + return Ok(existing); + } + return Err(SubjektivError::SurfaceGenerationConflict(format!( + "store revision {store_revision} already has a different published surface" + ))); + } + insert_surface_snapshot( + transaction, + subject_id, + body_md, + memory_refs, + store_revision, + ) + }) + } + + /// Records a bounded generator failure without changing confirmed Memory. A + /// late failure cannot replace a ready surface or dirty a newer generation. + pub fn fail_surface_generation( + &self, + subject_id: &str, + generation_id: &str, + reason_code: &str, + ) -> Result { + validate_label("surface generation id", generation_id)?; + validate_label("surface failure reason", reason_code)?; + self.database.try_transaction(|transaction| { + let subject = require_active_subject(transaction, subject_id)?; + let generation_revision = transaction + .query_row( + "SELECT store_revision FROM surface_generation_runs + WHERE subject_id = ?1 AND generation_id = ?2", + params![subject_id, generation_id], + |row| row.get::<_, i64>(0), + ) + .optional()? + .ok_or_else(|| { + SubjektivError::SurfaceGenerationConflict(format!( + "unknown generation `{generation_id}` for subject `{subject_id}`" + )) + })?; + let generation_revision = u64::try_from(generation_revision).map_err(|_| { + SubjektivError::InvalidRecord("surface generation revision is negative".into()) + })?; + if generation_revision == subject.store_revision { + let ready = transaction + .query_row( + "SELECT status FROM surface_generation_state WHERE subject_id = ?1", + [subject_id], + |row| row.get::<_, String>(0), + ) + .optional()? + .is_some_and(|status| status == "ready"); + if !ready { + transaction.execute( + "INSERT INTO surface_generation_state ( + subject_id, store_revision, status, snapshot_id, reason_code, updated_at + ) VALUES (?1, ?2, 'failed', NULL, ?3, ?4) + ON CONFLICT(subject_id) DO UPDATE SET + store_revision = excluded.store_revision, + status = 'failed', snapshot_id = NULL, + reason_code = excluded.reason_code, + updated_at = excluded.updated_at", + params![subject_id, to_i64(generation_revision)?, reason_code, now()], + )?; + } + } + Ok(subject.store_revision) + }) + } + + /// Returns only a current ready snapshot for injection. Other states are + /// explicit so callers never report missing/stale/failed generation as empty + /// confirmed Memory. + pub fn resident_surface(&self, subject_id: &str) -> Result { + self.database.try_with_connection(|connection| { + let subject = require_subject_in_connection(connection, subject_id)?; + let state = connection + .query_row( + "SELECT store_revision, status, snapshot_id + FROM surface_generation_state WHERE subject_id = ?1", + [subject_id], + |row| { + Ok(( + row.get::<_, i64>(0)?, + row.get::<_, String>(1)?, + row.get::<_, Option>(2)?, + )) + }, + ) + .optional()?; + let Some((revision, status, snapshot_id)) = state else { + return Ok(ResidentSurface { + availability: SurfaceAvailability::Ungenerated, + snapshot: None, + }); + }; + let revision = u64::try_from(revision).map_err(|_| { + SubjektivError::InvalidRecord("surface state revision is negative".into()) + })?; + if revision != subject.store_revision || status == "dirty" { + return Ok(ResidentSurface { + availability: SurfaceAvailability::Stale, + snapshot: None, + }); + } + if status == "failed" { + return Ok(ResidentSurface { + availability: SurfaceAvailability::Failed, + snapshot: None, + }); + } + let snapshot_id = snapshot_id.ok_or_else(|| { + SubjektivError::InvalidRecord("ready surface state has no snapshot id".into()) + })?; + let raw = connection.query_row( + "SELECT snapshot_json FROM surface_snapshots + WHERE subject_id = ?1 AND snapshot_id = ?2", + params![subject_id, snapshot_id], + |row| row.get::<_, String>(0), + )?; + let snapshot = parse_surface_snapshot(&raw)?; + if snapshot.built_from_store_revision != subject.store_revision { + return Err(SubjektivError::InvalidRecord( + "ready surface snapshot generation does not match subject".into(), + )); + } + Ok(ResidentSurface { + availability: SurfaceAvailability::Ready, + snapshot: Some(snapshot), + }) + }) + } + pub fn create_surface_snapshot( &self, subject_id: &str, @@ -1528,16 +1875,7 @@ impl SubjektivStore { ) -> Result { reject_duplicate_refs("surface snapshot", &memory_refs)?; reject_duplicate_memory_ids("surface snapshot", &memory_refs)?; - let snapshot = SurfaceSnapshot { - schema_version: SUBJEKTIV_SCHEMA_VERSION, - id: issued_id("surface"), - subject_id: subject_id.to_string(), - body_md: body_md.into(), - memory_refs, - built_from_store_revision, - created_at: now(), - }; - let raw = serde_json::to_string(&snapshot)?; + let body_md = body_md.into(); self.database.try_transaction(|transaction| { let subject = require_active_subject(transaction, subject_id)?; if subject.store_revision != built_from_store_revision { @@ -1546,38 +1884,31 @@ impl SubjektivStore { subject.store_revision ))); } - validate_memory_refs(transaction, subject_id, &snapshot.memory_refs)?; - transaction.execute( - "INSERT INTO surface_snapshots ( - subject_id, snapshot_id, built_from_store_revision, snapshot_json, created_at - ) VALUES (?1, ?2, ?3, ?4, ?5)", - params![ - subject_id, - snapshot.id, - to_i64(snapshot.built_from_store_revision)?, - raw, - snapshot.created_at - ], - )?; - for reference in &snapshot.memory_refs { - transaction.execute( - "INSERT INTO surface_snapshot_refs ( - subject_id, snapshot_id, memory_id, revision - ) VALUES (?1, ?2, ?3, ?4)", - params![ - subject_id, - snapshot.id, - reference.memory_id, - to_i64(reference.revision)? - ], - )?; + validate_memory_refs(transaction, subject_id, &memory_refs)?; + if estimated_tokens(&body_md) > SURFACE_BODY_TOKEN_BUDGET { + return Err(SubjektivError::InvalidRecord(format!( + "surface snapshot body exceeds {SURFACE_BODY_TOKEN_BUDGET} token estimate" + ))); } - transaction.execute( - "INSERT INTO surface_snapshot_seals (subject_id, snapshot_id) - VALUES (?1, ?2)", - params![subject_id, snapshot.id], - )?; - Ok(snapshot) + if let Some(existing) = surface_snapshot_for_revision( + transaction, + subject_id, + built_from_store_revision, + )? { + if existing.body_md == body_md && existing.memory_refs == memory_refs { + return Ok(existing); + } + return Err(SubjektivError::SurfaceGenerationConflict(format!( + "store revision {built_from_store_revision} already has a different published surface" + ))); + } + insert_surface_snapshot( + transaction, + subject_id, + body_md, + memory_refs, + built_from_store_revision, + ) }) } @@ -1600,6 +1931,218 @@ impl SubjektivStore { } } +fn select_surface_materials( + transaction: &Transaction<'_>, + subject_id: &str, +) -> Result<(usize, Vec)> { + let mut statement = transaction.prepare( + "SELECT record_json FROM memory_records + WHERE subject_id = ?1 AND state = 'active' + ORDER BY updated_at DESC, memory_id ASC", + )?; + let rows = statement.query_map([subject_id], |row| row.get::<_, String>(0))?; + let mut buckets: [Vec; 6] = std::array::from_fn(|_| Vec::new()); + let mut active_memory_count = 0usize; + for row in rows { + let memory = parse_memory(&row?)?; + active_memory_count = active_memory_count.saturating_add(1); + let bucket = surface_kind_index(&memory.kind); + if buckets[bucket].len() < SURFACE_PER_KIND_LIMIT { + buckets[bucket].push(SurfaceMaterial { + memory_id: memory.id, + revision: memory.revision, + kind: memory.kind, + body_md: memory.body_md, + why_useful: memory.why_useful, + staleness: memory.staleness, + }); + } + } + + let mut selected = Vec::new(); + let mut payload_bytes = 2usize; // JSON array delimiters. + for offset in 0..SURFACE_PER_KIND_LIMIT { + for bucket in &buckets { + let Some(material) = bucket.get(offset) else { + continue; + }; + if selected.len() == SURFACE_TOTAL_MATERIAL_LIMIT { + return Ok((active_memory_count, selected)); + } + let material_bytes = serde_json::to_vec(material)?.len().saturating_add(1); + if estimated_tokens_for_bytes(payload_bytes.saturating_add(material_bytes)) + > SURFACE_MATERIAL_TOKEN_BUDGET + { + continue; + } + payload_bytes = payload_bytes.saturating_add(material_bytes); + selected.push(material.clone()); + } + } + Ok((active_memory_count, selected)) +} + +fn surface_kind_index(kind: &CandidateKind) -> usize { + match kind { + CandidateKind::Preference => 0, + CandidateKind::Constraint => 1, + CandidateKind::Decision => 2, + CandidateKind::WorkingAssumption => 3, + CandidateKind::OpenQuestion => 4, + CandidateKind::Lesson => 5, + } +} + +fn validate_surface_points( + materials: &[SurfaceMaterial], + points: &[SurfacePoint], +) -> Result<(String, Vec)> { + if materials.is_empty() { + if points.is_empty() { + return Ok((String::new(), Vec::new())); + } + return Err(SubjektivError::InvalidRecord( + "an empty material set requires an empty surface".into(), + )); + } + if points.is_empty() { + return Err(SubjektivError::InvalidRecord( + "non-empty surface materials require at least one grounded point".into(), + )); + } + if points.len() > SURFACE_TOTAL_MATERIAL_LIMIT { + return Err(SubjektivError::InvalidRecord(format!( + "surface output exceeds {SURFACE_TOTAL_MATERIAL_LIMIT} points" + ))); + } + let allowed = materials + .iter() + .map(|material| (material.memory_id.as_str(), material.revision)) + .collect::>(); + let mut bodies = Vec::with_capacity(points.len()); + let mut seen = HashSet::new(); + let mut refs = Vec::new(); + for point in points { + let body = point.body_md.trim(); + validate_nonempty("surface point body", body)?; + if point.memory_refs.is_empty() { + return Err(SubjektivError::InvalidRecord( + "every surface point requires at least one Memory reference".into(), + )); + } + reject_duplicate_refs("surface point", &point.memory_refs)?; + for reference in &point.memory_refs { + if !allowed.contains(&(reference.memory_id.as_str(), reference.revision)) { + return Err(SubjektivError::SubjectScopeMismatch { + subject_id: "surface generation materials".into(), + reference: format!("{}@{}", reference.memory_id, reference.revision), + }); + } + if seen.insert((reference.memory_id.clone(), reference.revision)) { + refs.push(reference.clone()); + } + } + bodies.push(body.to_string()); + } + let body_md = bodies.join("\n\n"); + if estimated_tokens(&body_md) > SURFACE_BODY_TOKEN_BUDGET { + return Err(SubjektivError::InvalidRecord(format!( + "surface body exceeds {SURFACE_BODY_TOKEN_BUDGET} token estimate" + ))); + } + Ok((body_md, refs)) +} + +fn surface_snapshot_for_revision( + transaction: &Transaction<'_>, + subject_id: &str, + store_revision: u64, +) -> Result> { + let raw = transaction + .query_row( + "SELECT snapshot_json FROM surface_snapshots + WHERE subject_id = ?1 AND built_from_store_revision = ?2 + ORDER BY created_at ASC, snapshot_id ASC LIMIT 1", + params![subject_id, to_i64(store_revision)?], + |row| row.get::<_, String>(0), + ) + .optional()?; + raw.map(|raw| parse_surface_snapshot(&raw)).transpose() +} + +fn insert_surface_snapshot( + transaction: &Transaction<'_>, + subject_id: &str, + body_md: String, + memory_refs: Vec, + built_from_store_revision: u64, +) -> Result { + validate_memory_refs(transaction, subject_id, &memory_refs)?; + let snapshot = SurfaceSnapshot { + schema_version: SUBJEKTIV_SCHEMA_VERSION, + id: issued_id("surface"), + subject_id: subject_id.to_string(), + body_md, + memory_refs, + built_from_store_revision, + created_at: now(), + }; + let raw = serde_json::to_string(&snapshot)?; + transaction.execute( + "INSERT INTO surface_snapshots ( + subject_id, snapshot_id, built_from_store_revision, snapshot_json, created_at + ) VALUES (?1, ?2, ?3, ?4, ?5)", + params![ + subject_id, + snapshot.id, + to_i64(snapshot.built_from_store_revision)?, + raw, + snapshot.created_at + ], + )?; + for reference in &snapshot.memory_refs { + transaction.execute( + "INSERT INTO surface_snapshot_refs ( + subject_id, snapshot_id, memory_id, revision + ) VALUES (?1, ?2, ?3, ?4)", + params![ + subject_id, + snapshot.id, + reference.memory_id, + to_i64(reference.revision)? + ], + )?; + } + transaction.execute( + "INSERT INTO surface_snapshot_seals (subject_id, snapshot_id) VALUES (?1, ?2)", + params![subject_id, snapshot.id], + )?; + transaction.execute( + "INSERT INTO surface_generation_state ( + subject_id, store_revision, status, snapshot_id, reason_code, updated_at + ) VALUES (?1, ?2, 'ready', ?3, NULL, ?4) + ON CONFLICT(subject_id) DO UPDATE SET + store_revision = excluded.store_revision, + status = 'ready', snapshot_id = excluded.snapshot_id, + reason_code = NULL, updated_at = excluded.updated_at", + params![ + subject_id, + to_i64(built_from_store_revision)?, + snapshot.id, + snapshot.created_at + ], + )?; + Ok(snapshot) +} + +fn estimated_tokens(value: &str) -> usize { + estimated_tokens_for_bytes(value.len()) +} + +fn estimated_tokens_for_bytes(bytes: usize) -> usize { + bytes.saturating_add(3) / 4 +} + fn apply_candidates_in_transaction( transaction: &Transaction<'_>, subject_id: &str, @@ -1817,6 +2360,20 @@ fn write_memory_revision( subject.updated_at ], )?; + transaction.execute( + "INSERT INTO surface_generation_state ( + subject_id, store_revision, status, snapshot_id, reason_code, updated_at + ) VALUES (?1, ?2, 'dirty', NULL, NULL, ?3) + ON CONFLICT(subject_id) DO UPDATE SET + store_revision = excluded.store_revision, + status = 'dirty', snapshot_id = NULL, reason_code = NULL, + updated_at = excluded.updated_at", + params![ + subject_id, + to_i64(subject.store_revision)?, + subject.updated_at + ], + )?; Ok(record) } @@ -4193,6 +4750,256 @@ mod tests { ); } + #[test] + fn surface_material_selection_is_scoped_active_bounded_and_deterministic() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(temp.path(), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let other = store.create_subject(role()).unwrap(); + store + .create_memory(&other.id, draft("foreign", "foreign subject")) + .unwrap(); + + let kinds = [ + CandidateKind::Preference, + CandidateKind::Constraint, + CandidateKind::Decision, + CandidateKind::WorkingAssumption, + CandidateKind::OpenQuestion, + CandidateKind::Lesson, + ]; + let mut first_id = None; + for kind in &kinds { + for index in 0..9 { + let mut value = draft(&format!("{kind:?}-{index}"), "selection fixture"); + value.kind = kind.clone(); + let memory = store.create_memory(&subject.id, value).unwrap(); + first_id.get_or_insert(memory.id); + } + } + let resolved_id = first_id.unwrap(); + let current = store.memory(&subject.id, &resolved_id).unwrap().unwrap(); + let mut resolved = draft("resolved", "no longer active"); + resolved.kind = current.kind; + resolved.state = MemoryState::Resolved; + store + .revise_memory(&subject.id, &resolved_id, current.revision, resolved) + .unwrap(); + + let first = store.prepare_surface_generation(&subject.id).unwrap(); + let second = store.prepare_surface_generation(&subject.id).unwrap(); + assert_eq!(first.materials, second.materials); + assert_eq!(first.active_memory_count, 53); + assert_eq!(first.materials.len(), SURFACE_TOTAL_MATERIAL_LIMIT); + assert!( + first + .materials + .iter() + .all(|item| item.memory_id != resolved_id) + ); + assert!(first.materials.iter().all(|item| item.body_md != "foreign")); + for kind in &kinds { + assert!( + first + .materials + .iter() + .filter(|item| &item.kind == kind) + .count() + <= SURFACE_PER_KIND_LIMIT + ); + } + let serialized = serde_json::to_vec(&first.materials).unwrap(); + assert!(estimated_tokens_for_bytes(serialized.len()) <= SURFACE_MATERIAL_TOKEN_BUDGET); + } + + #[test] + fn surface_states_validation_races_and_idempotency_fail_closed() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(temp.path(), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + assert_eq!( + store.resident_surface(&subject.id).unwrap().availability, + SurfaceAvailability::Ungenerated + ); + + let empty_generation = store.prepare_surface_generation(&subject.id).unwrap(); + let empty = store + .publish_surface_generation(&subject.id, &empty_generation.id, Vec::new()) + .unwrap(); + assert!(empty.body_md.is_empty()); + assert_eq!( + store.resident_surface(&subject.id).unwrap().availability, + SurfaceAvailability::Ready + ); + + let memory = store + .create_memory(&subject.id, draft("Current constraint", "confirmed")) + .unwrap(); + assert_eq!( + store.resident_surface(&subject.id).unwrap().availability, + SurfaceAvailability::Stale + ); + let stale_generation = store.prepare_surface_generation(&subject.id).unwrap(); + let invalid = store.publish_surface_generation( + &subject.id, + &stale_generation.id, + vec![SurfacePoint { + body_md: "Invented".into(), + memory_refs: vec![MemoryRevisionRef { + memory_id: "foreign-memory".into(), + revision: 1, + }], + }], + ); + assert!(matches!( + invalid, + Err(SubjektivError::SubjectScopeMismatch { .. }) + )); + store + .fail_surface_generation(&subject.id, &stale_generation.id, "invalid_output") + .unwrap(); + assert_eq!( + store.resident_surface(&subject.id).unwrap().availability, + SurfaceAvailability::Failed + ); + assert_eq!( + store + .memory(&subject.id, &memory.id) + .unwrap() + .unwrap() + .revision, + 1, + "surface failure must not roll back confirmed Memory" + ); + + let raced = store.prepare_surface_generation(&subject.id).unwrap(); + store + .revise_memory( + &subject.id, + &memory.id, + 1, + draft("Corrected constraint", "correction"), + ) + .unwrap(); + assert!(matches!( + store.publish_surface_generation( + &subject.id, + &raced.id, + vec![SurfacePoint { + body_md: "Old wording".into(), + memory_refs: vec![MemoryRevisionRef { + memory_id: memory.id.clone(), + revision: 1, + }], + }], + ), + Err(SubjektivError::SurfaceGenerationConflict(_)) + )); + assert_eq!( + store.resident_surface(&subject.id).unwrap().availability, + SurfaceAvailability::Stale + ); + + let current = store.prepare_surface_generation(&subject.id).unwrap(); + let current_ref = MemoryRevisionRef { + memory_id: memory.id.clone(), + revision: 2, + }; + let points = vec![SurfacePoint { + body_md: "Keep the corrected constraint, including its condition.".into(), + memory_refs: vec![current_ref.clone()], + }]; + let published = store + .publish_surface_generation(&subject.id, ¤t.id, points.clone()) + .unwrap(); + let retry = store + .publish_surface_generation(&subject.id, ¤t.id, points) + .unwrap(); + assert_eq!(published.id, retry.id); + let resident = store.resident_surface(&subject.id).unwrap(); + assert_eq!(resident.availability, SurfaceAvailability::Ready); + assert_eq!(resident.snapshot.unwrap().memory_refs, vec![current_ref]); + } + + #[test] + fn active_memory_that_cannot_fit_is_a_failure_not_a_normal_empty_surface() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(temp.path(), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let mut oversized = draft("oversized", "fixture"); + oversized.body_md = "x".repeat((SURFACE_MATERIAL_TOKEN_BUDGET + 1) * 4); + store.create_memory(&subject.id, oversized).unwrap(); + + let generation = store.prepare_surface_generation(&subject.id).unwrap(); + assert_eq!(generation.active_memory_count, 1); + assert!(generation.materials.is_empty()); + assert!( + store + .publish_surface_generation(&subject.id, &generation.id, Vec::new()) + .is_err() + ); + store + .fail_surface_generation(&subject.id, &generation.id, "input_budget_exhausted") + .unwrap(); + assert_eq!( + store.resident_surface(&subject.id).unwrap().availability, + SurfaceAvailability::Failed + ); + } + + #[test] + fn concurrent_surface_generations_cannot_overwrite_one_revision() { + let temp = tempfile::tempdir().unwrap(); + let (_manager, _workspace, store) = open_store(temp.path(), "workspace-a"); + let subject = store.create_subject(role()).unwrap(); + let memory = store + .create_memory(&subject.id, draft("Concurrent constraint", "fixture")) + .unwrap(); + let first = store.prepare_surface_generation(&subject.id).unwrap(); + let second = store.prepare_surface_generation(&subject.id).unwrap(); + let barrier = Arc::new(Barrier::new(2)); + let mut joins = Vec::new(); + for (generation, body) in [(first, "first"), (second, "second")] { + let store = store.clone(); + let subject_id = subject.id.clone(); + let memory_id = memory.id.clone(); + let barrier = Arc::clone(&barrier); + joins.push(thread::spawn(move || { + barrier.wait(); + store.publish_surface_generation( + &subject_id, + &generation.id, + vec![SurfacePoint { + body_md: body.into(), + memory_refs: vec![MemoryRevisionRef { + memory_id, + revision: 1, + }], + }], + ) + })); + } + let results = joins + .into_iter() + .map(|join| join.join().unwrap()) + .collect::>(); + assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1); + assert_eq!( + results + .iter() + .filter(|result| matches!( + result, + Err(SubjektivError::SurfaceGenerationConflict(_)) + )) + .count(), + 1 + ); + assert_eq!( + store.resident_surface(&subject.id).unwrap().availability, + SurfaceAvailability::Ready + ); + } + #[test] fn reopen_and_backup_restore_preserve_exact_history() { let temp = tempfile::tempdir().unwrap(); diff --git a/docs/design/subjektiv-surface.md b/docs/design/subjektiv-surface.md new file mode 100644 index 000000000..41d46f311 --- /dev/null +++ b/docs/design/subjektiv-surface.md @@ -0,0 +1,46 @@ +# Subject Memory surface generation + +The subject Memory surface is a bounded start-time context derived from confirmed, revisioned Memory. It is not another Memory authority and is never used as extraction or consolidation input. + +## Fixed policy + +The Backend captures one generation from current `active` revisions for exactly one subject. Staging records, Session history, historical revisions, and prior surfaces are not queried. + +Selection is deterministic: + +1. categories use the fixed order `preference`, `constraint`, `decision`, `working_assumption`, `open_question`, `lesson`; +2. each category is ordered by `updated_at DESC, memory_id ASC`; +3. at most 8 records are retained per category; +4. records are allocated round-robin by category, up to 24 records total; +5. the serialized material payload uses the repository's provider-independent token estimate, `ceil(UTF-8 bytes / 4)`, and is capped at 10,000 estimated tokens; the complete editor-input policy budget is 12,000 tokens. + +The clean-context editor uses the effective `builtin:subjektiv-memory-consolidation` profile model (`codex-oauth/gpt-5.6-luna`, medium reasoning). A generation permits three model turns (the initial attempt plus at most two correction turns). A store-generation conflict causes at most one complete re-read and regeneration. The published Markdown is capped at 1,024 tokens by the same estimate. + +Every editor point carries one or more exact `{memory_id, revision}` references. The Host checks the output shape, non-empty grounding, budget, subject scope, membership in the captured materials, and current store generation. These checks prove reference existence and freshness only; they do not claim semantic correctness. + +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. Worker prompt materialization remains a start/restore boundary and does not rewrite recorded history. + +## Semantic fixture and verification + +Use a test subject with these active materials: + +- constraint: “Never send credentials to tools.” (`staleness`: “Until credential brokering is redesigned.”) +- constraint: “A trusted broker may deliver short-lived credentials to the Git provider.” +- decision: “Publish only the Ticket work branch; never force-push.” +- working assumption: “The provider resolves selectors after a normal push.” +- open question: “Does the provider preserve approval after target-only movement?” +- lesson: “Checking only the local branch previously produced non-reviewable evidence.” +- duplicate lesson: “Local HEAD alone is not immutable review evidence.” + +A valid review fixture must: + +- retain “never” and the credential-redesign expiry condition; +- present the apparently conflicting broker constraint without silently declaring one side invalid; +- preserve the target-only-movement item as an open question rather than a decision; +- combine the two review-evidence lessons only if both meanings remain; +- cite only exact refs supplied with those materials; +- distinguish mechanical citation validation from a human/model semantic review of the wording. + +The repository tests exercise deterministic bounds, subject isolation, active-current-only selection, output/reference rejection, empty/un-generated/failed/stale states, generation races, parallel publication, correction/retraction staleness, and publication idempotency. The prompt fixture above is the manual semantic check for condition, negation, reservation, conflict, expiry, and duplicate editing behavior. diff --git a/resources/prompts/catalog.dcdl b/resources/prompts/catalog.dcdl index 271420d6c..13d8fc12c 100644 --- a/resources/prompts/catalog.dcdl +++ b/resources/prompts/catalog.dcdl @@ -24,6 +24,7 @@ internalBackendJobSystem = import "./internal/backend_job_system.md"; internalTicketItemChecker = import "./internal/ticket_item_checker.md"; internalMemoryConsolidationSystem = import "./internal/memory_consolidation_system.md"; internalSubjektivMemoryConsolidationSystem = import "./internal/subjektiv_memory_consolidation_system.md"; +internalSubjektivMemorySurfaceSystem = import "./internal/subjektiv_memory_surface_system.md"; internalMemoryExtractSystem = import "./internal/memory_extract_system.md"; internalWorkspaceOrchestratorQueueAttention = import "./internal/workspace_orchestrator_queue_attention.md"; internalNotifyWrapper = import "./internal/notify_wrapper.md"; @@ -62,6 +63,7 @@ in ticket_item_checker = internalTicketItemChecker.content; memory_consolidation_system = internalMemoryConsolidationSystem.content; subjektiv_memory_consolidation_system = internalSubjektivMemoryConsolidationSystem.content; + subjektiv_memory_surface_system = internalSubjektivMemorySurfaceSystem.content; memory_extract_system = internalMemoryExtractSystem.content; workspace_orchestrator_queue_attention = internalWorkspaceOrchestratorQueueAttention.content; notify_wrapper = internalNotifyWrapper.content; diff --git a/resources/prompts/internal/subjektiv_memory_consolidation_system.md b/resources/prompts/internal/subjektiv_memory_consolidation_system.md index b04cef77b..7cb150dc7 100644 --- a/resources/prompts/internal/subjektiv_memory_consolidation_system.md +++ b/resources/prompts/internal/subjektiv_memory_consolidation_system.md @@ -26,7 +26,7 @@ Rules: - `source_candidate_ids` are host-derived from the applied candidate. Do not attempt to author them. - `derived_from` must contain only exact existing Memory revision refs returned by reads. - Applied responses return affected refs with `{memory_id, revision, operation}`. `duplicate` and `already_covered` may reference exact existing revisions; `discarded` and `invalid` require a concrete reason. -- Surface/snapshot generation is deferred. `surface_dirty: true` reports that a later process should rebuild it; never treat surface generation failure as failure of an already returned committed decision. +- Surface generation is a separate clean-context Host lifecycle that runs once after this consolidation turn commits. `surface_dirty: true` means confirmed Memory changed; never treat later surface generation failure as failure of an already returned committed decision. - An empty model response or transport failure is not a candidate disposition. Only a successful `MemoryApplyCandidate` response closes a candidate. Valid non-applied close actions are `discarded`, `invalid`, `duplicate`, and `already_covered`. Always provide a specific reason. diff --git a/resources/prompts/internal/subjektiv_memory_surface_system.md b/resources/prompts/internal/subjektiv_memory_surface_system.md new file mode 100644 index 000000000..d403fb5fb --- /dev/null +++ b/resources/prompts/internal/subjektiv_memory_surface_system.md @@ -0,0 +1,32 @@ +# Subject Memory surface editor + +You are a clean-context editor for one subject's confirmed Memory surface. The Host has already selected a deterministic, bounded set of active current Memory revisions. Those materials are the only facts and references you may use. + +Answer this fixed question: + +> この主体が次の作業を始める際、毎回思い出しておくべきことは何か。継続的な制約、判断の前提、未解決事項、再発を避けたい教訓を、指定予算内でまとめる。 + +Call `SubmitMemorySurface` exactly once with `{ "points": [...] }`. Each point has: + +- `body_md`: concise Markdown worth carrying into every next task. +- `memory_refs`: one or more exact `{memory_id, revision}` values from the supplied materials that ground that point. + +Rules: + +- Do not add a fact, decision, identifier, or reference absent from the materials. +- Preserve conditions, negation, reservations, and stated staleness/expiry conditions. +- Similar material may be combined only when the combined wording preserves each source's meaning. +- Do not resolve contradictions. State conflicting constraints or assumptions as unresolved when both matter. +- This is not a complete index. Keep only continuously useful constraints/preferences, active decisions/working assumptions, open questions, and recurrence-preventing lessons. +- Every point needs at least one supplied exact reference. Never invent a citation. +- Reference existence and subject scope are mechanically checked. That check is not semantic verification; you remain responsible for faithful wording. +- Stay within the supplied body token budget. The Host uses the documented provider-independent UTF-8 byte estimate. +- If the materials array is empty, submit an empty points array without inventing content. +- A rejected tool call may be corrected, but this Worker is bounded to three model turns total (initial attempt plus at most two correction turns). + +## Language + +- `language`: `{{language}}` +- Write the surface in this language. +- Preserve literal identifiers, paths, commands, branch names, issue IDs, tool names, model names, and quoted user/system text as-is. +- If the configured language is unclear, use English.