diff --git a/crates/memory/src/backend.rs b/crates/memory/src/backend.rs index fcbf8b68..45c17633 100644 --- a/crates/memory/src/backend.rs +++ b/crates/memory/src/backend.rs @@ -11,8 +11,11 @@ use std::path::PathBuf; use serde::{Deserialize, Serialize}; -use crate::extract::{ExtractedPayload, write_staging}; -use crate::schema::SourceRef; +use crate::audit::{AuditEvent, append_audit_event}; +use crate::extract::{ + ExtractedCandidate, ExtractedPayload, StagingEvidence, write_staging, write_staging_candidate, +}; +use crate::schema::{SourceEvidenceRef, SourceRef}; use crate::tool::MemoryToolKind; use crate::workspace::WorkspaceLayout; @@ -24,6 +27,9 @@ pub enum MemoryBackendOperation { Write(MemoryWriteOperation), Edit(MemoryEditOperation), Delete(MemoryDeleteOperation), + ResidentSummary(MemoryResidentSummaryOperation), + AppendAudit(MemoryAppendAuditOperation), + StageCandidate(MemoryStageCandidateOperation), StageExtracted(MemoryStageExtractedOperation), } @@ -42,6 +48,7 @@ pub enum MemoryBackendHttpResponse { #[serde(tag = "kind", rename_all = "snake_case")] pub enum MemoryBackendOperationResult { ToolOutput(MemoryToolOutput), + Acknowledged(MemoryBackendAckOutput), StagingWritten(MemoryStagingWriteOutput), } @@ -95,12 +102,36 @@ pub struct MemoryDeleteOperation { pub slug: Option, } +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct MemoryResidentSummaryOperation {} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MemoryAppendAuditOperation { + pub event: AuditEvent, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MemoryStageCandidateOperation { + pub source: SourceRef, + pub extract_run_id: String, + pub candidate: ExtractedCandidate, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub evidence: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub source_refs: Vec, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MemoryStageExtractedOperation { pub source: SourceRef, pub payload: ExtractedPayload, } +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct MemoryBackendAckOutput { + pub summary: String, +} + #[derive(Debug, Clone, Serialize, Deserialize)] pub struct MemoryStagingWriteOutput { pub staging_count: usize, @@ -127,6 +158,33 @@ pub fn execute_memory_backend_operation( MemoryBackendOperation::Delete(operation) => { execute_delete(layout, operation).map(MemoryBackendOperationResult::ToolOutput) } + MemoryBackendOperation::ResidentSummary(_operation) => Ok( + MemoryBackendOperationResult::ToolOutput(execute_resident_summary(layout)), + ), + MemoryBackendOperation::AppendAudit(operation) => { + append_audit_event(layout, &operation.event)?; + Ok(MemoryBackendOperationResult::Acknowledged( + MemoryBackendAckOutput { + summary: "memory audit event appended".to_string(), + }, + )) + } + MemoryBackendOperation::StageCandidate(operation) => { + let written = write_staging_candidate( + layout, + operation.source, + &operation.extract_run_id, + operation.candidate, + operation.evidence, + operation.source_refs, + )?; + Ok(MemoryBackendOperationResult::StagingWritten( + MemoryStagingWriteOutput { + staging_count: 1, + staging_ids: vec![written.id.to_string()], + }, + )) + } MemoryBackendOperation::StageExtracted(operation) => { let written = write_staging(layout, operation.source, operation.payload)?; Ok(MemoryBackendOperationResult::StagingWritten( @@ -139,6 +197,19 @@ pub fn execute_memory_backend_operation( } } +fn execute_resident_summary(layout: &WorkspaceLayout) -> MemoryToolOutput { + match crate::collect_resident_summary(layout) { + Some(summary) => MemoryToolOutput { + summary: "resident memory summary collected".to_string(), + content: Some(summary), + }, + None => MemoryToolOutput { + summary: "resident memory summary unavailable".to_string(), + content: None, + }, + } +} + fn execute_query( layout: &WorkspaceLayout, operation: MemoryQueryOperation, diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index c9df0b0d..bddec8fb 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -9,7 +9,7 @@ use std::collections::HashMap; use std::future::Future; -use std::path::{Path, PathBuf}; +use std::path::PathBuf; use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, Mutex, mpsc}; use std::time::Duration; @@ -240,9 +240,6 @@ impl ProfileRuntimeWorkerFactory { #[derive(Debug, Clone)] enum RuntimeWorkspaceBackendRef { None, - LocalFilesystem { - root: PathBuf, - }, Http { workspace_id: String, base_url: String, @@ -250,32 +247,19 @@ enum RuntimeWorkspaceBackendRef { } impl RuntimeWorkspaceBackendRef { - fn from_worker_request( - request: &CreateWorkerRequest, - binding: Option<&WorkingDirectoryBinding>, - ) -> Self { + fn from_worker_request(request: &CreateWorkerRequest) -> Self { if let Some(api) = request.workspace_api.as_ref() { return Self::Http { workspace_id: api.workspace_id.clone(), base_url: api.base_url.clone(), }; } - Self::from_working_directory(binding) - } - - fn from_working_directory(binding: Option<&WorkingDirectoryBinding>) -> Self { - match binding { - Some(binding) => Self::LocalFilesystem { - root: binding.root().to_path_buf(), - }, - None => Self::None, - } + Self::None } fn worker_context(&self) -> WorkerWorkspaceContext { match self { Self::None => WorkerWorkspaceContext::no_workspace(), - Self::LocalFilesystem { root } => local_workspace_context(root), Self::Http { workspace_id, base_url, @@ -287,17 +271,6 @@ impl RuntimeWorkspaceBackendRef { } } -fn local_workspace_context(root: &Path) -> WorkerWorkspaceContext { - WorkerWorkspaceContext::local_filesystem(read_workspace_id_hint(root)) -} - -fn read_workspace_id_hint(root: &Path) -> Option { - let contents = std::fs::read_to_string(root.join(".yoi/workspace.toml")).ok()?; - let value = toml::from_str::(&contents).ok()?; - let id = value.get("id")?.as_str()?.to_string(); - WorkspaceId::new(id).ok() -} - #[cfg(feature = "http-server")] async fn fetch_profile_source_archive_http( location: &ProfileSourceArchiveHttpRef, @@ -377,10 +350,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { ) }) .unwrap_or(WorkerFilesystemAuthority::None); - let workspace_backend_ref = RuntimeWorkspaceBackendRef::from_worker_request( - &request.request, - request.working_directory.as_ref(), - ); + let workspace_backend_ref = + RuntimeWorkspaceBackendRef::from_worker_request(&request.request); let workspace_context = workspace_backend_ref.worker_context(); let selector = profile.as_deref().unwrap_or("builtin:default"); let archive = self @@ -458,10 +429,8 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { ) }) .unwrap_or(WorkerFilesystemAuthority::None); - let workspace_backend_ref = RuntimeWorkspaceBackendRef::from_worker_request( - &request.request, - request.working_directory.as_ref(), - ); + let workspace_backend_ref = + RuntimeWorkspaceBackendRef::from_worker_request(&request.request); let workspace_context = workspace_backend_ref.worker_context(); let (manifest, loader) = Self::restore_fallback_manifest(&worker_name)?; @@ -1264,10 +1233,8 @@ mod tests { .as_ref() .map(|binding| binding.root().to_path_buf()) .unwrap_or_else(|| self.cwd.clone()); - let workspace_backend_ref = RuntimeWorkspaceBackendRef::from_worker_request( - &request.request, - request.working_directory.as_ref(), - ); + let workspace_backend_ref = + RuntimeWorkspaceBackendRef::from_worker_request(&request.request); let workspace_context = workspace_backend_ref.worker_context(); self.observed_workspace_clients .lock() @@ -1664,6 +1631,12 @@ mod tests { assert!(cwd.starts_with(runtime_base.path())); assert!(!cwd.starts_with(repo.path())); assert!(cwd.join("README.md").exists()); + assert_eq!( + observed_workspace_clients.lock().unwrap().as_slice(), + &[WorkspaceClient::Unavailable { + reason: "no workspace configured".to_string() + }] + ); } #[test] diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index b671fc68..7566cbb0 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -6,10 +6,7 @@ use llm_engine::EngineError; use llm_engine::llm_client::client::LlmClient; use session_store::WorkerMetadataStore; use session_store::{LogEntry, Store}; -use ticket::LocalTicketBackend; -use ticket::config::TicketConfig; use tokio::sync::{broadcast, mpsc, oneshot}; -use tracing::{debug, warn}; use crate::discovery::{ WorkerDiscovery, list_workers_tool, restore_worker_tool, send_to_peer_worker_tool, @@ -29,9 +26,6 @@ use crate::shutdown_after_idle::{ use crate::spawn::comm_tools::{read_worker_output_tool, send_to_worker_tool, stop_worker_tool}; use crate::spawn::registry::SpawnedWorkerRegistry; use crate::spawn::tool::spawn_worker_tool; -use crate::ticket_event_notify::{ - TicketEventCompanionNotifyHook, companion_worker_name_for_workspace, -}; use crate::worker::{SystemItemCommitter, Worker, WorkerError, WorkerRunResult, WorkspaceClient}; use protocol::{ AlertLevel, AlertSource, ErrorCode, Event, Method, RewindTargetId, RunResult, Segment, @@ -288,12 +282,6 @@ impl WorkerController { ) .await?; - install_ticket_event_companion_notify_hook( - &mut worker, - runtime_base.to_path_buf(), - spawned_registry.clone(), - ); - // Intake role Workers self-terminate only after a successful // TicketIntakeReady turn has fully settled back to Idle. The request // is transient controller state, not model-visible context or ticket @@ -533,84 +521,6 @@ fn wire_event_bridges_on_engine( // per-item commit channel is wired at the top of this function. } -fn install_ticket_event_companion_notify_hook( - worker: &mut Worker, - runtime_base: PathBuf, - spawned_registry: Arc, -) where - C: LlmClient + Clone + 'static, - St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, -{ - if !is_ticket_orchestrator_role(worker.runtime_ticket_role()) { - return; - } - - let ticket_feature = &worker.manifest().feature.ticket; - if !ticket_feature.enabled || !ticket_feature.orchestration_control { - return; - } - - let Some(local) = worker.local_working_directory() else { - return; - }; - let Some(companion_worker_name) = companion_worker_name_for_workspace(&local.root) else { - return; - }; - if companion_worker_name == worker.manifest().worker.name { - return; - } - - let Ok(ticket_config) = TicketConfig::load_workspace(&local.cwd) else { - return; - }; - let backend_root = ticket_config.backend_root().to_path_buf(); - if !backend_root.is_dir() { - return; - } - - let discovery = WorkerDiscovery::new( - worker.worker_metadata_store(), - worker.manifest().worker.name.clone(), - runtime_base, - Some(local.cwd.clone()), - spawned_registry, - ); - match discovery.ensure_existing_peer(&companion_worker_name) { - Ok(Some(_)) => { - debug!( - companion = %companion_worker_name, - orchestrator = %worker.manifest().worker.name, - "ensured Companion peer relationship for Orchestrator Ticket event notifications" - ); - } - Ok(None) => { - debug!( - companion = %companion_worker_name, - orchestrator = %worker.manifest().worker.name, - "Companion metadata is missing; Ticket event notifications will skip until Companion exists" - ); - } - Err(error) => { - warn!( - companion = %companion_worker_name, - orchestrator = %worker.manifest().worker.name, - error = %error, - "failed to ensure Companion peer relationship for Orchestrator Ticket event notifications" - ); - } - } - worker.add_post_tool_call_hook(TicketEventCompanionNotifyHook::new( - LocalTicketBackend::new(backend_root), - discovery, - companion_worker_name, - )); -} - -fn is_ticket_orchestrator_role(role: Option<&str>) -> bool { - role.map(|role| role.eq_ignore_ascii_case("orchestrator")) - .unwrap_or(false) -} - /// Register the builtin file-manipulation tools, optional memory tools, /// and the Worker-orchestration tools (SpawnWorker + comm) on the Worker's /// Engine. Returns the `ScopedFs` clone used to attach a `WorkerFsView` to @@ -632,7 +542,6 @@ where let local_filesystem = worker.local_working_directory().cloned(); let local_workspace_root = local_filesystem.as_ref().map(|local| local.root.clone()); let task_feature = worker.task_feature(); - let session_id_for_usage = worker.segment_id().to_string(); let memory_config = worker.manifest().memory.clone(); let web_config = worker.manifest().web.clone(); let mcp_config = worker.manifest().mcp.clone(); @@ -679,8 +588,8 @@ where orchestration_control: feature_config.ticket.orchestration_control, }; // Ticket tools are typed operations over the current workspace Ticket backend. - // Runtime-hosted Workers prefer the workspace API URI carried by the - // Worker context; legacy/local Workers fall back to the checked-out worktree. + // Workspace access must be authority-bound to the Backend Workspace API; the + // Worker must not fall back to a local `.yoi/tickets` store. let ticket_backend = match worker.workspace_client() { WorkspaceClient::Http { workspace_id, @@ -690,18 +599,10 @@ where base_url: base_url.clone(), }, _ => { - let ticket_cwd = local_filesystem - .as_ref() - .map(|local| &local.cwd) - .ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "ticket tools require local Worker filesystem authority", - ) - })?; - crate::feature::builtin::ticket::TicketFeatureBackend::LocalWorkspace { - workspace_root: ticket_cwd.clone(), - } + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "ticket tools require Backend Workspace API authority", + )); } }; feature_registry.add_module( @@ -729,27 +630,18 @@ where let workspace_client = worker.workspace_client().clone(); let worker = worker.engine_mut(); - // Memory tools require explicit feature exposure. Storage access may be - // provided by local filesystem authority or the path-free Workspace HTTP API. + // Memory tools require explicit feature exposure. Workspace memory access + // is authority-bound to the Backend Workspace API; the Worker must not + // register local filesystem memory tools even when it has local cwd/root + // authority for shell/file tools. if feature_config.memory.enabled { - if let Some(workspace_root) = local_workspace_root.as_ref() { - let mem = memory_config.as_ref().ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "[feature.memory].enabled = true requires a [memory] configuration section", - ) - })?; - let layout = memory::WorkspaceLayout::resolve(mem, workspace_root); - let query_cfg = memory::tool::QueryConfig::from(mem); - worker.register_tool(memory::tool::read_tool_with_usage( - layout.clone(), - session_id_for_usage, - )); - worker.register_tool(memory::tool::write_tool(layout.clone())); - worker.register_tool(memory::tool::edit_tool(layout.clone())); - worker.register_tool(memory::tool::delete_tool(layout.clone())); - worker.register_tool(memory::tool::memory_query_tool(layout, query_cfg)); - } else if let WorkspaceClient::Http { + let _mem = memory_config.as_ref().ok_or_else(|| { + std::io::Error::new( + std::io::ErrorKind::InvalidInput, + "[feature.memory].enabled = true requires a [memory] configuration section", + ) + })?; + if let WorkspaceClient::Http { workspace_id, base_url, } = workspace_client @@ -763,7 +655,7 @@ where } else { return Err(std::io::Error::new( std::io::ErrorKind::InvalidInput, - "memory tools require Workspace HTTP API or local Worker filesystem authority", + "memory tools require Backend Workspace API authority", )); } } diff --git a/crates/worker/src/entrypoint.rs b/crates/worker/src/entrypoint.rs index 762c31b8..cb84ec7a 100644 --- a/crates/worker/src/entrypoint.rs +++ b/crates/worker/src/entrypoint.rs @@ -247,17 +247,14 @@ fn load_profile( } pub fn resolve_runtime_profile_manifest( - profile: Option<&str>, - workspace_root: &Path, - worker_name: &str, + _profile: Option<&str>, + _workspace_root: &Path, + _worker_name: &str, ) -> Result<(WorkerManifest, PromptLoader), String> { - let selector = profile - .map(ProfileSelector::parse_cli) - .unwrap_or(ProfileSelector::Default); - let (mut manifest, loader) = load_profile(&selector, workspace_root, worker_name)?; - apply_profile_launch_policy(&mut manifest, workspace_root, None)?; - apply_plugin_resolution_plan(&mut manifest, workspace_root); - Ok((manifest, loader)) + Err( + "runtime profile resolution requires a pre-resolved manifest/profile archive from Backend authority" + .to_string(), + ) } pub fn resolve_runtime_profile_manifest_from_manifest( @@ -269,13 +266,15 @@ pub fn resolve_runtime_profile_manifest_from_manifest( manifest.worker.name = worker_name.to_string(); } apply_profile_launch_policy(&mut manifest, workspace_root, None)?; - apply_plugin_resolution_plan(&mut manifest, workspace_root); + // Do not run plugin discovery here: runtime-created Workers receive their + // resolved manifest/profile archive from Backend authority, not by scanning + // the materialized workdir's `.yoi/plugins`. Ok((manifest, PromptLoader::builtins_only())) } pub fn resolve_runtime_profile_manifest_from_manifest_without_filesystem( mut manifest: WorkerManifest, - workspace_root: &Path, + _workspace_root: &Path, worker_name: &str, ) -> Result<(WorkerManifest, PromptLoader), String> { if manifest.worker.name.is_empty() { @@ -283,7 +282,7 @@ pub fn resolve_runtime_profile_manifest_from_manifest_without_filesystem( } manifest.scope = ScopeConfig::default(); manifest.delegation_scope = ScopeConfig::default(); - apply_plugin_resolution_plan(&mut manifest, workspace_root); + // Same as the filesystem-capable runtime path: no local `.yoi` discovery. Ok((manifest, PromptLoader::builtins_only())) } diff --git a/crates/worker/src/feature/builtin/memory.rs b/crates/worker/src/feature/builtin/memory.rs index 86b62b43..b7542e07 100644 --- a/crates/worker/src/feature/builtin/memory.rs +++ b/crates/worker/src/feature/builtin/memory.rs @@ -18,6 +18,8 @@ use memory::backend::{ use serde::de::DeserializeOwned; use serde_json::json; +use crate::worker::WorkspaceClient; + #[derive(Clone, Debug)] pub struct WorkspaceHttpMemoryBackend { workspace_id: String, @@ -32,43 +34,92 @@ impl WorkspaceHttpMemoryBackend { } } + pub fn execute_operation( + &self, + operation: MemoryBackendOperation, + ) -> Result { + execute_http_memory_backend(&self.workspace_id, &self.base_url, operation) + } + fn execute(&self, operation: MemoryBackendOperation) -> Result { - let url = format!( - "{}/api/w/{}/memory/backend", - self.base_url.trim_end_matches('/'), - self.workspace_id - ); - let response = reqwest::blocking::Client::new() - .post(url) - .json(&operation) - .send() - .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; - let status = response.status(); - let body = response - .text() - .map_err(|error| ToolError::ExecutionFailed(error.to_string()))?; - if !status.is_success() { - return Err(ToolError::ExecutionFailed(format!( - "workspace memory backend returned HTTP {status}: {body}" - ))); - } - let response: MemoryBackendHttpResponse = serde_json::from_str(&body).map_err(|error| { - ToolError::ExecutionFailed(format!("decode memory backend response: {error}")) - })?; - match response { - MemoryBackendHttpResponse::Ok { - result: MemoryBackendOperationResult::ToolOutput(output), - } => Ok(tool_output(output)), - MemoryBackendHttpResponse::Ok { result } => Err(ToolError::ExecutionFailed(format!( + match self.execute_operation(operation) { + Ok(MemoryBackendOperationResult::ToolOutput(output)) => Ok(tool_output(output)), + Ok(result) => Err(ToolError::ExecutionFailed(format!( "unexpected memory backend result for model-visible tool: {result:?}" ))), - MemoryBackendHttpResponse::Error { message } => { - Err(ToolError::ExecutionFailed(message)) + Err(error) => Err(ToolError::ExecutionFailed(error.to_string())), + } + } +} + +#[derive(Debug, thiserror::Error)] +pub enum WorkspaceMemoryBackendError { + #[error("workspace memory backend is unavailable: {reason}")] + Unavailable { reason: String }, + #[error("workspace memory backend request failed: {0}")] + Request(#[from] reqwest::Error), + #[error("workspace memory backend returned HTTP {status}: {body}")] + Http { + status: reqwest::StatusCode, + body: String, + }, + #[error("decode memory backend response: {0}")] + Decode(#[from] serde_json::Error), + #[error("workspace memory backend rejected operation: {0}")] + Backend(String), +} + +impl WorkspaceClient { + pub fn execute_memory_backend_operation( + &self, + operation: MemoryBackendOperation, + ) -> Result { + match self { + WorkspaceClient::Http { + workspace_id, + base_url, + } => execute_http_memory_backend(workspace_id, base_url, operation), + WorkspaceClient::Available { kind } => Err(WorkspaceMemoryBackendError::Unavailable { + reason: format!( + "workspace client kind `{kind}` does not expose the Backend Workspace API" + ), + }), + WorkspaceClient::Unavailable { reason } => { + Err(WorkspaceMemoryBackendError::Unavailable { + reason: reason.clone(), + }) } } } } +fn execute_http_memory_backend( + workspace_id: &str, + base_url: &str, + operation: MemoryBackendOperation, +) -> Result { + let url = format!( + "{}/api/w/{}/memory/backend", + base_url.trim_end_matches('/'), + workspace_id + ); + let response = reqwest::blocking::Client::new() + .post(url) + .json(&operation) + .send()?; + let status = response.status(); + let body = response.text()?; + if !status.is_success() { + return Err(WorkspaceMemoryBackendError::Http { status, body }); + } + match serde_json::from_str::(&body)? { + MemoryBackendHttpResponse::Ok { result } => Ok(result), + MemoryBackendHttpResponse::Error { message } => { + Err(WorkspaceMemoryBackendError::Backend(message)) + } + } +} + pub fn workspace_http_memory_tools( workspace_id: impl Into, base_url: impl Into, diff --git a/crates/worker/src/feature/builtin/session_explore.rs b/crates/worker/src/feature/builtin/session_explore.rs index b7153b29..23cfaa94 100644 --- a/crates/worker/src/feature/builtin/session_explore.rs +++ b/crates/worker/src/feature/builtin/session_explore.rs @@ -2,11 +2,11 @@ use std::sync::{Arc, Mutex}; use async_trait::async_trait; use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; -use memory::extract::{ - CandidateKind, ExtractedCandidate, StagingWriteResult, write_staging_candidate, +use memory::backend::{ + MemoryBackendOperation, MemoryBackendOperationResult, MemoryStageCandidateOperation, }; +use memory::extract::{CandidateKind, ExtractedCandidate}; use memory::schema::SourceRef; -use memory::workspace::WorkspaceLayout; use schemars::JsonSchema; use serde::Deserialize; use uuid::Uuid; @@ -19,6 +19,7 @@ use crate::session_reference::{ ReadDetail, ReadOptions, ReadSelector, ReferenceKind, SearchOptions, SessionReferenceView, ToolPart, }; +use crate::worker::WorkspaceClient; const SEARCH_EVIDENCE_DESCRIPTION: &str = "Search the host-created session evidence index. Use this to find stable evidence ids before staging a memory candidate. Supports kind=user|assistant|system|tool and tool_part=input|output|both."; const READ_EVIDENCE_DESCRIPTION: &str = "Read bounded session evidence by evidence_id or entry_range. Use compact mode for normal verification and full mode only when exact tool arguments or result content are necessary."; @@ -28,22 +29,22 @@ const FINISH_EXTRACTION_DESCRIPTION: &str = "Finish the extract worker run after #[derive(Clone)] pub(crate) struct SessionExploreState { view: Arc, - layout: WorkspaceLayout, + workspace_client: WorkspaceClient, source: SourceRef, extract_run_id: String, - staged: Arc>>, + staged: Arc>>, finished: Arc>>, } impl SessionExploreState { pub(crate) fn new( view: SessionReferenceView, - layout: WorkspaceLayout, + workspace_client: WorkspaceClient, source: SourceRef, ) -> Self { Self { view: Arc::new(view), - layout, + workspace_client, source, extract_run_id: Uuid::now_v7().to_string(), staged: Arc::new(Mutex::new(Vec::new())), @@ -55,7 +56,7 @@ impl SessionExploreState { &self.view } - pub(crate) fn staged(&self) -> Vec { + pub(crate) fn staged(&self) -> Vec { self.staged .lock() .expect("session explore staged state poisoned") @@ -420,21 +421,45 @@ impl Tool for StageCandidateTool { staleness: params.staleness, evidence_ids: params.evidence_ids, }; - let written = write_staging_candidate( - &self.state.layout, - self.state.source.clone(), - &self.state.extract_run_id, - candidate, - evidence, - source_refs, - ) - .map_err(|e| ToolError::ExecutionFailed(format!("write staging failed: {e}")))?; - let id = written.id.to_string(); + let result = self + .state + .workspace_client + .execute_memory_backend_operation(MemoryBackendOperation::StageCandidate( + MemoryStageCandidateOperation { + source: self.state.source.clone(), + extract_run_id: self.state.extract_run_id.clone(), + candidate, + evidence, + source_refs, + }, + )) + .map_err(|e| ToolError::ExecutionFailed(format!("write staging failed: {e}")))?; + let ids = match result { + MemoryBackendOperationResult::StagingWritten(output) if output.staging_count == 1 => { + output.staging_ids + } + MemoryBackendOperationResult::StagingWritten(output) => { + return Err(ToolError::ExecutionFailed(format!( + "stage_candidate expected one staging record, backend wrote {}", + output.staging_count + ))); + } + other => { + return Err(ToolError::ExecutionFailed(format!( + "unexpected memory backend result for stage_candidate: {other:?}" + ))); + } + }; + let id = ids.into_iter().next().ok_or_else(|| { + ToolError::ExecutionFailed( + "stage_candidate backend did not return a staging id".to_string(), + ) + })?; self.state .staged .lock() .expect("session explore staged state poisoned") - .push(written); + .push(id.clone()); Ok(ToolOutput { summary: format!("Staged memory candidate {id}."), content: Some(format!("staging_id: {id}")), @@ -583,14 +608,66 @@ fn truncate_line(text: &str, max_chars: usize) -> String { mod tests { use super::*; use llm_engine::Item; - use tempfile::TempDir; + use std::io::{Read, Write}; + use std::net::TcpListener; + use std::sync::mpsc; + + fn stub_memory_backend_response( + body: &'static str, + ) -> (WorkspaceClient, mpsc::Receiver) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let (tx, rx) = mpsc::channel(); + std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut buffer = Vec::new(); + let mut temp = [0_u8; 1024]; + let header_end = loop { + let read = stream.read(&mut temp).unwrap(); + if read == 0 { + break buffer.len(); + } + buffer.extend_from_slice(&temp[..read]); + if let Some(pos) = buffer.windows(4).position(|window| window == b"\r\n\r\n") { + break pos + 4; + } + }; + let headers = String::from_utf8_lossy(&buffer[..header_end]); + let content_length = headers + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse::().ok())? + }) + .unwrap_or(0); + while buffer.len() < header_end + content_length { + let read = stream.read(&mut temp).unwrap(); + if read == 0 { + break; + } + buffer.extend_from_slice(&temp[..read]); + } + let request = String::from_utf8_lossy(&buffer).into_owned(); + tx.send(request).unwrap(); + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + body.len(), + body + ); + stream.write_all(response.as_bytes()).unwrap(); + }); + ( + WorkspaceClient::http("test-workspace", format!("http://{addr}")), + rx, + ) + } #[test] fn descriptor_declares_session_explore_tools() { - let temp = TempDir::new().unwrap(); let state = SessionExploreState::new( SessionReferenceView::new("segment-1", vec![Item::user_message("remember this")]), - WorkspaceLayout::new(temp.path()), + WorkspaceClient::available("test-backend"), SourceRef { segment_id: "segment-1".to_string(), range: [0, 0], @@ -635,10 +712,12 @@ mod tests { #[tokio::test] async fn stage_candidate_writes_staging_record_with_source_evidence() { - let temp = TempDir::new().unwrap(); + let (client, request_rx) = stub_memory_backend_response( + r#"{"Ok":{"result":{"StagingWritten":{"staging_count":1,"staging_ids":["00000000-0000-7000-8000-000000000001"]}}}}"#, + ); let state = SessionExploreState::new( SessionReferenceView::new("segment-1", vec![Item::user_message("durable decision")]), - WorkspaceLayout::new(temp.path()), + client, SourceRef { segment_id: "segment-1".to_string(), range: [0, 0], @@ -661,13 +740,14 @@ mod tests { .unwrap(); let staged = state.staged(); - assert_eq!(staged.len(), 1); - let bytes = std::fs::read(&staged[0].path).unwrap(); - let record: memory::extract::StagingRecord = serde_json::from_slice(&bytes).unwrap(); - assert_eq!(record.kind, CandidateKind::Decision); - assert_eq!(record.evidence.len(), 1); - assert_eq!(record.evidence[0].id, "M0000"); - assert_eq!(record.source_refs.len(), 1); - assert_eq!(record.source_refs[0].evidence_id.as_deref(), Some("M0000")); + assert_eq!( + staged, + vec!["00000000-0000-7000-8000-000000000001".to_string()] + ); + let request = request_rx.recv().unwrap(); + assert!(request.contains("\"StageCandidate\"")); + assert!(request.contains("\"kind\":\"decision\"")); + assert!(request.contains("\"id\":\"M0000\"")); + assert!(request.contains("\"evidence_id\":\"M0000\"")); } } diff --git a/crates/worker/src/feature/builtin/ticket.rs b/crates/worker/src/feature/builtin/ticket.rs index 2a086cbd..2330f658 100644 --- a/crates/worker/src/feature/builtin/ticket.rs +++ b/crates/worker/src/feature/builtin/ticket.rs @@ -129,6 +129,7 @@ const THREAD_TOOL_NAMES: &[&str] = &["TicketComment", "TicketReview"]; const INTAKE_TOOL_NAMES: &[&str] = &["TicketIntakeReady"]; +#[cfg(test)] const WORKSPACE_AUTHORING_TOOL_NAMES: &[&str] = &[ "TicketCreate", "TicketEditItem", @@ -145,6 +146,7 @@ const WORKSPACE_AUTHORING_TOOL_NAMES: &[&str] = &[ "TicketOrchestrationPlanQuery", ]; +#[cfg(test)] const ORCHESTRATION_CONTROL_TOOL_NAMES: &[&str] = &[ "TicketList", "TicketShow", @@ -172,9 +174,6 @@ pub enum TicketFeatureBackend { Local { root: PathBuf, }, - LocalWorkspace { - workspace_root: PathBuf, - }, WorkspaceHttp { workspace_id: String, base_url: String, @@ -224,9 +223,6 @@ impl TicketFeature { } pub fn with_backend(backend: TicketFeatureBackend, access: TicketFeatureAccess) -> Self { - if let TicketFeatureBackend::LocalWorkspace { workspace_root } = backend { - return Self::for_workspace_with_access(workspace_root, access); - } Self { backend, record_language: None, @@ -266,7 +262,6 @@ impl TicketFeature { pub fn backend_root(&self) -> Option<&Path> { match &self.backend { TicketFeatureBackend::Local { root } => Some(root), - TicketFeatureBackend::LocalWorkspace { workspace_root } => Some(workspace_root), TicketFeatureBackend::WorkspaceHttp { .. } => None, } } @@ -293,8 +288,7 @@ impl TicketFeature { } fn tool_backend(&self, context: &mut FeatureInstallContext<'_>) -> Option { match &self.backend { - TicketFeatureBackend::Local { root: _ } - | TicketFeatureBackend::LocalWorkspace { workspace_root: _ } => { + TicketFeatureBackend::Local { root: _ } => { let usable_root = match self.usable_backend_root() { Ok(root) => root, Err(reason) => { diff --git a/crates/worker/src/lib.rs b/crates/worker/src/lib.rs index 3cf5ed2c..e792863e 100644 --- a/crates/worker/src/lib.rs +++ b/crates/worker/src/lib.rs @@ -20,7 +20,6 @@ pub mod spawn; mod internal_worker; mod interrupt_prep; mod permission; -mod ticket_event_notify; mod worker; pub use compact::token_counter::{EstimateSource, SplitPoint, TokenEstimate}; diff --git a/crates/worker/src/ticket_event_notify.rs b/crates/worker/src/ticket_event_notify.rs deleted file mode 100644 index 65c176b7..00000000 --- a/crates/worker/src/ticket_event_notify.rs +++ /dev/null @@ -1,511 +0,0 @@ -use std::sync::Arc; - -use async_trait::async_trait; -use minijinja::Value as TemplateValue; -use serde_json::Value; -use std::collections::BTreeMap; -use ticket::{LocalTicketBackend, TicketBackend, TicketIdOrSlug}; -use tracing::{debug, warn}; - -use crate::discovery::{WeakNotifyDelivery, WorkerDiscovery}; -use crate::hook::{Hook, HookPostToolAction, PostToolCall, ToolResultSummary}; -use crate::prompt::catalog::{PromptCatalog, WorkerPrompt}; -use session_store::WorkerMetadataStore; - -const MAX_TITLE_CHARS: usize = 96; -const MAX_SUMMARY_CHARS: usize = 160; -const MAX_EVENT_KIND_CHARS: usize = 80; -const MAX_MESSAGE_CHARS: usize = 768; - -#[derive(Clone)] -pub(crate) struct TicketEventCompanionNotifyHook< - St: WorkerMetadataStore + Clone + Send + Sync + 'static, -> { - backend: Arc, - discovery: WorkerDiscovery, - companion_worker_name: String, -} - -impl TicketEventCompanionNotifyHook { - pub(crate) fn new( - backend: LocalTicketBackend, - discovery: WorkerDiscovery, - companion_worker_name: impl Into, - ) -> Self { - Self { - backend: Arc::new(backend), - discovery, - companion_worker_name: companion_worker_name.into(), - } - } -} - -#[async_trait] -impl Hook - for TicketEventCompanionNotifyHook -{ - async fn call(&self, summary: &ToolResultSummary) -> HookPostToolAction { - let Some(notice) = build_ticket_event_notice(&self.backend, summary) else { - return HookPostToolAction::Continue; - }; - match self - .discovery - .ensure_existing_peer(&self.companion_worker_name) - { - Ok(Some(_)) => { - debug!( - ticket = %notice.ticket_id, - event_kind = %notice.event_kind, - companion = %self.companion_worker_name, - "ensured Companion peer relationship before Ticket event notification" - ); - } - Ok(None) => { - debug!( - ticket = %notice.ticket_id, - event_kind = %notice.event_kind, - companion = %self.companion_worker_name, - "skipping Companion peer registration because Companion metadata is missing" - ); - } - Err(error) => { - warn!( - ticket = %notice.ticket_id, - event_kind = %notice.event_kind, - companion = %self.companion_worker_name, - error = %error, - "failed to ensure Companion peer relationship before Ticket event notification" - ); - } - } - let delivery = self - .discovery - .send_weak_notify_to_live_peer(&self.companion_worker_name, notice.message) - .await; - match delivery { - WeakNotifyDelivery::Delivered => { - debug!( - ticket = %notice.ticket_id, - event_kind = %notice.event_kind, - companion = %self.companion_worker_name, - "delivered weak Ticket event notification to Companion peer" - ); - } - skipped => { - warn!( - ticket = %notice.ticket_id, - event_kind = %notice.event_kind, - companion = %self.companion_worker_name, - delivery = %skipped, - "skipped weak Ticket event notification to Companion peer" - ); - } - } - HookPostToolAction::Continue - } -} - -#[derive(Debug, Clone, PartialEq, Eq)] -struct TicketEventNotice { - ticket_id: String, - event_kind: String, - message: String, -} - -fn build_ticket_event_notice( - backend: &LocalTicketBackend, - summary: &ToolResultSummary, -) -> Option { - if summary.is_error { - return None; - } - let output = &summary.output; - let content = output.content.as_deref()?; - let content: Value = serde_json::from_str(content).ok()?; - if !content.get("ok").and_then(Value::as_bool).unwrap_or(false) { - return None; - } - - let event_kind = explicit_ticket_event_kind(summary.tool_name.as_str(), &content)?; - let ticket_query = content.get("ticket").and_then(Value::as_str)?; - let ticket = backend - .show(TicketIdOrSlug::Query(ticket_query.to_string())) - .ok()?; - - let event_kind = sanitize_one_line(&event_kind, MAX_EVENT_KIND_CHARS); - let ticket_id = ticket.meta.id.as_str(); - let title = sanitize_one_line(&ticket.meta.title, MAX_TITLE_CHARS); - let state = ticket.meta.workflow_state.as_str(); - let output_summary = sanitize_one_line(&output.summary, MAX_SUMMARY_CHARS); - let ref_path = event_ref_path(ticket_id, summary.tool_name.as_str()); - let message = render_ticket_event_notice_message(TicketEventNoticeValues { - ticket_id, - title: &title, - state, - event_kind: &event_kind, - summary: &output_summary, - ref_path: &ref_path, - })?; - - Some(TicketEventNotice { - ticket_id: ticket_id.to_string(), - event_kind, - message: bound_chars(&message, MAX_MESSAGE_CHARS), - }) -} - -struct TicketEventNoticeValues<'a> { - ticket_id: &'a str, - title: &'a str, - state: &'a str, - event_kind: &'a str, - summary: &'a str, - ref_path: &'a str, -} - -fn render_ticket_event_notice_message(values: TicketEventNoticeValues<'_>) -> Option { - PromptCatalog::builtins_only() - .ok()? - .render( - WorkerPrompt::TicketEventCompanionNotice, - values.to_template(), - ) - .ok() -} - -impl TicketEventNoticeValues<'_> { - fn to_template(&self) -> TemplateValue { - let mut values: BTreeMap<&'static str, TemplateValue> = BTreeMap::new(); - values.insert("ticket_id", TemplateValue::from(self.ticket_id)); - values.insert("title", TemplateValue::from(self.title)); - values.insert("state", TemplateValue::from(self.state)); - values.insert("event_kind", TemplateValue::from(self.event_kind)); - values.insert("summary", TemplateValue::from(self.summary)); - values.insert("ref_path", TemplateValue::from(self.ref_path)); - TemplateValue::from(values) - } -} - -fn explicit_ticket_event_kind(tool_name: &str, content: &Value) -> Option { - match tool_name { - "TicketComment" => content - .get("event") - .and_then(Value::as_str) - .map(|event| format!("comment/{event}")), - "TicketReview" => content - .get("review") - .and_then(Value::as_str) - .map(|review| format!("review/{review}")), - "TicketWorkflowState" => { - let from = content.get("from").and_then(Value::as_str).unwrap_or("?"); - let to = content.get("to").and_then(Value::as_str).unwrap_or("?"); - Some(format!("state/{from}->{to}")) - } - "TicketIntakeReady" => Some("state/planning->ready".to_string()), - "TicketClose" => Some("close/resolution".to_string()), - _ => None, - } -} - -fn event_ref_path(ticket_id: &str, tool_name: &str) -> String { - let leaf = match tool_name { - "TicketClose" => "resolution.md", - "TicketIntakeReady" | "TicketWorkflowState" => "item.md", - _ => "thread.md", - }; - format!(".yoi/tickets/{ticket_id}/{leaf}") -} - -fn sanitize_one_line(input: &str, limit: usize) -> String { - let collapsed = input.split_whitespace().collect::>().join(" "); - bound_chars(&collapsed, limit) -} - -fn bound_chars(input: &str, limit: usize) -> String { - let mut out = String::new(); - for (idx, ch) in input.chars().filter(|ch| !ch.is_control()).enumerate() { - if idx >= limit { - out.push('…'); - break; - } - out.push(ch); - } - out -} - -pub(crate) fn companion_worker_name_for_workspace( - workspace_root: &std::path::Path, -) -> Option { - workspace_root - .file_name() - .and_then(|name| name.to_str()) - .map(str::trim) - .filter(|name| !name.is_empty()) - .map(ToOwned::to_owned) -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::WorkerStatus; - use crate::runtime::dir::RuntimeDir; - use crate::spawn::registry::SpawnedWorkerRegistry; - use llm_engine::tool::ToolOutput; - use protocol::stream::{JsonLineReader, JsonLineWriter}; - use protocol::{Event, Method}; - use serde_json::json; - use session_store::FsWorkerStore; - use session_store::WorkerMetadata; - use std::sync::Arc; - use tempfile::tempdir; - use ticket::NewTicket; - use tokio::net::UnixListener; - - fn create_backend_with_ticket(title: &str) -> (tempfile::TempDir, LocalTicketBackend, String) { - let dir = tempdir().expect("tempdir"); - let backend = LocalTicketBackend::new(dir.path().to_path_buf()); - let mut input = NewTicket::new(title); - input.body = ticket::MarkdownText::new("body"); - let ticket = backend.create(input).expect("create ticket"); - (dir, backend, ticket.id) - } - - fn tool_summary(tool_name: &str, output: ToolOutput) -> ToolResultSummary { - ToolResultSummary { - call_id: "test-call".to_string(), - tool_name: tool_name.to_string(), - output, - is_error: false, - } - } - - #[test] - fn builds_bounded_event_scoped_notice_for_ticket_state_change() { - let (_dir, backend, ticket_id) = create_backend_with_ticket( - "A very long title that should be bounded but still identify the ticket precisely enough for Companion", - ); - let output = ToolOutput { - summary: "Changed ticket state from queued to inprogress with a deliberately long summary that should be bounded before entering the weak notification payload and should not contain large logs".into(), - content: Some( - json!({ - "ok": true, - "ticket": ticket_id, - "from": "queued", - "to": "inprogress", - }) - .to_string(), - ), - }; - - let notice = - build_ticket_event_notice(&backend, &tool_summary("TicketWorkflowState", output)) - .expect("notice"); - - assert_eq!(notice.ticket_id, ticket_id); - assert_eq!(notice.event_kind, "state/queued->inprogress"); - assert!(notice.message.contains("auto_run=false")); - assert!(notice.message.contains("event: state/queued->inprogress")); - assert!(notice.message.contains("ref: .yoi/tickets/")); - assert!(notice.message.chars().count() <= MAX_MESSAGE_CHARS + 1); - - let expected = PromptCatalog::builtins_only() - .expect("load prompt catalog") - .render( - WorkerPrompt::TicketEventCompanionNotice, - TicketEventNoticeValues { - ticket_id: ¬ice.ticket_id, - title: &sanitize_one_line( - "A very long title that should be bounded but still identify the ticket precisely enough for Companion", - MAX_TITLE_CHARS, - ), - state: "planning", - event_kind: "state/queued->inprogress", - summary: &sanitize_one_line( - "Changed ticket state from queued to inprogress with a deliberately long summary that should be bounded before entering the weak notification payload and should not contain large logs", - MAX_SUMMARY_CHARS, - ), - ref_path: &format!(".yoi/tickets/{}/item.md", ticket_id), - } - .to_template(), - ) - .expect("render prompt resource"); - assert_eq!(notice.message, bound_chars(&expected, MAX_MESSAGE_CHARS)); - } - - #[test] - fn ignores_passive_or_non_event_ticket_tools() { - let (_dir, backend, ticket_id) = create_backend_with_ticket("Passive list test"); - let output = ToolOutput { - summary: "Listed tickets".into(), - content: Some(json!({"ok": true, "ticket": ticket_id}).to_string()), - }; - - assert!(build_ticket_event_notice(&backend, &tool_summary("TicketList", output)).is_none()); - } - - #[test] - fn notice_does_not_include_tool_content_body_or_error_details() { - let (_dir, backend, ticket_id) = create_backend_with_ticket("Safe payload"); - let output = ToolOutput { - summary: "Appended implementation_report to ticket".into(), - content: Some( - json!({ - "ok": true, - "ticket": ticket_id, - "event": "implementation_report", - "body": "SECRET_TOKEN provider stack trace long diagnostic should not be copied", - "error": "provider error details should not be copied" - }) - .to_string(), - ), - }; - - let notice = build_ticket_event_notice(&backend, &tool_summary("TicketComment", output)) - .expect("notice"); - - assert!( - notice - .message - .contains("event: comment/implementation_report") - ); - assert!(!notice.message.contains("SECRET_TOKEN")); - assert!(!notice.message.contains("provider error details")); - } - - #[tokio::test(flavor = "current_thread")] - async fn ticket_event_hook_ensures_peer_and_delivers_weak_companion_notification() { - let root = tempdir().expect("tempdir"); - let runtime_base = root.path().join("runtime"); - let store_dir = root.path().join("store"); - std::fs::create_dir_all(runtime_base.join("companion")).unwrap(); - let store = FsWorkerStore::new(&store_dir).unwrap(); - store - .write(&WorkerMetadata { - worker_name: "orchestrator".into(), - active: None, - workspace_root: None, - workspace_id: None, - spawned_children: Vec::new(), - reclaimed_children: Vec::new(), - peers: Vec::new(), - resolved_manifest_snapshot: None, - }) - .unwrap(); - store - .write(&WorkerMetadata { - worker_name: "companion".into(), - active: None, - workspace_root: None, - workspace_id: None, - spawned_children: Vec::new(), - reclaimed_children: Vec::new(), - peers: Vec::new(), - resolved_manifest_snapshot: None, - }) - .unwrap(); - let (_ticket_dir, backend, ticket_id) = create_backend_with_ticket("Companion event hook"); - let runtime_dir = Arc::new( - RuntimeDir::create(&runtime_base, "orchestrator") - .await - .unwrap(), - ); - let store_for_assert = store.clone(); - let hook = TicketEventCompanionNotifyHook::new( - backend, - WorkerDiscovery::new( - store, - "orchestrator".into(), - runtime_base.clone(), - Some(root.path().to_path_buf()), - SpawnedWorkerRegistry::new(runtime_dir), - ), - "companion", - ); - - let socket = runtime_base.join("companion").join("sock"); - let listener = UnixListener::bind(&socket).unwrap(); - let (tx, mut rx) = tokio::sync::mpsc::channel(1); - let companion = tokio::spawn(async move { - let (stream, _) = listener.accept().await.unwrap(); - let mut writer = JsonLineWriter::new(stream); - writer - .write(&Event::Snapshot { - entries: Vec::new(), - greeting: protocol::Greeting { - worker_name: "companion".into(), - cwd: "/tmp".into(), - provider: "test".into(), - model: "test".into(), - scope_summary: String::new(), - tools: Vec::new(), - context_window: 0, - context_tokens: 0, - }, - status: WorkerStatus::Idle, - in_flight: Default::default(), - }) - .await - .unwrap(); - - let (stream, _) = listener.accept().await.unwrap(); - let (reader_half, writer_half) = stream.into_split(); - let mut reader = JsonLineReader::new(reader_half); - let mut writer = JsonLineWriter::new(writer_half); - writer - .write(&Event::Snapshot { - entries: Vec::new(), - greeting: protocol::Greeting { - worker_name: "companion".into(), - cwd: "/tmp".into(), - provider: "test".into(), - model: "test".into(), - scope_summary: String::new(), - tools: Vec::new(), - context_window: 0, - context_tokens: 0, - }, - status: WorkerStatus::Idle, - in_flight: Default::default(), - }) - .await - .unwrap(); - let method = reader.next::().await.unwrap().unwrap(); - if let Method::Notify { message, auto_run } = method { - assert!(!auto_run); - tx.send(message).await.unwrap(); - } else { - panic!("expected Notify, got {method:?}"); - } - }); - - let output = ToolOutput { - summary: "Changed ticket state from queued to inprogress".into(), - content: Some( - json!({ - "ok": true, - "ticket": ticket_id, - "from": "queued", - "to": "inprogress", - }) - .to_string(), - ), - }; - let action = hook - .call(&tool_summary("TicketWorkflowState", output)) - .await; - assert_eq!(action, HookPostToolAction::Continue); - let message = rx.recv().await.unwrap(); - assert!(message.contains("event: state/queued->inprogress")); - assert!(message.contains("title: Companion event hook")); - let orchestrator = store_for_assert - .read_by_name("orchestrator") - .unwrap() - .unwrap(); - assert_eq!(orchestrator.peers.len(), 1); - assert_eq!(orchestrator.peers[0].worker_name, "companion"); - let companion_metadata = store_for_assert.read_by_name("companion").unwrap().unwrap(); - assert_eq!(companion_metadata.peers.len(), 1); - assert_eq!(companion_metadata.peers[0].worker_name, "orchestrator"); - companion.await.unwrap(); - } -} diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index ed166628..95698ae7 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -30,6 +30,7 @@ use manifest::{ use crate::compact::state::CompactState; use crate::compact::usage_tracker::UsageTracker; +use crate::feature::builtin::memory::WorkspaceMemoryBackendError; use crate::feature::builtin::{ SessionExploreFeature, SessionExploreState, TaskFeature, render_extract_input, }; @@ -506,8 +507,6 @@ pub struct Worker { /// [`Self::from_manifest`], or defaults to the builtin pack when a /// Worker is constructed through lower-level paths that have no loader. prompts: Arc, - /// Memory workspace layout used for Memory record operations. - memory_layout: Option, /// When true (default), the system-prompt assembler may append the /// workspace memory summary (`memory/summary.md`). Internal disposable /// workers disable this so resident memory exposure is opt-in per Worker. @@ -607,7 +606,6 @@ impl Worker usage_history: self.usage_history.clone(), tracker: None, task_feature: self.task_feature.clone(), - memory_layout: self.memory_layout.clone(), system_prompt_template: None, alerter: self.alerter.clone(), event_tx: self.event_tx.clone(), @@ -806,7 +804,6 @@ impl Worker { callback_socket: None, runtime_ticket_role: None, prompts, - memory_layout: None, inject_resident_summary: true, extract_in_flight: Arc::new(AtomicBool::new(false)), consolidation_in_flight: Arc::new(AtomicBool::new(false)), @@ -900,6 +897,20 @@ impl Worker { self.workspace_context.client() } + fn resident_summary_from_workspace_authority(&self) -> Result, WorkerError> { + let result = self.workspace_client().execute_memory_backend_operation( + memory::backend::MemoryBackendOperation::ResidentSummary( + memory::backend::MemoryResidentSummaryOperation::default(), + ), + )?; + match result { + memory::backend::MemoryBackendOperationResult::ToolOutput(output) => Ok(output.content), + other => Err(WorkerError::FeatureInstall(format!( + "unexpected memory backend result for resident summary: {other:?}" + ))), + } + } + /// Activate an Agent Skill through the Workspace backend/client and commit /// the returned SKILL.md body to history before it can influence an LLM run. /// @@ -925,13 +936,6 @@ impl Worker { Ok(activation) } - pub(crate) fn worker_metadata_store(&self) -> St - where - St: Clone, - { - self.store.clone() - } - /// The Worker's directory scope, as a shared atomically-swappable /// handle. Clone it to share scope state with another consumer /// (e.g. a tool that needs to mutate scope dynamically). @@ -1487,17 +1491,20 @@ impl Worker { } } } - let memory_layout = self.memory_layout.as_ref(); let inject_summary = self.inject_resident_summary - && memory_layout.is_some() && self .manifest .memory .as_ref() - .and_then(|m| m.inject_summary) - .unwrap_or(true); + .is_some_and(|m| m.inject_summary.unwrap_or(true)); let resident_summary: Option = if inject_summary { - memory_layout.and_then(memory::collect_resident_summary) + match self.resident_summary_from_workspace_authority() { + Ok(summary) => summary, + Err(error) => { + tracing::debug!(%error, "resident memory summary unavailable"); + None + } + } } else { None }; @@ -2854,15 +2861,6 @@ impl Worker { total_now.saturating_sub(total_at_pointer) } - fn local_memory_layout( - &self, - memory_cfg: &manifest::MemoryConfig, - ) -> Option { - self.filesystem_authority - .as_local() - .map(|local| memory::WorkspaceLayout::resolve(memory_cfg, &local.root)) - } - /// extract (memory.extract) post-run trigger. /// /// Called by the Controller before spawning the background memory task so @@ -2879,10 +2877,6 @@ impl Worker { let Some(memory_cfg) = self.manifest.memory.clone() else { return Ok(()); }; - let Some(layout) = self.local_memory_layout(&memory_cfg) else { - tracing::debug!("workspace memory extract unavailable: no local filesystem authority"); - return Ok(()); - }; // `Some(0)` means disabled, same as `None`. Otherwise the // `tokens_since >= 0` comparison would fire on every post-run. let Some(threshold) = memory_cfg.extract_threshold.filter(|n| *n > 0) else { @@ -2896,7 +2890,7 @@ impl Worker { Some(model_audit_from_manifest(model)), ) .emit( - &layout, + self.workspace_client(), self.event_tx.as_ref(), memory::audit::WorkerLifecycleStatus::Skipped, "extract_threshold_disabled", @@ -2925,7 +2919,7 @@ impl Worker { Some(model_audit_from_manifest(model)), ) .emit( - &layout, + self.workspace_client(), self.event_tx.as_ref(), memory::audit::WorkerLifecycleStatus::Skipped, "extract_already_in_flight", @@ -2970,10 +2964,6 @@ impl Worker { ) -> Result { use memory::extract; - let Some(layout) = self.local_memory_layout(memory_cfg) else { - tracing::debug!("workspace memory extract unavailable: no local filesystem authority"); - return Ok(ExtractDecision::Skipped); - }; let model = memory_cfg .extract_model .as_ref() @@ -2998,7 +2988,7 @@ impl Worker { let tokens_since = self.tokens_added_since(processed_history_len); if tokens_since < threshold { audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Skipped, format!( @@ -3019,7 +3009,7 @@ impl Worker { .len(); if current_history_len <= processed_history_len { audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Skipped, "no_new_history_items", @@ -3042,7 +3032,7 @@ impl Worker { .len(); if entries_now == 0 { audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Skipped, "empty_segment_log", @@ -3059,7 +3049,7 @@ impl Worker { .unwrap_or(0); if start_entry > end_entry { audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Skipped, "no_new_segment_entries", @@ -3084,7 +3074,7 @@ impl Worker { ..Default::default() }; audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Started, format!("token_threshold_reached tokens_since={tokens_since} threshold={threshold}"), @@ -3105,7 +3095,7 @@ impl Worker { Ok(client) => client, Err(err) => { audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Failed, format!("client_build_failed: {err}"), @@ -3121,7 +3111,7 @@ impl Worker { Ok(prompt) => prompt, Err(err) => { audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Failed, format!("prompt_render_failed: {err}"), @@ -3141,7 +3131,8 @@ impl Worker { source_segment_id.to_string(), items_to_extract, ); - let session_explore_state = SessionExploreState::new(session_view, layout.clone(), source); + let session_explore_state = + SessionExploreState::new(session_view, self.workspace_client().clone(), source); let input_text = render_extract_input(session_explore_state.view()); let mut internal_tools = Vec::new(); let mut internal_hook_builder = HookRegistryBuilder::new(); @@ -3161,7 +3152,7 @@ impl Worker { .any(|installed| installed == name) }) { audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Failed, "session_explore_feature_install_failed", @@ -3188,7 +3179,7 @@ impl Worker { Err(err) => { let usage = err.usage.as_ref().map(usage_audit_from_event); audit.emit( - &layout, + self.workspace_client(), event_tx, lifecycle_status_for_worker_error(&err.source), format!("worker_failed: {}", err.source), @@ -3207,10 +3198,7 @@ impl Worker { "extract worker did not call finish_extraction; advancing pointer with staged output" ); } - let staging_id = staging_results - .first() - .map(|result| result.id.to_string()) - .unwrap_or_default(); + let staging_id = staging_results.first().cloned().unwrap_or_default(); let pointer_payload = extract::ExtractPointerPayload { processed_through_entry: end_entry, @@ -3232,11 +3220,8 @@ impl Worker { let mut extract_audit = extract_audit_base; extract_audit.staging_count = staging_results.len(); - for result in &staging_results { - extract_audit.staging_ids.push(result.id.to_string()); - extract_audit - .staging_paths - .push(result.path.display().to_string()); + for id in &staging_results { + extract_audit.staging_ids.push(id.clone()); } let reason = if staging_id.is_empty() { "completed_no_staging_output" @@ -3244,7 +3229,7 @@ impl Worker { "completed_staging_written" }; audit.emit( - &layout, + self.workspace_client(), event_tx, memory::audit::WorkerLifecycleStatus::Completed, reason, @@ -3256,357 +3241,46 @@ impl Worker { Ok(ExtractDecision::Completed) } - /// Build the LlmClient for the consolidation (memory.consolidation) Engine. - /// - /// Uses `memory.consolidation_model` from manifest if set, otherwise - /// clones the main client. Mirrors [`build_extractor_client`]. - fn build_consolidator_client( - &self, - memory_cfg: &manifest::MemoryConfig, - ) -> Result, WorkerError> { - if let Some(ref m) = memory_cfg.consolidation_model { - let client = crate::model_client::build_client(m)?; - return Ok(client); - } - let worker = self.engine.as_ref().expect("worker taken during run"); - Ok(worker.client().clone_boxed()) - } - /// consolidation (memory.consolidation) trigger. /// - /// Intended to run from a background memory task after extract may have - /// added staging entries. Compact is deferred until the next turn starts, - /// so consolidation no longer blocks the controller's post-run path. - /// - /// Behaviour follows `docs/plan/memory.md` §Consolidation / §並走防止: - /// the staging-side `StagingLock` enforces cross-process exclusion; - /// `consolidation_in_flight` keeps in-process callers honest. On - /// success, the lock is released *with* consumed-id cleanup; on - /// worker failure, only the lock file is unlinked so the staging - /// entries remain for a future retry. + /// Worker no longer has direct Workspace filesystem authority. Until consolidation is + /// exposed as a Backend Workspace Authority operation, the Worker must not inspect + /// staging, acquire staging locks, or register local memory tools directly. pub async fn try_post_run_consolidate(&mut self) -> Result<(), WorkerError> { let Some(memory_cfg) = self.manifest.memory.clone() else { return Ok(()); }; - let Some(layout) = self.local_memory_layout(&memory_cfg) else { - tracing::debug!( - "workspace memory consolidation unavailable: no local filesystem authority" - ); - return Ok(()); - }; - // `Some(0)` collapses to `None` — staging count / bytes always - // satisfies `>= 0`, which would fire consolidation on every post-run. - // Treating zero as disabled lines up with `extract_threshold` and - // matches the "no threshold ⇒ consolidation off" invariant in the - // ticket's §Trigger. - let files_threshold = memory_cfg.consolidation_threshold_files.filter(|n| *n > 0); - let bytes_threshold = memory_cfg.consolidation_threshold_bytes.filter(|n| *n > 0); - if files_threshold.is_none() && bytes_threshold.is_none() { - let model = memory_cfg - .consolidation_model - .as_ref() - .unwrap_or(&self.manifest.model); - WorkerAuditBase::new( - memory::audit::AuditWorker::MemoryConsolidation, - memory::audit::AuditTrigger::StagingBacklog, - Some(model_audit_from_manifest(model)), - ) - .emit( - &layout, - self.event_tx.as_ref(), - memory::audit::WorkerLifecycleStatus::Skipped, - "consolidation_threshold_disabled", - None, - None, - None, - ); - return Ok(()); - } - - loop { - if self - .consolidation_in_flight - .compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire) - .is_err() - { - let model = memory_cfg - .consolidation_model - .as_ref() - .unwrap_or(&self.manifest.model); - WorkerAuditBase::new( - memory::audit::AuditWorker::MemoryConsolidation, - memory::audit::AuditTrigger::StagingBacklog, - Some(model_audit_from_manifest(model)), - ) - .emit( - &layout, - self.event_tx.as_ref(), - memory::audit::WorkerLifecycleStatus::Skipped, - "consolidation_already_in_flight", - None, - None, - None, - ); - return Ok(()); - } - let result = self - .run_consolidate_once(&memory_cfg, files_threshold, bytes_threshold) - .await; - self.consolidation_in_flight.store(false, Ordering::Release); - - match result { - Ok(ConsolidateDecision::Skipped) => return Ok(()), - Ok(ConsolidateDecision::Completed) => continue, - Err(e) => { - tracing::warn!(error = %e, "consolidation failed"); - self.alert( - AlertLevel::Warn, - AlertSource::Worker, - format!("memory consolidation failed: {e}"), - ); - return Ok(()); - } - } - } - } - - /// Single consolidation iteration: snapshot staging, decide whether to - /// fire, run the worker if so, release the lock and clean up consumed - /// IDs. - async fn run_consolidate_once( - &mut self, - memory_cfg: &manifest::MemoryConfig, - files_threshold: Option, - bytes_threshold: Option, - ) -> Result { - use memory::consolidate; - - let Some(layout) = self.local_memory_layout(memory_cfg) else { - tracing::debug!( - "workspace memory consolidation unavailable: no local filesystem authority" - ); - return Ok(ConsolidateDecision::Skipped); - }; let model = memory_cfg .consolidation_model .as_ref() .unwrap_or(&self.manifest.model); - let audit = WorkerAuditBase::new( + let files_threshold = memory_cfg.consolidation_threshold_files.filter(|n| *n > 0); + let bytes_threshold = memory_cfg.consolidation_threshold_bytes.filter(|n| *n > 0); + let reason = if files_threshold.is_none() && bytes_threshold.is_none() { + "consolidation_threshold_disabled" + } else { + "consolidation_backend_operation_unavailable" + }; + WorkerAuditBase::new( memory::audit::AuditWorker::MemoryConsolidation, memory::audit::AuditTrigger::StagingBacklog, Some(model_audit_from_manifest(model)), - ); - let event_tx = self.event_tx.as_ref(); - - let staging_snapshot = consolidate::list_staging_entries_snapshot(&layout); - let invalid_staging_count = staging_snapshot.invalid_count; - let entries = staging_snapshot.entries; - if entries.is_empty() { - let reason = if invalid_staging_count == 0 { - "no_staging_entries".to_string() - } else { - format!("no_valid_staging_entries invalid={invalid_staging_count}") - }; - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Skipped, - reason, - None, - None, - Some(memory::audit::ConsolidationAudit { - invalid_staging_count, - ..Default::default() - }), - ); - return Ok(ConsolidateDecision::Skipped); - } - - let total_files = entries.len(); - let total_bytes: u64 = entries.iter().map(|e| e.bytes).sum(); - let consumed_ids: Vec = entries.iter().map(|e| e.id).collect(); - let base_consolidation = memory::audit::ConsolidationAudit { - staging_count: total_files, - invalid_staging_count, - staging_bytes: total_bytes, - consumed_staging_ids: consumed_ids.iter().map(ToString::to_string).collect(), - operations: memory::audit::OperationCounts::default(), - }; - let files_hit = files_threshold.is_some_and(|n| total_files >= n); - let bytes_hit = bytes_threshold.is_some_and(|n| total_bytes >= n); - if !files_hit && !bytes_hit { - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Skipped, - format!( - "threshold_not_reached files={total_files} bytes={total_bytes} files_threshold={files_threshold:?} bytes_threshold={bytes_threshold:?}" - ), - None, - None, - Some(base_consolidation), - ); - return Ok(ConsolidateDecision::Skipped); - } - - let lock = match consolidate::StagingLock::acquire( - &layout, - std::process::id(), - self.manifest.worker.name.clone(), - consumed_ids, - ) { - Ok(l) => l, - Err(memory::consolidate::LockError::InUse { .. }) => { - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Skipped, - "staging_lock_in_use", - None, - None, - Some(base_consolidation), - ); - return Ok(ConsolidateDecision::Skipped); - } - Err(e) => { - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Failed, - format!("staging_lock_failed: {e}"), - None, - None, - Some(base_consolidation), - ); - return Err(WorkerError::ConsolidationLock(e)); - } - }; - - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Started, - format!("staging_threshold_reached files={total_files} bytes={total_bytes}"), + ) + .emit( + self.workspace_client(), + self.event_tx.as_ref(), + memory::audit::WorkerLifecycleStatus::Skipped, + reason, + None, None, None, - Some(base_consolidation.clone()), ); - - let before_records = memory::audit::snapshot_records(&layout); - - let client = match self.build_consolidator_client(memory_cfg) { - Ok(c) => c, - Err(e) => { - lock.release_only(); - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Failed, - format!("client_build_failed: {e}"), - None, - None, - Some(base_consolidation), - ); - return Err(e); - } - }; - let memory_language = memory_language(memory_cfg); - let consolidation_system_prompt = - match self.prompts.memory_consolidation_system(memory_language) { - Ok(p) => p, - Err(e) => { - lock.release_only(); - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Failed, - format!("prompt_render_failed: {e}"), - None, - None, - Some(base_consolidation), - ); - return Err(WorkerError::PromptCatalog(e)); - } - }; - let mut worker = Engine::new(client).system_prompt(consolidation_system_prompt); - worker.set_cache_key(Some(self.segment_id().to_string())); - - let usage_capture = Arc::new(Mutex::new(None)); - let usage_capture_for_worker = usage_capture.clone(); - worker.on_usage(move |event| { - *usage_capture_for_worker - .lock() - .expect("memory consolidation usage capture poisoned") = - Some(usage_audit_from_event(event)); - }); - - // Memory tools are self-contained — they bypass ScopedFs and write - // directly under the workspace via WorkspaceLayout. Resident section - // injection is a Worker-level concern; this disposable Engine is built - // without it by construction, in keeping with `docs/plan/memory.md` - let query_cfg = memory::tool::QueryConfig::from(memory_cfg); - worker.register_tool(memory::tool::read_tool_with_usage( - layout.clone(), - self.segment_id().to_string(), - )); - worker.register_tool(memory::tool::write_tool(layout.clone())); - worker.register_tool(memory::tool::edit_tool(layout.clone())); - worker.register_tool(memory::tool::delete_tool(layout.clone())); - worker.register_tool(memory::tool::memory_query_tool(layout.clone(), query_cfg)); - - let tidy = consolidate::collect_tidy_hints(&layout); - let usage_report = match memory::build_usage_report(&layout) { - Ok(report) => report, - Err(err) => { - warn!(error = %err, "failed to build memory usage report for consolidation"); - memory::UsageReport::empty() - } - }; - let input_text = - consolidate::build_consolidate_input(&layout, &entries, &tidy, &usage_report); - - let run_result = worker.run(input_text).await; - let usage = usage_capture - .lock() - .expect("memory consolidation usage capture poisoned") - .clone(); - match run_result { - Ok(_) => { - lock.release_with_cleanup(&layout); - let after_records = memory::audit::snapshot_records(&layout); - let mut consolidation = base_consolidation; - consolidation.operations = - memory::audit::operation_counts_from_snapshots(&before_records, &after_records); - let reason = if consolidation.operations.total_record_changes() == 0 { - "completed_no_record_changes" - } else { - "completed_record_changes" - }; - audit.emit( - &layout, - event_tx, - memory::audit::WorkerLifecycleStatus::Completed, - reason, - usage, - None, - Some(consolidation), - ); - Ok(ConsolidateDecision::Completed) - } - Err(e) => { - lock.release_only(); - audit.emit( - &layout, - event_tx, - lifecycle_status_for_worker_error(&e), - format!("worker_failed: {e}"), - usage, - None, - Some(base_consolidation), - ); - Err(WorkerError::Engine(e)) - } + if reason == "consolidation_backend_operation_unavailable" { + tracing::debug!( + "workspace memory consolidation skipped: backend operation is unavailable" + ); } + Ok(()) } } @@ -3685,7 +3359,7 @@ impl WorkerAuditBase { fn emit( &self, - layout: &memory::WorkspaceLayout, + workspace_client: &WorkspaceClient, event_tx: Option<&broadcast::Sender>, status: memory::audit::WorkerLifecycleStatus, reason: impl Into, @@ -3694,19 +3368,25 @@ impl WorkerAuditBase { consolidation: Option, ) { let reason = reason.into(); - let _ = memory::audit::append_worker_lifecycle( - layout, - memory::audit::WorkerLifecycleAudit { - run_id: self.run_id, - worker: self.worker, - status, - trigger: self.trigger, - reason: reason.clone(), - model: self.model.clone(), - usage, - extract, - consolidation, - }, + let payload = memory::audit::WorkerLifecycleAudit { + run_id: self.run_id, + worker: self.worker, + status, + trigger: self.trigger, + reason: reason.clone(), + model: self.model.clone(), + usage, + extract, + consolidation, + }; + let _ = workspace_client.execute_memory_backend_operation( + memory::backend::MemoryBackendOperation::AppendAudit( + memory::backend::MemoryAppendAuditOperation { + event: memory::audit::AuditEvent::new( + memory::audit::AuditPayload::WorkerLifecycle(payload), + ), + }, + ), ); if should_emit_memory_worker_event(self.worker, status, &reason) { emit_memory_worker_event( @@ -3766,16 +3446,6 @@ enum ExtractDecision { Completed, } -/// Outcome of a single consolidation iteration. Internal to -/// `try_post_run_consolidate` / `run_consolidate_once`. -enum ConsolidateDecision { - /// Either threshold not met, no staging, or another Worker holds the lock. - Skipped, - /// Consolidation ran. Caller re-evaluates threshold against any - /// staging entries that arrived during the run (Coalesce). - Completed, -} - impl Worker, St> where St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, @@ -3878,7 +3548,6 @@ where callback_socket: None, runtime_ticket_role: None, prompts: common.prompts, - memory_layout: common.memory_layout, inject_resident_summary: true, extract_in_flight: Arc::new(AtomicBool::new(false)), consolidation_in_flight: Arc::new(AtomicBool::new(false)), @@ -3984,7 +3653,6 @@ where callback_socket: Some(callback_socket), runtime_ticket_role: None, prompts: common.prompts, - memory_layout: common.memory_layout, inject_resident_summary: true, extract_in_flight: Arc::new(AtomicBool::new(false)), consolidation_in_flight: Arc::new(AtomicBool::new(false)), @@ -4216,7 +3884,6 @@ where callback_socket: None, runtime_ticket_role: None, prompts: common.prompts, - memory_layout: common.memory_layout, inject_resident_summary: true, extract_in_flight: Arc::new(AtomicBool::new(false)), consolidation_in_flight: Arc::new(AtomicBool::new(false)), @@ -4821,8 +4488,8 @@ pub enum WorkerError { #[error(transparent)] Skill(#[from] SkillClientError), - #[error("memory extract staging write failed: {0}")] - ExtractStaging(#[source] std::io::Error), + #[error(transparent)] + WorkspaceMemoryBackend(#[from] WorkspaceMemoryBackendError), #[error("feature install failed: {0}")] FeatureInstall(String), @@ -4869,7 +4536,6 @@ struct WorkerCommon { delegation_scope: DelegationScope, client: Box, prompts: Arc, - memory_layout: Option, system_prompt_template: Option, } @@ -5011,11 +4677,6 @@ fn prepare_worker_common_from_scope( let client = crate::model_client::build_client(&manifest.model)?; let prompts = PromptCatalog::load(loader, manifest.worker.prompt_pack.as_deref())?; - let memory_layout = manifest.memory.as_ref().and_then(|mem| { - filesystem_authority - .as_local() - .map(|local| memory::WorkspaceLayout::resolve(mem, &local.root)) - }); let system_prompt_template = if parse_template { Some( SystemPromptTemplate::parse(&manifest.engine.instruction, loader.clone()) @@ -5032,7 +4693,6 @@ fn prepare_worker_common_from_scope( delegation_scope, client, prompts, - memory_layout, system_prompt_template, }) } @@ -5089,10 +4749,6 @@ mod spawned_context_tests { common.filesystem_authority.as_local().unwrap().cwd, cwd.canonicalize().unwrap() ); - assert_eq!( - common.memory_layout.as_ref().unwrap().root(), - workspace_root.canonicalize().unwrap() - ); } #[test] @@ -5127,7 +4783,6 @@ mod spawned_context_tests { Some(workspace_id.as_str()) ); assert!(common.workspace_context.client().is_available()); - assert!(common.memory_layout.is_none()); } #[test] @@ -5867,29 +5522,29 @@ mod build_summary_prompt_tests { let store = session_store::FsStore::new(dir.path().join("sessions")).unwrap(); let cwd = dir.path().join("workspace"); std::fs::create_dir_all(&cwd).unwrap(); - if let Some(doc) = summary_doc { - std::fs::create_dir_all(cwd.join(".yoi/memory")).unwrap(); - std::fs::write(cwd.join(".yoi/memory/summary.md"), doc).unwrap(); - } let mut manifest = minimal_manifest(); - manifest.memory = memory_config; + manifest.memory = memory_config.clone(); let scope = Scope::writable(&cwd).unwrap(); let authority = WorkerFilesystemAuthority::local(cwd.clone(), cwd.clone()); + let workspace_context = if memory_config + .as_ref() + .is_some_and(|cfg| cfg.inject_summary.unwrap_or(true)) + && gates.summary + { + stub_memory_backend_context(summary_doc.and_then(summary_content_for_backend)) + } else { + WorkerWorkspaceContext::local_filesystem(None) + }; let mut worker = Worker::new( manifest, Engine::new(NoopClient), store, - WorkerWorkspaceContext::local_filesystem(None), + workspace_context, authority, scope, ) .await .unwrap(); - worker.memory_layout = worker - .manifest - .memory - .as_ref() - .map(|mem| memory::WorkspaceLayout::resolve(mem, &cwd)); worker.set_resident_memory_injection(gates.summary); let template = SystemPromptTemplate::parse( "$yoi/default", @@ -5905,6 +5560,56 @@ mod build_summary_prompt_tests { format!("---\nupdated_at: 2026-01-01T00:00:00Z\n---\n{body}") } + fn summary_content_for_backend(doc: &str) -> Option { + if doc.contains("this is not yaml") { + return None; + } + if let Some(rest) = doc.strip_prefix("---\n") { + if let Some((_, body)) = rest.split_once("\n---\n") { + return Some(body.to_string()); + } + } + Some(doc.to_string()) + } + + fn stub_memory_backend_context(content: Option) -> WorkerWorkspaceContext { + use std::io::{Read, Write}; + use std::net::TcpListener; + + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + std::thread::spawn(move || { + let (mut stream, _) = listener.accept().unwrap(); + let mut buffer = [0_u8; 1024]; + let _ = stream.read(&mut buffer).unwrap(); + let body = serde_json::json!({ + "Ok": { + "result": { + "ToolOutput": { + "summary": if content.is_some() { + "resident memory summary collected" + } else { + "resident memory summary unavailable" + }, + "content": content, + } + } + } + }) + .to_string(); + let response = format!( + "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}", + body.len(), + body + ); + stream.write_all(response.as_bytes()).unwrap(); + }); + WorkerWorkspaceContext::with_client( + Some(WorkspaceId::new("test-memory").unwrap()), + WorkspaceClient::http("test-memory", format!("http://{addr}")), + ) + } + #[tokio::test] async fn resident_summary_body_is_injected_without_frontmatter() { let rendered = render_system_prompt_with_summary(