Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

feat: route worker workspace access through backend authority

Browse Source
This commit is contained in:
Keisuke Hirata 2026-07-21 11:21:13 +09:00
parent 2572dde691
commit 1251edae04
No known key found for this signature in database
10 changed files with 459 additions and 1206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

75
crates/memory/src/backend.rs
View File

@ -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<String>,
}
#[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<StagingEvidence>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub source_refs: Vec<SourceEvidenceRef>,
}
#[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,

57
crates/worker-runtime/src/worker_backend.rs
View File

@ -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<WorkspaceId> {
let contents = std::fs::read_to_string(root.join(".yoi/workspace.toml")).ok()?;
let value = toml::from_str::<toml::Value>(&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]

132
crates/worker/src/controller.rs
View File

@ -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<C, St>(
// per-item commit channel is wired at the top of this function.
}
fn install_ticket_event_companion_notify_hook<C, St>(
worker: &mut Worker<C, St>,
runtime_base: PathBuf,
spawned_registry: Arc<SpawnedWorkerRegistry>,
) 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(
return Err(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(),
}
"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(|| {
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 {
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",
));
}
}

25
crates/worker/src/entrypoint.rs
View File

@ -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()))
}

95
crates/worker/src/feature/builtin/memory.rs
View File

@ -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,39 +34,88 @@ impl WorkspaceHttpMemoryBackend {
}
}
pub fn execute_operation(
&self,
operation: MemoryBackendOperation,
) -> Result<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
execute_http_memory_backend(&self.workspace_id, &self.base_url, operation)
}
fn execute(&self, operation: MemoryBackendOperation) -> Result<ToolOutput, ToolError> {
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:?}"
))),
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<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
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<MemoryBackendOperationResult, WorkspaceMemoryBackendError> {
let url = format!(
"{}/api/w/{}/memory/backend",
self.base_url.trim_end_matches('/'),
self.workspace_id
base_url.trim_end_matches('/'),
workspace_id
);
let response = reqwest::blocking::Client::new()
.post(url)
.json(&operation)
.send()
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
.send()?;
let status = response.status();
let body = response
.text()
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?;
let body = response.text()?;
if !status.is_success() {
return Err(ToolError::ExecutionFailed(format!(
"workspace memory backend returned HTTP {status}: {body}"
)));
return Err(WorkspaceMemoryBackendError::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!(
"unexpected memory backend result for model-visible tool: {result:?}"
))),
match serde_json::from_str::<MemoryBackendHttpResponse>(&body)? {
MemoryBackendHttpResponse::Ok { result } => Ok(result),
MemoryBackendHttpResponse::Error { message } => {
Err(ToolError::ExecutionFailed(message))
}
Err(WorkspaceMemoryBackendError::Backend(message))
}
}
}

136
crates/worker/src/feature/builtin/session_explore.rs
View File

@ -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<SessionReferenceView>,
layout: WorkspaceLayout,
workspace_client: WorkspaceClient,
source: SourceRef,
extract_run_id: String,
staged: Arc<Mutex<Vec<StagingWriteResult>>>,
staged: Arc<Mutex<Vec<String>>>,
finished: Arc<Mutex<Option<FinishExtractionParams>>>,
}
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<StagingWriteResult> {
pub(crate) fn staged(&self) -> Vec<String> {
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,
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 id = written.id.to_string();
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<String>) {
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::<usize>().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\""));
}
}

12
crates/worker/src/feature/builtin/ticket.rs
View File

@ -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<TicketToolBackend> {
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) => {

1
crates/worker/src/lib.rs
View File

@ -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};

511
crates/worker/src/ticket_event_notify.rs
View File

@ -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<LocalTicketBackend>,
discovery: WorkerDiscovery<St>,
companion_worker_name: String,
}
impl<St: WorkerMetadataStore + Clone + Send + Sync + 'static> TicketEventCompanionNotifyHook<St> {
pub(crate) fn new(
backend: LocalTicketBackend,
discovery: WorkerDiscovery<St>,
companion_worker_name: impl Into<String>,
) -> Self {
Self {
backend: Arc::new(backend),
discovery,
companion_worker_name: companion_worker_name.into(),
}
}
}
#[async_trait]
impl<St: WorkerMetadataStore + Clone + Send + Sync + 'static> Hook<PostToolCall>
for TicketEventCompanionNotifyHook<St>
{
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<TicketEventNotice> {
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<String> {
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<String> {
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::<Vec<_>>().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<String> {
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: &notice.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::<Method>().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();
}
}

555
crates/worker/src/worker.rs
View File

@ -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<C: LlmClient, St: Store> {
/// [`Self::from_manifest`], or defaults to the builtin pack when a
/// Worker is constructed through lower-level paths that have no loader.
prompts: Arc<PromptCatalog>,
/// Memory workspace layout used for Memory record operations.
memory_layout: Option<memory::WorkspaceLayout>,
/// 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<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Worker<C, St>
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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
self.workspace_context.client()
}
fn resident_summary_from_workspace_authority(&self) -> Result<Option<String>, 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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
}
}
}
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<String> = 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<C: LlmClient, St: Store> Worker<C, St> {
total_now.saturating_sub(total_at_pointer)
}
fn local_memory_layout(
&self,
memory_cfg: &manifest::MemoryConfig,
) -> Option<memory::WorkspaceLayout> {
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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
) -> Result<ExtractDecision, WorkerError> {
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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
.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<C: LlmClient, St: Store> Worker<C, St> {
.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<C: LlmClient, St: Store> Worker<C, St> {
.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<C: LlmClient, St: Store> Worker<C, St> {
..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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
.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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
"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<C: LlmClient, St: Store> Worker<C, St> {
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<C: LlmClient, St: Store> Worker<C, St> {
"completed_staging_written"
};
audit.emit(
&layout,
self.workspace_client(),
event_tx,
memory::audit::WorkerLifecycleStatus::Completed,
reason,
@ -3256,357 +3241,46 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
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<Box<dyn LlmClient>, 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 model = memory_cfg
.consolidation_model
.as_ref()
.unwrap_or(&self.manifest.model);
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);
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)),
)
.emit(
&layout,
self.workspace_client(),
self.event_tx.as_ref(),
memory::audit::WorkerLifecycleStatus::Skipped,
"consolidation_threshold_disabled",
reason,
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<usize>,
bytes_threshold: Option<u64>,
) -> Result<ConsolidateDecision, WorkerError> {
use memory::consolidate;
let Some(layout) = self.local_memory_layout(memory_cfg) else {
if reason == "consolidation_backend_operation_unavailable" {
tracing::debug!(
"workspace memory consolidation unavailable: no local filesystem authority"
"workspace memory consolidation skipped: backend operation is unavailable"
);
return Ok(ConsolidateDecision::Skipped);
};
let model = memory_cfg
.consolidation_model
.as_ref()
.unwrap_or(&self.manifest.model);
let audit = 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<uuid::Uuid> = 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}"),
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))
}
}
Ok(())
}
}
@ -3685,7 +3359,7 @@ impl WorkerAuditBase {
fn emit(
&self,
layout: &memory::WorkspaceLayout,
workspace_client: &WorkspaceClient,
event_tx: Option<&broadcast::Sender<Event>>,
status: memory::audit::WorkerLifecycleStatus,
reason: impl Into<String>,
@ -3694,9 +3368,7 @@ impl WorkerAuditBase {
consolidation: Option<memory::audit::ConsolidationAudit>,
) {
let reason = reason.into();
let _ = memory::audit::append_worker_lifecycle(
layout,
memory::audit::WorkerLifecycleAudit {
let payload = memory::audit::WorkerLifecycleAudit {
run_id: self.run_id,
worker: self.worker,
status,
@ -3706,7 +3378,15 @@ impl WorkerAuditBase {
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<St> Worker<Box<dyn LlmClient>, 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<dyn LlmClient>,
prompts: Arc<PromptCatalog>,
memory_layout: Option<memory::WorkspaceLayout>,
system_prompt_template: Option<SystemPromptTemplate>,
}
@ -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<String> {
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<String>) -> 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(