From 05e8b00bf0ed4864c028e8423c4e7815b909e6cc Mon Sep 17 00:00:00 2001 From: Hare Date: Mon, 3 Aug 2026 18:12:47 +0900 Subject: [PATCH] workdir: separate identity from worker session --- crates/manifest/src/scope.rs | 4 +- crates/memory/src/scope.rs | 4 +- crates/tools/src/bash.rs | 22 +- crates/tools/src/edit.rs | 16 +- crates/tools/src/error.rs | 8 +- crates/tools/src/glob.rs | 10 +- crates/tools/src/grep.rs | 12 +- crates/tools/src/lib.rs | 47 +-- crates/tools/src/read.rs | 18 +- crates/tools/src/tracker.rs | 8 +- crates/tools/src/write.rs | 20 +- crates/tools/tests/edge_cases.rs | 5 +- crates/tools/tests/integration.rs | 18 +- crates/workdir/src/lib.rs | 110 ++++--- crates/workdir/src/local.rs | 306 ++++++++++++++------ crates/worker-runtime/src/worker_backend.rs | 32 +- crates/worker/src/compact/worker.rs | 22 +- crates/worker/src/controller.rs | 42 ++- crates/worker/src/fs_view.rs | 46 +-- crates/worker/src/shared_state.rs | 2 +- crates/worker/src/worker.rs | 50 ++-- crates/worker/tests/controller_test.rs | 53 +++- 22 files changed, 541 insertions(+), 314 deletions(-) diff --git a/crates/manifest/src/scope.rs b/crates/manifest/src/scope.rs index 426eda29..fad12dea 100644 --- a/crates/manifest/src/scope.rs +++ b/crates/manifest/src/scope.rs @@ -422,12 +422,12 @@ impl Scope { /// Shared, atomically-swappable view of a [`Scope`]. /// /// Built around [`ArcSwap`] so the hot path (permission checks inside a local -/// Workdir provider) reads the current scope lock-free. Mutators are +/// WorkdirSession provider) reads the current scope lock-free. Mutators are /// serialised by an internal `Mutex` so concurrent `update` calls do /// not lose each other's contributions. /// /// All clones share the same underlying state — a `SharedScope` cloned -/// out to multiple consumers (Worker, local Workdir providers, future +/// out to multiple consumers (Worker, local WorkdirSession providers, future /// grant/revoke callers) sees every update. #[derive(Debug, Clone)] pub struct SharedScope { diff --git a/crates/memory/src/scope.rs b/crates/memory/src/scope.rs index 54baae11..4b5326ec 100644 --- a/crates/memory/src/scope.rs +++ b/crates/memory/src/scope.rs @@ -3,8 +3,8 @@ //! //! Worker is expected to call [`deny_write_rules`] when memory is enabled //! and append the result to the manifest's `scope.deny` list before -//! constructing the [`Scope`] passed to the local Workdir provider. The -//! memory tools themselves bypass generic Workdir filesystem operations and +//! constructing the [`Scope`] passed to the local WorkdirSession provider. The +//! memory tools themselves bypass generic WorkdirSession filesystem operations and //! write directly under the workspace root, so this deny does not affect them. use std::path::Path; diff --git a/crates/tools/src/bash.rs b/crates/tools/src/bash.rs index 70bde8fc..34b35b76 100644 --- a/crates/tools/src/bash.rs +++ b/crates/tools/src/bash.rs @@ -5,7 +5,7 @@ use async_trait::async_trait; use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use schemars::JsonSchema; use serde::Deserialize; -use workdir::{CommandHandle, CommandOutputRequest, CommandRequest, WorkdirHandle}; +use workdir::{CommandHandle, CommandOutputRequest, CommandRequest, WorkdirSessionHandle}; const DEFAULT_TIMEOUT_SECS: u64 = 120; const MAX_TIMEOUT_SECS: u64 = 600; @@ -19,18 +19,18 @@ struct BashParams { } pub(crate) struct BashTool { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, } struct CommandGuard { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, handle: Option, } impl Drop for CommandGuard { fn drop(&mut self) { if let Some(handle) = self.handle.take() { - let workdir = self.workdir.clone(); + let workdir = self.session.clone(); tokio::spawn(async move { let _ = workdir.cancel_command(handle).await; }); @@ -53,7 +53,7 @@ impl Tool for BashTool { .clamp(1, MAX_TIMEOUT_SECS); let cmd_summary = truncate_for_summary(¶ms.command); let handle = self - .workdir + .session .start_command(CommandRequest { command: params.command, timeout_secs, @@ -62,11 +62,11 @@ impl Tool for BashTool { .await .map_err(crate::ToolsError::from)?; let mut guard = CommandGuard { - workdir: self.workdir.clone(), + session: self.session.clone(), handle: Some(handle.clone()), }; let output = self - .workdir + .session .command_output(CommandOutputRequest { handle, cursor: 0, @@ -90,7 +90,7 @@ impl Tool for BashTool { None } else if output.truncated { Some(format!( - "[showing bounded Workdir command output; additional output was truncated]\n{}", + "[showing bounded WorkdirSession command output; additional output was truncated]\n{}", output.content )) } else { @@ -110,14 +110,14 @@ fn truncate_for_summary(command: &str) -> String { summary } -pub fn bash_tool(workdir: WorkdirHandle, _output_dir: PathBuf) -> ToolDefinition { +pub fn bash_tool(session: WorkdirSessionHandle, _output_dir: PathBuf) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(BashParams); let meta = ToolMeta::new("Bash") - .description("Execute a shell command in the bound Workdir. Process start, bounded output, timeout and cancellation are owned by the Workdir provider. This is not a sandbox.") + .description("Execute a shell command in the bound Workdir. Process start, bounded output, timeout and cancellation are owned by the WorkdirSession provider. This is not a sandbox.") .input_schema(serde_json::to_value(schema).expect("Bash schema serialization")); let tool: Arc = Arc::new(BashTool { - workdir: workdir.clone(), + session: session.clone(), }); (meta, tool) }) diff --git a/crates/tools/src/edit.rs b/crates/tools/src/edit.rs index b05a4080..317db17b 100644 --- a/crates/tools/src/edit.rs +++ b/crates/tools/src/edit.rs @@ -9,7 +9,7 @@ use serde::Deserialize; use crate::error::ToolsError; use crate::tracker::Tracker; -use workdir::{EditRequest, WorkdirHandle, WorkdirPath}; +use workdir::{EditRequest, WorkdirPath, WorkdirSessionHandle}; const DESCRIPTION: &str = "Replace a substring in an existing file. By default \ `old_string` must be unique in the file; set `replace_all: true` to replace \ @@ -30,7 +30,7 @@ pub(crate) struct EditParams { } pub(crate) struct EditTool { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, tracker: Tracker, } @@ -62,7 +62,7 @@ impl Tool for EditTool { let _mutation_permit = self.tracker.acquire_mutation(&mutation_key, &ctx).await; let expected_hash = self.tracker.expected_workdir_hash(&path)?; let result = self - .workdir + .session .edit(EditRequest { path: path.clone(), old_string: params.old_string.clone(), @@ -115,7 +115,7 @@ fn make_preview(text: &str, needle: &str) -> String { } /// Factory for the `Edit` tool. -pub fn edit_tool(workdir: WorkdirHandle, tracker: Tracker) -> ToolDefinition { +pub fn edit_tool(session: WorkdirSessionHandle, tracker: Tracker) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(EditParams); let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); @@ -123,7 +123,7 @@ pub fn edit_tool(workdir: WorkdirHandle, tracker: Tracker) -> ToolDefinition { .description(DESCRIPTION) .input_schema(schema_value); let tool: Arc = Arc::new(EditTool { - workdir: workdir.clone(), + session: session.clone(), tracker: tracker.clone(), }); (meta, tool) @@ -137,16 +137,16 @@ mod tests { use manifest::Scope; use tempfile::TempDir; - fn setup() -> (TempDir, WorkdirHandle, Tracker) { + fn setup() -> (TempDir, WorkdirSessionHandle, Tracker) { let dir = TempDir::new().unwrap(); - let fs: WorkdirHandle = Arc::new(workdir::LocalWorkdir::new( + let fs: WorkdirSessionHandle = Arc::new(workdir::LocalWorkdirSession::new( Scope::writable(dir.path()).unwrap(), dir.path().to_path_buf(), )); (dir, fs, Tracker::new()) } - async fn read_first(fs: &WorkdirHandle, tracker: &Tracker, file: &std::path::Path) { + async fn read_first(fs: &WorkdirSessionHandle, tracker: &Tracker, file: &std::path::Path) { let def = read_tool(fs.clone(), tracker.clone()); let (_, reader) = def(); let inp = serde_json::json!({ "file_path": file.file_name().unwrap().to_str().unwrap() }); diff --git a/crates/tools/src/error.rs b/crates/tools/src/error.rs index 0c476b04..e6f5f84d 100644 --- a/crates/tools/src/error.rs +++ b/crates/tools/src/error.rs @@ -1,6 +1,6 @@ //! Error types for builtin tools. //! -//! `ToolsError` keeps tool-specific policy failures separate from Workdir +//! `ToolsError` keeps tool-specific policy failures separate from WorkdirSession //! operation failures. Filesystem, search, and command errors originate in //! `workdir` and remain transparent here. @@ -11,7 +11,7 @@ use llm_engine::tool::ToolError; #[derive(Debug, thiserror::Error)] pub enum ToolsError { #[error(transparent)] - Workdir(#[from] workdir::WorkdirError), + WorkdirSession(#[from] workdir::WorkdirError), #[error("file has not been read in this session; read it first: {}", .0.display())] NotRead(PathBuf), @@ -35,12 +35,12 @@ pub enum ToolsError { impl From for ToolError { fn from(err: ToolsError) -> Self { match &err { - ToolsError::Workdir( + ToolsError::WorkdirSession( workdir::WorkdirError::NotFound(_) | workdir::WorkdirError::Io { .. } | workdir::WorkdirError::Unavailable(_), ) => ToolError::ExecutionFailed(err.to_string()), - ToolsError::Workdir(_) + ToolsError::WorkdirSession(_) | ToolsError::NotRead(_) | ToolsError::ExternallyModified(_) | ToolsError::StringNotFound { .. } diff --git a/crates/tools/src/glob.rs b/crates/tools/src/glob.rs index 6dcc1196..8f2a5942 100644 --- a/crates/tools/src/glob.rs +++ b/crates/tools/src/glob.rs @@ -4,7 +4,7 @@ use async_trait::async_trait; use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use schemars::JsonSchema; use serde::Deserialize; -use workdir::{GlobRequest, WorkdirHandle, WorkdirPath}; +use workdir::{GlobRequest, WorkdirPath, WorkdirSessionHandle}; use crate::ToolsError; @@ -20,7 +20,7 @@ struct GlobParams { } struct GlobTool { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, } #[async_trait] @@ -39,7 +39,7 @@ impl Tool for GlobTool { let pattern = params.pattern; tracing::debug!(%pattern, %path, "Glob"); let result = self - .workdir + .session .glob(GlobRequest { pattern: pattern.clone(), path, @@ -73,14 +73,14 @@ impl Tool for GlobTool { } } -pub fn glob_tool(workdir: WorkdirHandle) -> ToolDefinition { +pub fn glob_tool(session: WorkdirSessionHandle) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(GlobParams); let meta = ToolMeta::new("Glob") .description("Find files matching a glob pattern inside the bound Workdir. Results are sorted and capped at 1000 entries. Paths are Workdir-relative.") .input_schema(serde_json::to_value(schema).expect("Glob schema serialization")); let tool: Arc = Arc::new(GlobTool { - workdir: workdir.clone(), + session: session.clone(), }); (meta, tool) }) diff --git a/crates/tools/src/grep.rs b/crates/tools/src/grep.rs index 32e9ec67..2119e135 100644 --- a/crates/tools/src/grep.rs +++ b/crates/tools/src/grep.rs @@ -4,7 +4,7 @@ use async_trait::async_trait; use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use schemars::JsonSchema; use serde::Deserialize; -use workdir::{GrepOutputMode, GrepRequest, WorkdirHandle, WorkdirPath}; +use workdir::{GrepOutputMode, GrepRequest, WorkdirPath, WorkdirSessionHandle}; use crate::ToolsError; @@ -48,7 +48,7 @@ struct GrepParams { } struct GrepTool { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, } #[async_trait] @@ -75,7 +75,7 @@ impl Tool for GrepTool { .unwrap_or((params.before.unwrap_or(0), params.after.unwrap_or(0))); let head_limit = params.head_limit.unwrap_or(DEFAULT_HEAD_LIMIT); let result = self - .workdir + .session .grep(GrepRequest { pattern: params.pattern, path, @@ -124,14 +124,14 @@ impl Tool for GrepTool { } } -pub fn grep_tool(workdir: WorkdirHandle) -> ToolDefinition { +pub fn grep_tool(session: WorkdirSessionHandle) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(GrepParams); let meta = ToolMeta::new("Grep") - .description("Search Workdir file contents with a regex. Glob/Grep traversal executes inside the Workdir provider. Results are bounded and Workdir-relative.") + .description("Search Workdir file contents with a regex. Glob/Grep traversal executes inside the WorkdirSession provider. Results are bounded and Workdir-relative.") .input_schema(serde_json::to_value(schema).expect("Grep schema serialization")); let tool: Arc = Arc::new(GrepTool { - workdir: workdir.clone(), + session: session.clone(), }); (meta, tool) }) diff --git a/crates/tools/src/lib.rs b/crates/tools/src/lib.rs index a708990e..5dd3ae42 100644 --- a/crates/tools/src/lib.rs +++ b/crates/tools/src/lib.rs @@ -1,10 +1,11 @@ //! Built-in tools for the Yoi LLM agent. //! //! Read / Write / Edit / Glob / Grep / Bash operate through a host-owned -//! [`workdir::Workdir`] handle. This crate owns tool schemas, rendering, and -//! read-before-edit tracking; it does not own Workdir identity or lifecycle. +//! [`workdir::WorkdirSession`] handle. This crate owns tool schemas, rendering, and +//! read-before-edit tracking; it does not own Workdir identity/materialization or +//! WorkdirSession lifecycle. //! -//! Bash is intentionally not sandboxed. The Workdir supplies its initial cwd +//! Bash is intentionally not sandboxed. The WorkdirSession supplies its initial cwd //! and command capability, while the Runtime process and OS user remain the //! trusted execution boundary. @@ -29,34 +30,34 @@ pub use tracker::Tracker; pub use web::{web_fetch_tool, web_search_tool}; pub use write::write_tool; -/// Build the local filesystem/command tool surface implemented by a Workdir. +/// Build the local filesystem/command tool surface implemented by a WorkdirSession. /// Profile/manifest policy may narrow this set further in the Engine. pub fn core_builtin_tools( - workdir: workdir::WorkdirHandle, + session: workdir::WorkdirSessionHandle, tracker: Tracker, bash_output_dir: std::path::PathBuf, ) -> Vec { - use workdir::WorkdirCapability; + use workdir::WorkdirSessionCapability; - let capabilities = workdir.capabilities(); + let capabilities = session.capabilities(); let mut tools = Vec::with_capacity(6); - if capabilities.supports(WorkdirCapability::Read) { - tools.push(read_tool(workdir.clone(), tracker.clone())); + if capabilities.supports(WorkdirSessionCapability::Read) { + tools.push(read_tool(session.clone(), tracker.clone())); } - if capabilities.supports(WorkdirCapability::Write) { - tools.push(write_tool(workdir.clone(), tracker.clone())); + if capabilities.supports(WorkdirSessionCapability::Write) { + tools.push(write_tool(session.clone(), tracker.clone())); } - if capabilities.supports(WorkdirCapability::Edit) { - tools.push(edit_tool(workdir.clone(), tracker)); + if capabilities.supports(WorkdirSessionCapability::Edit) { + tools.push(edit_tool(session.clone(), tracker)); } - if capabilities.supports(WorkdirCapability::Glob) { - tools.push(glob_tool(workdir.clone())); + if capabilities.supports(WorkdirSessionCapability::Glob) { + tools.push(glob_tool(session.clone())); } - if capabilities.supports(WorkdirCapability::Grep) { - tools.push(grep_tool(workdir.clone())); + if capabilities.supports(WorkdirSessionCapability::Grep) { + tools.push(grep_tool(session.clone())); } - if capabilities.supports(WorkdirCapability::Command) { - tools.push(bash_tool(workdir, bash_output_dir)); + if capabilities.supports(WorkdirSessionCapability::Command) { + tools.push(bash_tool(session, bash_output_dir)); } tools } @@ -76,18 +77,18 @@ mod workdir_tool_tests { use manifest::{Scope, SharedScope}; use std::sync::Arc; use tempfile::TempDir; - use workdir::{LocalWorkdir, WorkdirCapabilities, WorkdirHandle}; + use workdir::{LocalWorkdirSession, WorkdirSessionCapabilities, WorkdirSessionHandle}; #[test] fn read_only_workdir_exposes_only_observation_tools() { let dir = TempDir::new().unwrap(); - let workdir: WorkdirHandle = Arc::new(LocalWorkdir::materialized( + let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized( dir.path().to_path_buf(), dir.path().to_path_buf(), SharedScope::new(Scope::writable(dir.path()).unwrap()), - WorkdirCapabilities::READ_ONLY, + WorkdirSessionCapabilities::READ_ONLY, )); - let names = core_builtin_tools(workdir, Tracker::new(), dir.path().join("output")) + let names = core_builtin_tools(session, Tracker::new(), dir.path().join("output")) .into_iter() .map(|definition| definition().0.name) .collect::>(); diff --git a/crates/tools/src/read.rs b/crates/tools/src/read.rs index e6c13051..acbd40b9 100644 --- a/crates/tools/src/read.rs +++ b/crates/tools/src/read.rs @@ -8,7 +8,7 @@ use serde::Deserialize; use crate::error::ToolsError; use crate::tracker::Tracker; -use workdir::{ReadRequest, WorkdirHandle, WorkdirPath}; +use workdir::{ReadRequest, WorkdirPath, WorkdirSessionHandle}; const DESCRIPTION: &str = "Read a text file from the local filesystem. \ Supports offset/limit for large files. Returns line-numbered output (1-based). \ @@ -31,7 +31,7 @@ pub(crate) struct ReadParams { } pub(crate) struct ReadTool { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, tracker: Tracker, } @@ -51,7 +51,7 @@ impl Tool for ReadTool { tracing::debug!(path = %path, offset, limit, "Read"); let result = self - .workdir + .session .read(ReadRequest { path: path.clone(), offset, @@ -144,7 +144,7 @@ fn render_numbered(text: &str, offset: usize, limit: usize) -> Rendered { } /// Factory for the `Read` tool. -pub fn read_tool(workdir: WorkdirHandle, tracker: Tracker) -> ToolDefinition { +pub fn read_tool(session: WorkdirSessionHandle, tracker: Tracker) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(ReadParams); let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); @@ -152,7 +152,7 @@ pub fn read_tool(workdir: WorkdirHandle, tracker: Tracker) -> ToolDefinition { .description(DESCRIPTION) .input_schema(schema_value); let tool: Arc = Arc::new(ReadTool { - workdir: workdir.clone(), + session: session.clone(), tracker: tracker.clone(), }); (meta, tool) @@ -164,15 +164,15 @@ mod tests { use super::*; use manifest::Scope; use tempfile::TempDir; - use workdir::LocalWorkdir; + use workdir::LocalWorkdirSession; - fn setup() -> (TempDir, WorkdirHandle, Tracker) { + fn setup() -> (TempDir, WorkdirSessionHandle, Tracker) { let dir = TempDir::new().unwrap(); - let workdir: WorkdirHandle = Arc::new(LocalWorkdir::new( + let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::new( Scope::writable(dir.path()).unwrap(), dir.path().to_path_buf(), )); - (dir, workdir, Tracker::new()) + (dir, session, Tracker::new()) } #[tokio::test] diff --git a/crates/tools/src/tracker.rs b/crates/tools/src/tracker.rs index d875cad5..40a771be 100644 --- a/crates/tools/src/tracker.rs +++ b/crates/tools/src/tracker.rs @@ -21,7 +21,7 @@ //! A `Tracker` is **Worker-process scoped**: the Worker layer creates a fresh //! instance at the start of each Worker run (including resume) and discards //! it when the process exits — it is not persisted, so a resumed -//! conversation starts with an empty read/edit history. The local Workdir +//! conversation starts with an empty read/edit history. The local WorkdirSession //! scope boundary is likewise Worker-process scoped (derived from the //! manifest). The two are orthogonal and the Worker wires them together //! when registering builtin tools. @@ -31,15 +31,15 @@ //! # use std::sync::Arc; //! # use manifest::Scope; //! # use tools::{Tracker, core_builtin_tools}; -//! # use workdir::{LocalWorkdir, WorkdirHandle}; +//! # use workdir::{LocalWorkdirSession, WorkdirSessionHandle}; //! let scope = Scope::writable("/workspace").unwrap(); -//! let workdir: WorkdirHandle = Arc::new(LocalWorkdir::new( +//! let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::new( //! scope, //! PathBuf::from("/workspace"), //! )); //! let tracker = Tracker::new(); // session lifetime //! let bash_outputs = PathBuf::from("/run/yoi/bash-output"); -//! let defs = core_builtin_tools(workdir, tracker, bash_outputs); +//! let defs = core_builtin_tools(session, tracker, bash_outputs); //! ``` use std::collections::{HashMap, VecDeque}; diff --git a/crates/tools/src/write.rs b/crates/tools/src/write.rs index 130d5c3a..3276208d 100644 --- a/crates/tools/src/write.rs +++ b/crates/tools/src/write.rs @@ -9,7 +9,7 @@ use serde::Deserialize; use crate::error::ToolsError; use crate::tracker::Tracker; -use workdir::{StatRequest, WorkdirError, WorkdirHandle, WorkdirPath, WriteRequest}; +use workdir::{StatRequest, WorkdirError, WorkdirPath, WorkdirSessionHandle, WriteRequest}; const DESCRIPTION: &str = "Create a new file or overwrite an existing one with \ the given content. Missing parent directories within scope are created \ @@ -25,7 +25,7 @@ pub(crate) struct WriteParams { } pub(crate) struct WriteTool { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, tracker: Tracker, } @@ -44,14 +44,14 @@ impl Tool for WriteTool { let mutation_key = PathBuf::from(path.as_str()); let _mutation_permit = self.tracker.acquire_mutation(&mutation_key, &ctx).await; - let expected_hash = match self.workdir.stat(StatRequest { path: path.clone() }).await { + let expected_hash = match self.session.stat(StatRequest { path: path.clone() }).await { Ok(_) => Some(self.tracker.expected_workdir_hash(&path)?), Err(WorkdirError::NotFound(_)) => None, Err(error) => return Err(ToolsError::from(error).into()), }; let outcome = self - .workdir + .session .write(WriteRequest { path: path.clone(), content: params.content.as_bytes().to_vec(), @@ -81,7 +81,7 @@ impl Tool for WriteTool { } /// Factory for the `Write` tool. -pub fn write_tool(workdir: WorkdirHandle, tracker: Tracker) -> ToolDefinition { +pub fn write_tool(session: WorkdirSessionHandle, tracker: Tracker) -> ToolDefinition { Arc::new(move || { let schema = schemars::schema_for!(WriteParams); let schema_value = serde_json::to_value(schema).unwrap_or(serde_json::json!({})); @@ -89,7 +89,7 @@ pub fn write_tool(workdir: WorkdirHandle, tracker: Tracker) -> ToolDefinition { .description(DESCRIPTION) .input_schema(schema_value); let tool: Arc = Arc::new(WriteTool { - workdir: workdir.clone(), + session: session.clone(), tracker: tracker.clone(), }); (meta, tool) @@ -102,15 +102,15 @@ mod tests { use crate::read::read_tool; use manifest::Scope; use tempfile::TempDir; - use workdir::LocalWorkdir; + use workdir::LocalWorkdirSession; - fn setup() -> (TempDir, WorkdirHandle, Tracker) { + fn setup() -> (TempDir, WorkdirSessionHandle, Tracker) { let dir = TempDir::new().unwrap(); - let workdir: WorkdirHandle = Arc::new(LocalWorkdir::new( + let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::new( Scope::writable(dir.path()).unwrap(), dir.path().to_path_buf(), )); - (dir, workdir, Tracker::new()) + (dir, session, Tracker::new()) } #[tokio::test] diff --git a/crates/tools/tests/edge_cases.rs b/crates/tools/tests/edge_cases.rs index 13997aa7..e87c7d0c 100644 --- a/crates/tools/tests/edge_cases.rs +++ b/crates/tools/tests/edge_cases.rs @@ -7,7 +7,7 @@ use manifest::{Permission, Scope, ScopeConfig, ScopeRule}; use serde_json::json; use tempfile::TempDir; use tools::{Tracker, core_builtin_tools}; -use workdir::{LocalWorkdir, WorkdirHandle}; +use workdir::{LocalWorkdirSession, WorkdirSessionHandle}; struct Registry { entries: Vec<(llm_engine::tool::ToolMeta, Arc)>, @@ -42,7 +42,8 @@ fn setup() -> (TempDir, TempDir, Registry) { recursive: true, }); let scope = Scope::from_config(&config).unwrap(); - let fs: WorkdirHandle = std::sync::Arc::new(LocalWorkdir::new(scope, dir.path().to_path_buf())); + let fs: WorkdirSessionHandle = + std::sync::Arc::new(LocalWorkdirSession::new(scope, dir.path().to_path_buf())); let tracker = Tracker::new(); let reg = Registry::new(core_builtin_tools(fs, tracker, spill.path().to_path_buf())); (dir, spill, reg) diff --git a/crates/tools/tests/integration.rs b/crates/tools/tests/integration.rs index 5d11e7c9..44730fe8 100644 --- a/crates/tools/tests/integration.rs +++ b/crates/tools/tests/integration.rs @@ -12,7 +12,7 @@ use manifest::{Permission, Scope, ScopeConfig, ScopeRule}; use serde_json::json; use tempfile::TempDir; use tools::{Tracker, core_builtin_tools}; -use workdir::{LocalWorkdir, WorkdirHandle}; +use workdir::{LocalWorkdirSession, WorkdirSessionHandle}; fn scope_with_spill(workspace: &Path, spill: &Path) -> Scope { let base = Scope::writable(workspace).unwrap(); @@ -55,7 +55,8 @@ fn setup() -> (TempDir, TempDir, Registry) { let dir = TempDir::new().unwrap(); let spill = TempDir::new().unwrap(); let scope = scope_with_spill(dir.path(), spill.path()); - let fs: WorkdirHandle = Arc::new(LocalWorkdir::new(scope, dir.path().to_path_buf())); + let fs: WorkdirSessionHandle = + Arc::new(LocalWorkdirSession::new(scope, dir.path().to_path_buf())); let tracker = Tracker::new(); let reg = Registry::new(core_builtin_tools(fs, tracker, spill.path().to_path_buf())); (dir, spill, reg) @@ -198,7 +199,7 @@ async fn absolute_path_is_rejected() { }), ) .await; - // Absolute paths are rejected at the logical Workdir boundary. + // Absolute paths are rejected at the logical WorkdirSession boundary. let msg = format!("{err}"); assert!(msg.contains("invalid Workdir path"), "unexpected: {msg}"); } @@ -224,7 +225,7 @@ async fn write_to_existing_without_read_fails() { #[tokio::test] async fn shared_workdir_across_tools() { - // The key invariant: all builtin tools share the same Workdir instance, + // The key invariant: all builtin tools share the same WorkdirSession instance, // so read-history set by Read is visible to Edit and Write. let (dir, _spill, reg) = setup(); let file = dir.path().join("shared.txt"); @@ -239,7 +240,7 @@ async fn shared_workdir_across_tools() { json!({ "file_path": file.file_name().unwrap().to_str().unwrap() }), ) .await; - // Write via Write tool — must succeed because the shared Workdir has the read + // Write via Write tool — must succeed because the shared WorkdirSession has the read call( &write, json!({ @@ -301,7 +302,8 @@ async fn tracker_recent_files_tracks_read_write_edit() { let dir = TempDir::new().unwrap(); let spill = TempDir::new().unwrap(); let scope = scope_with_spill(dir.path(), spill.path()); - let fs: WorkdirHandle = Arc::new(LocalWorkdir::new(scope, dir.path().to_path_buf())); + let fs: WorkdirSessionHandle = + Arc::new(LocalWorkdirSession::new(scope, dir.path().to_path_buf())); let tracker = Tracker::new(); let reg = Registry::new(core_builtin_tools( fs, @@ -346,7 +348,7 @@ async fn tracker_recent_files_tracks_read_write_edit() { #[tokio::test] async fn bash_inherits_workdir_cwd() { - // The Bash tool starts at the Workdir's pwd. Without any `cd`, its + // The Bash tool starts at the WorkdirSession's pwd. Without any `cd`, its // `pwd` should canonicalize to the workspace root we set up. let (dir, _spill, reg) = setup(); let bash = reg.get("Bash"); @@ -363,7 +365,7 @@ async fn bash_provider_output_does_not_expose_internal_paths() { let bash = reg.get("Bash"); let out = call(&bash, json!({ "command": "printf 'x%.0s' {1..20480}" })).await; let body = out.content.unwrap(); - assert!(body.contains("bounded Workdir command output")); + assert!(body.contains("bounded WorkdirSession command output")); assert!(!body.contains(spill.path().to_str().unwrap())); assert_eq!(std::fs::read_dir(spill.path()).unwrap().count(), 0); } diff --git a/crates/workdir/src/lib.rs b/crates/workdir/src/lib.rs index 6272d63e..3deff787 100644 --- a/crates/workdir/src/lib.rs +++ b/crates/workdir/src/lib.rs @@ -1,8 +1,9 @@ -//! Workdir authority and local materialization provider. +//! Persistent Workdir identity and Worker-bound operation sessions. //! -//! A Workdir is the host-owned execution context bound to one Worker. Tools -//! consume this interface; they do not own Workdir identity, paths, scope, or -//! lifecycle. +//! A [`Workdir`] identifies a materialized repository execution context across +//! Worker lifetimes. A [`WorkdirSession`] is the live operation attachment +//! bound to one Worker. Tools consume sessions; they do not own Workdir +//! materialization or cleanup. mod local; mod operation; @@ -14,12 +15,47 @@ use std::sync::Arc; use async_trait::async_trait; use serde::{Deserialize, Serialize}; -pub use local::{LocalWorkdir, SymlinkInfo, direct_symlink, first_symlink}; +pub use local::{LocalWorkdirSession, SymlinkInfo, direct_symlink, first_symlink}; pub use operation::*; +/// Persistent, opaque identity of one materialized Workdir. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +pub struct Workdir { + id: WorkdirId, +} + +impl Workdir { + pub fn new(id: impl Into) -> Self { + Self { + id: WorkdirId(id.into()), + } + } + + pub fn id(&self) -> &WorkdirId { + &self.id + } +} + +/// Opaque Workdir identifier assigned by the materialization authority. +#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] +#[serde(transparent)] +pub struct WorkdirId(String); + +impl WorkdirId { + pub fn as_str(&self) -> &str { + &self.0 + } +} + +impl std::fmt::Display for WorkdirId { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(&self.0) + } +} + #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] -pub enum WorkdirCapability { +pub enum WorkdirSessionCapability { Read, Write, Edit, @@ -29,11 +65,11 @@ pub enum WorkdirCapability { } #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -pub struct WorkdirCapabilities { +pub struct WorkdirSessionCapabilities { bits: u8, } -impl WorkdirCapabilities { +impl WorkdirSessionCapabilities { const READ: u8 = 1 << 0; const WRITE: u8 = 1 << 1; const EDIT: u8 = 1 << 2; @@ -43,20 +79,22 @@ impl WorkdirCapabilities { pub const EMPTY: Self = Self { bits: 0 }; - pub fn from_capabilities(capabilities: impl IntoIterator) -> Self { + pub fn from_capabilities( + capabilities: impl IntoIterator, + ) -> Self { capabilities .into_iter() .fold(Self::EMPTY, |set, capability| set.with(capability)) } - pub const fn with(mut self, capability: WorkdirCapability) -> Self { + pub const fn with(mut self, capability: WorkdirSessionCapability) -> Self { self.bits |= match capability { - WorkdirCapability::Read => Self::READ, - WorkdirCapability::Write => Self::WRITE, - WorkdirCapability::Edit => Self::EDIT, - WorkdirCapability::Glob => Self::GLOB, - WorkdirCapability::Grep => Self::GREP, - WorkdirCapability::Command => Self::COMMAND, + WorkdirSessionCapability::Read => Self::READ, + WorkdirSessionCapability::Write => Self::WRITE, + WorkdirSessionCapability::Edit => Self::EDIT, + WorkdirSessionCapability::Glob => Self::GLOB, + WorkdirSessionCapability::Grep => Self::GREP, + WorkdirSessionCapability::Command => Self::COMMAND, }; self } @@ -69,14 +107,14 @@ impl WorkdirCapabilities { bits: Self::READ | Self::GLOB | Self::GREP, }; - pub const fn supports(self, capability: WorkdirCapability) -> bool { + pub const fn supports(self, capability: WorkdirSessionCapability) -> bool { let bit = match capability { - WorkdirCapability::Read => Self::READ, - WorkdirCapability::Write => Self::WRITE, - WorkdirCapability::Edit => Self::EDIT, - WorkdirCapability::Glob => Self::GLOB, - WorkdirCapability::Grep => Self::GREP, - WorkdirCapability::Command => Self::COMMAND, + WorkdirSessionCapability::Read => Self::READ, + WorkdirSessionCapability::Write => Self::WRITE, + WorkdirSessionCapability::Edit => Self::EDIT, + WorkdirSessionCapability::Glob => Self::GLOB, + WorkdirSessionCapability::Grep => Self::GREP, + WorkdirSessionCapability::Command => Self::COMMAND, }; self.bits & bit != 0 } @@ -84,15 +122,16 @@ impl WorkdirCapabilities { pub type WriteOutcome = WriteResult; -/// Network-capable operations available on one bound Workdir. +/// Live, Worker-bound operations for one persistent [`Workdir`]. /// /// Implementations execute filesystem search and command work on the host -/// that owns the materialization. Requests and results never contain the raw -/// materialized root. +/// that owns the materialization. Structured requests and results never +/// contain the raw materialized root. Closing a session is terminal and does +/// not delete the persistent Workdir or its materialization. #[async_trait] -pub trait Workdir: std::fmt::Debug + Send + Sync { - fn binding_id(&self) -> Option<&str>; - fn capabilities(&self) -> WorkdirCapabilities; +pub trait WorkdirSession: std::fmt::Debug + Send + Sync { + fn workdir(&self) -> &Workdir; + fn capabilities(&self) -> WorkdirSessionCapabilities; async fn stat(&self, request: StatRequest) -> Result; async fn read(&self, request: ReadRequest) -> Result; @@ -108,26 +147,27 @@ pub trait Workdir: std::fmt::Debug + Send + Sync { request: CommandOutputRequest, ) -> Result; async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError>; - async fn shutdown(&self) -> Result<(), WorkdirError>; + /// Terminal, idempotent release of this Worker-bound operation session. + async fn close(&self) -> Result<(), WorkdirError>; } -pub type WorkdirHandle = Arc; +pub type WorkdirSessionHandle = Arc; #[derive(Debug, thiserror::Error)] pub enum WorkdirError { - #[error("Workdir does not support {0:?}")] - Unsupported(WorkdirCapability), + #[error("Workdir session does not support {0:?}")] + Unsupported(WorkdirSessionCapability), #[error("invalid Workdir path: {0}")] InvalidPath(String), - #[error("Workdir provider is unavailable: {0}")] + #[error("Workdir session is unavailable: {0}")] Unavailable(String), #[error("Workdir content was modified externally before the operation could be applied: {0}")] Conflict(String), - #[error("unknown Workdir command: {0}")] + #[error("unknown Workdir session command: {0}")] UnknownCommand(String), #[error("path must be absolute: {}", .0.display())] diff --git a/crates/workdir/src/local.rs b/crates/workdir/src/local.rs index c3beb319..bb3a228e 100644 --- a/crates/workdir/src/local.rs +++ b/crates/workdir/src/local.rs @@ -1,11 +1,11 @@ //! Scope-aware filesystem primitive. //! -//! `LocalWorkdir` is the write/read gate layered on top of a [`manifest::Scope`] +//! `LocalWorkdirSession` is the write/read gate layered on top of a [`manifest::Scope`] //! and a Worker's working directory. The scope decides which paths are //! readable and writable; the cwd is carried alongside for convenience //! (Glob/Grep default their search base to it). //! -//! `LocalWorkdir` is cheap to clone (`Arc` inside). Tool-specific session +//! `LocalWorkdirSession` is cheap to clone (`Arc` inside). Tool-specific session //! state, such as read-before-edit tracking, remains owned by the tool layer. use std::collections::HashMap; @@ -13,7 +13,7 @@ use std::io::{Read as _, Seek as _, SeekFrom, Write as _}; use std::path::{Path, PathBuf}; use std::process::Stdio; use std::sync::Arc; -use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::time::Duration; use async_trait::async_trait; @@ -29,8 +29,8 @@ use crate::{ CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest, EditResult, EntryKind, GlobRequest, GlobResult, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult, ReadRequest, ReadResult, StatRequest, StatResult, Workdir, - WorkdirCapabilities, WorkdirCapability, WorkdirError, WorkdirPath, WriteOutcome, WriteRequest, - WriteResult, + WorkdirError, WorkdirPath, WorkdirSession, WorkdirSessionCapabilities, + WorkdirSessionCapability, WriteOutcome, WriteRequest, WriteResult, }; #[derive(Debug)] @@ -43,17 +43,19 @@ enum LocalCommand { } #[derive(Debug)] -struct LocalWorkdirInner { - binding_id: Option, +struct LocalWorkdirSessionInner { + workdir: Workdir, root: PathBuf, scope: SharedScope, cwd: PathBuf, - capabilities: WorkdirCapabilities, + capabilities: WorkdirSessionCapabilities, + closed: AtomicBool, + close_lock: Mutex<()>, next_command_id: AtomicU64, commands: Mutex>, } -impl Drop for LocalWorkdirInner { +impl Drop for LocalWorkdirSessionInner { fn drop(&mut self) { if let Ok(mut commands) = self.commands.try_lock() { for (_, command) in commands.drain() { @@ -68,13 +70,13 @@ impl Drop for LocalWorkdirInner { /// Scope-aware filesystem handle. Clone-cheap (`Arc` inside). /// /// The wrapped [`SharedScope`] is shared with every clone of this -/// `LocalWorkdir` and with whoever else holds the same `SharedScope` +/// `LocalWorkdirSession` and with whoever else holds the same `SharedScope` /// handle (typically the owning Worker). Mutations to that `SharedScope` /// propagate atomically; the next permission check inside any -/// `LocalWorkdir` reads the new view. +/// `LocalWorkdirSession` reads the new view. #[derive(Debug, Clone)] -pub struct LocalWorkdir { - inner: Arc, +pub struct LocalWorkdirSession { + inner: Arc, } /// First symlink encountered while resolving a path. @@ -94,48 +96,63 @@ pub struct SymlinkInfo { pub target_exists: bool, } -impl LocalWorkdir { - /// Create a new [`LocalWorkdir`] wrapping `scope` and `cwd` in a fresh - /// [`SharedScope`]. Use [`LocalWorkdir::with_shared_scope`] when you - /// need the resulting `LocalWorkdir` to share scope state with another +fn local_workdir_identity(root: &Path) -> Workdir { + let canonical = root.canonicalize().unwrap_or_else(|_| root.to_path_buf()); + let digest = Sha256::digest(canonical.to_string_lossy().as_bytes()); + let digest = digest + .iter() + .map(|byte| format!("{byte:02x}")) + .collect::(); + Workdir::new(format!("local-{digest}")) +} + +impl LocalWorkdirSession { + /// Create a new [`LocalWorkdirSession`] wrapping `scope` and `cwd` in a fresh + /// [`SharedScope`]. Use [`LocalWorkdirSession::with_shared_scope`] when you + /// need the resulting `LocalWorkdirSession` to share scope state with another /// holder of the `SharedScope` (typically the Worker). pub fn new(scope: Scope, cwd: PathBuf) -> Self { Self::materialized( cwd.clone(), cwd, SharedScope::new(scope), - WorkdirCapabilities::ALL, + WorkdirSessionCapabilities::ALL, ) } pub fn with_shared_scope(scope: SharedScope, cwd: PathBuf) -> Self { - Self::materialized(cwd.clone(), cwd, scope, WorkdirCapabilities::ALL) + Self::materialized(cwd.clone(), cwd, scope, WorkdirSessionCapabilities::ALL) } - /// Construct the local provider for an existing Worker–Workdir binding. + /// Construct a standalone local session with a deterministic identity + /// derived from the canonical materialization root. pub fn materialized( root: PathBuf, cwd: PathBuf, scope: SharedScope, - capabilities: WorkdirCapabilities, + capabilities: WorkdirSessionCapabilities, ) -> Self { - Self::materialized_bound(None, root, cwd, scope, capabilities) + let workdir = local_workdir_identity(&root); + Self::materialized_bound(workdir, root, cwd, scope, capabilities) } + /// Open a local session for an authority-assigned persistent Workdir. pub fn materialized_bound( - binding_id: Option, + workdir: Workdir, root: PathBuf, cwd: PathBuf, scope: SharedScope, - capabilities: WorkdirCapabilities, + capabilities: WorkdirSessionCapabilities, ) -> Self { Self { - inner: Arc::new(LocalWorkdirInner { - binding_id, + inner: Arc::new(LocalWorkdirSessionInner { + workdir, root, scope, cwd, capabilities, + closed: AtomicBool::new(false), + close_lock: Mutex::new(()), next_command_id: AtomicU64::new(1), commands: Mutex::new(HashMap::new()), }), @@ -153,7 +170,7 @@ impl LocalWorkdir { self.inner.scope.snapshot() } - /// Shared scope handle backing this `LocalWorkdir`. Cloning it lets a + /// Shared scope handle backing this `LocalWorkdirSession`. Cloning it lets a /// caller (usually the Worker) hold the same view and push updates /// that are immediately reflected in subsequent permission checks. pub fn shared_scope(&self) -> &SharedScope { @@ -298,7 +315,19 @@ impl LocalWorkdir { }) } - fn ensure_capability(&self, capability: WorkdirCapability) -> Result<(), WorkdirError> { + fn ensure_open(&self) -> Result<(), WorkdirError> { + if self.inner.closed.load(Ordering::Acquire) { + Err(WorkdirError::Unavailable(format!( + "Workdir session for {} is closed", + self.inner.workdir.id() + ))) + } else { + Ok(()) + } + } + + fn ensure_capability(&self, capability: WorkdirSessionCapability) -> Result<(), WorkdirError> { + self.ensure_open()?; if self.inner.capabilities.supports(capability) { Ok(()) } else { @@ -323,17 +352,17 @@ impl LocalWorkdir { } #[async_trait] -impl Workdir for LocalWorkdir { - fn binding_id(&self) -> Option<&str> { - self.inner.binding_id.as_deref() +impl WorkdirSession for LocalWorkdirSession { + fn workdir(&self) -> &Workdir { + &self.inner.workdir } - fn capabilities(&self) -> WorkdirCapabilities { + fn capabilities(&self) -> WorkdirSessionCapabilities { self.inner.capabilities } async fn stat(&self, request: StatRequest) -> Result { - self.ensure_capability(WorkdirCapability::Read)?; + self.ensure_capability(WorkdirSessionCapability::Read)?; let path = self.resolve(&request.path); let metadata = std::fs::symlink_metadata(&path).map_err(|error| { let error = match error.kind() { @@ -359,9 +388,9 @@ impl Workdir for LocalWorkdir { } async fn read(&self, request: ReadRequest) -> Result { - self.ensure_capability(WorkdirCapability::Read)?; + self.ensure_capability(WorkdirSessionCapability::Read)?; let path = self.resolve(&request.path); - let bytes = LocalWorkdir::read_bytes(self, &path) + let bytes = LocalWorkdirSession::read_bytes(self, &path) .map_err(|error| sanitize_error(error, &request.path))?; let content_hash = Sha256::digest(&bytes).into(); let lines = bytes @@ -403,10 +432,10 @@ impl Workdir for LocalWorkdir { } async fn write(&self, request: WriteRequest) -> Result { - self.ensure_capability(WorkdirCapability::Write)?; + self.ensure_capability(WorkdirSessionCapability::Write)?; let path = self.resolve(&request.path); if path.exists() { - let current = LocalWorkdir::read_bytes(self, &path) + let current = LocalWorkdirSession::read_bytes(self, &path) .map_err(|error| sanitize_error(error, &request.path))?; let current_hash: [u8; 32] = Sha256::digest(¤t).into(); if request.expected_hash != Some(current_hash) { @@ -415,14 +444,14 @@ impl Workdir for LocalWorkdir { } else if request.expected_hash.is_some() { return Err(WorkdirError::Conflict(request.path.to_string())); } - LocalWorkdir::write(self, &path, &request.content) + LocalWorkdirSession::write(self, &path, &request.content) .map_err(|error| sanitize_error(error, &request.path)) } async fn edit(&self, request: EditRequest) -> Result { - self.ensure_capability(WorkdirCapability::Edit)?; + self.ensure_capability(WorkdirSessionCapability::Edit)?; let path = self.resolve(&request.path); - let bytes = LocalWorkdir::read_bytes(self, &path) + let bytes = LocalWorkdirSession::read_bytes(self, &path) .map_err(|error| sanitize_error(error, &request.path))?; let current_hash: [u8; 32] = Sha256::digest(&bytes).into(); if current_hash != request.expected_hash { @@ -446,7 +475,7 @@ impl Workdir for LocalWorkdir { } else { text.replacen(&request.old_string, &request.new_string, 1) }; - let outcome = LocalWorkdir::write(self, &path, edited.as_bytes()) + let outcome = LocalWorkdirSession::write(self, &path, edited.as_bytes()) .map_err(|error| sanitize_error(error, &request.path))?; let content_hash = Sha256::digest(edited.as_bytes()).into(); Ok(EditResult { @@ -457,7 +486,7 @@ impl Workdir for LocalWorkdir { } async fn list(&self, request: ListRequest) -> Result { - self.ensure_capability(WorkdirCapability::Read)?; + self.ensure_capability(WorkdirSessionCapability::Read)?; let base = self.resolve(&request.path); let scope = self.inner.scope.snapshot(); if !scope.is_readable(&base) { @@ -520,7 +549,7 @@ impl Workdir for LocalWorkdir { } async fn glob(&self, request: GlobRequest) -> Result { - self.ensure_capability(WorkdirCapability::Glob)?; + self.ensure_capability(WorkdirSessionCapability::Glob)?; let base = self.resolve(&request.path); if let Some(info) = direct_symlink(&base) && info.target_exists @@ -562,7 +591,7 @@ impl Workdir for LocalWorkdir { } async fn grep(&self, request: GrepRequest) -> Result { - self.ensure_capability(WorkdirCapability::Grep)?; + self.ensure_capability(WorkdirSessionCapability::Grep)?; let base = self.resolve(&request.path); let logical = request.path.clone(); crate::search::run_grep( @@ -575,7 +604,7 @@ impl Workdir for LocalWorkdir { } async fn start_command(&self, request: CommandRequest) -> Result { - self.ensure_capability(WorkdirCapability::Command)?; + self.ensure_capability(WorkdirSessionCapability::Command)?; let id = self.inner.next_command_id.fetch_add(1, Ordering::Relaxed); let handle = CommandHandle(format!("command-{id}")); let cwd = self.inner.cwd.clone(); @@ -586,16 +615,18 @@ impl Workdir for LocalWorkdir { task_completion.notify_one(); output }); - self.inner - .commands - .lock() - .await - .insert(handle.0.clone(), LocalCommand::Running { task, completion }); + let mut commands = self.inner.commands.lock().await; + if let Err(error) = self.ensure_open() { + task.abort(); + completion.notify_one(); + return Err(error); + } + commands.insert(handle.0.clone(), LocalCommand::Running { task, completion }); Ok(handle) } async fn command_status(&self, handle: CommandHandle) -> Result { - self.ensure_capability(WorkdirCapability::Command)?; + self.ensure_capability(WorkdirSessionCapability::Command)?; let commands = self.inner.commands.lock().await; let command = commands .get(&handle.0) @@ -611,8 +642,9 @@ impl Workdir for LocalWorkdir { &self, request: CommandOutputRequest, ) -> Result { - self.ensure_capability(WorkdirCapability::Command)?; + self.ensure_capability(WorkdirSessionCapability::Command)?; let command = loop { + self.ensure_open()?; let mut commands = self.inner.commands.lock().await; let Some(command) = commands.get(&request.handle.0) else { return Err(WorkdirError::UnknownCommand(request.handle.0.clone())); @@ -651,17 +683,16 @@ impl Workdir for LocalWorkdir { }; let page = command_output_page(&output, request.cursor, request.limit); if page.next_cursor.is_some() { - self.inner - .commands - .lock() - .await - .insert(request.handle.0, LocalCommand::Completed(output)); + let mut commands = self.inner.commands.lock().await; + if !self.inner.closed.load(Ordering::Acquire) { + commands.insert(request.handle.0, LocalCommand::Completed(output)); + } } Ok(page) } async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> { - self.ensure_capability(WorkdirCapability::Command)?; + self.ensure_capability(WorkdirSessionCapability::Command)?; let command = self .inner .commands @@ -676,7 +707,11 @@ impl Workdir for LocalWorkdir { Ok(()) } - async fn shutdown(&self) -> Result<(), WorkdirError> { + async fn close(&self) -> Result<(), WorkdirError> { + let _close_guard = self.inner.close_lock.lock().await; + if self.inner.closed.swap(true, Ordering::AcqRel) { + return Ok(()); + } let mut commands = self.inner.commands.lock().await; for (_, command) in commands.drain() { if let LocalCommand::Running { task, completion } = command { @@ -955,8 +990,8 @@ mod tests { use std::fs; use tempfile::TempDir; - fn make_fs(dir: &TempDir) -> LocalWorkdir { - LocalWorkdir::new( + fn make_fs(dir: &TempDir) -> LocalWorkdirSession { + LocalWorkdirSession::new( Scope::writable(dir.path()).unwrap(), dir.path().to_path_buf(), ) @@ -968,7 +1003,7 @@ mod tests { let workdir = make_fs(&dir); let path = WorkdirPath::new("notes/item.txt").unwrap(); - let written = Workdir::write( + let written = WorkdirSession::write( &workdir, WriteRequest { path: path.clone(), @@ -980,7 +1015,7 @@ mod tests { .unwrap(); assert!(written.created); - let read = Workdir::read( + let read = WorkdirSession::read( &workdir, ReadRequest { path: path.clone(), @@ -994,7 +1029,7 @@ mod tests { assert_eq!(read.bytes, b"alpha\nbeta\n"); assert!(!read.truncated); - let bounded = Workdir::read( + let bounded = WorkdirSession::read( &workdir, ReadRequest { path: path.clone(), @@ -1009,7 +1044,7 @@ mod tests { assert!(bounded.truncated); assert_eq!(bounded.content_hash, read.content_hash); - let edited = Workdir::edit( + let edited = WorkdirSession::edit( &workdir, EditRequest { path: path.clone(), @@ -1023,13 +1058,13 @@ mod tests { .unwrap(); assert_eq!(edited.replacements, 1); - let stat = Workdir::stat(&workdir, StatRequest { path: path.clone() }) + let stat = WorkdirSession::stat(&workdir, StatRequest { path: path.clone() }) .await .unwrap(); assert_eq!(stat.path, path); assert_eq!(stat.kind, EntryKind::File); - let listed = Workdir::list( + let listed = WorkdirSession::list( &workdir, ListRequest { path: WorkdirPath::new("notes").unwrap(), @@ -1041,7 +1076,7 @@ mod tests { assert_eq!(listed.total_entries, 1); assert_eq!(listed.entries[0].path.as_str(), "notes/item.txt"); - let error = Workdir::edit( + let error = WorkdirSession::edit( &workdir, EditRequest { path: path.clone(), @@ -1056,7 +1091,7 @@ mod tests { assert!(matches!(error, WorkdirError::Conflict(_))); std::fs::remove_file(dir.path().join("notes/item.txt")).unwrap(); - let error = Workdir::write( + let error = WorkdirSession::write( &workdir, WriteRequest { path, @@ -1069,19 +1104,108 @@ mod tests { assert!(matches!(error, WorkdirError::Conflict(_))); } + #[tokio::test] + async fn close_is_terminal_for_one_session_without_deleting_workdir_identity() { + let dir = TempDir::new().unwrap(); + std::fs::write(dir.path().join("item.txt"), "persisted").unwrap(); + let workdir = Workdir::new("working-directory-42"); + let scope = SharedScope::new(Scope::writable(dir.path()).unwrap()); + let session = LocalWorkdirSession::materialized_bound( + workdir.clone(), + dir.path().to_path_buf(), + dir.path().to_path_buf(), + scope.clone(), + WorkdirSessionCapabilities::ALL, + ); + assert_eq!(session.workdir(), &workdir); + + let command = WorkdirSession::start_command( + &session, + CommandRequest { + command: "sleep 30".to_owned(), + timeout_secs: 60, + output_limit: 1024, + }, + ) + .await + .unwrap(); + let waiting_session = session.clone(); + let waiting_command = command.clone(); + let waiter = tokio::spawn(async move { + WorkdirSession::command_output( + &waiting_session, + CommandOutputRequest { + handle: waiting_command, + cursor: 0, + limit: 1024, + wait: true, + }, + ) + .await + }); + tokio::task::yield_now().await; + WorkdirSession::close(&session).await.unwrap(); + WorkdirSession::close(&session).await.unwrap(); + let waiter_error = tokio::time::timeout(Duration::from_secs(1), waiter) + .await + .expect("close should wake command output waiters") + .unwrap() + .unwrap_err(); + assert!(matches!(waiter_error, WorkdirError::Unavailable(_))); + + assert!(matches!( + WorkdirSession::command_status(&session, command).await, + Err(WorkdirError::Unavailable(_)) + )); + assert!(matches!( + WorkdirSession::read( + &session, + ReadRequest { + path: WorkdirPath::new("item.txt").unwrap(), + offset: 0, + limit: 10, + max_bytes: 1024, + }, + ) + .await, + Err(WorkdirError::Unavailable(_)) + )); + + let restored = LocalWorkdirSession::materialized_bound( + workdir.clone(), + dir.path().to_path_buf(), + dir.path().to_path_buf(), + scope, + WorkdirSessionCapabilities::ALL, + ); + assert_eq!(restored.workdir(), &workdir); + let read = WorkdirSession::read( + &restored, + ReadRequest { + path: WorkdirPath::new("item.txt").unwrap(), + offset: 0, + limit: 10, + max_bytes: 1024, + }, + ) + .await + .unwrap(); + assert_eq!(read.bytes, b"persisted"); + } + #[tokio::test] async fn capability_boundary_rejects_direct_unsupported_operation() { let dir = TempDir::new().unwrap(); - let workdir = LocalWorkdir::materialized( + let workdir = LocalWorkdirSession::materialized( dir.path().to_path_buf(), dir.path().to_path_buf(), SharedScope::new(Scope::writable(dir.path()).unwrap()), - WorkdirCapabilities::READ_ONLY, + WorkdirSessionCapabilities::READ_ONLY, ); assert_eq!(workdir.root(), dir.path()); assert_eq!(workdir.cwd(), dir.path()); - let error = Workdir::write( + let error = WorkdirSession::write( &workdir, WriteRequest { path: WorkdirPath::new("blocked.txt").unwrap(), @@ -1093,7 +1217,7 @@ mod tests { .unwrap_err(); assert!(matches!( error, - WorkdirError::Unsupported(WorkdirCapability::Write) + WorkdirError::Unsupported(WorkdirSessionCapability::Write) )); assert!(!dir.path().join("blocked.txt").exists()); } @@ -1331,7 +1455,7 @@ mod tests { }], }; let scope = Scope::from_config(&cfg).unwrap(); - let scoped = LocalWorkdir::new(scope, dir.path().to_path_buf()); + let scoped = LocalWorkdirSession::new(scope, dir.path().to_path_buf()); let err = scoped.write(&sub.join("locked.txt"), b"x").unwrap_err(); assert!( matches!(err, WorkdirError::ReadOnly(_)), @@ -1365,7 +1489,7 @@ mod tests { } // ------------------------------------------------------------------------- - // Dynamic scope: SharedScope mutations propagate into LocalWorkdir decisions + // Dynamic scope: SharedScope mutations propagate into LocalWorkdirSession decisions // ------------------------------------------------------------------------- #[test] @@ -1378,7 +1502,7 @@ mod tests { fs::write(&extra_file, b"hi").unwrap(); let shared = SharedScope::new(Scope::writable(dir.path()).unwrap()); - let fs = LocalWorkdir::with_shared_scope(shared.clone(), dir.path().to_path_buf()); + let fs = LocalWorkdirSession::with_shared_scope(shared.clone(), dir.path().to_path_buf()); // Before: extra is out of scope. let err = fs.read_bytes(&extra_file).unwrap_err(); @@ -1415,7 +1539,7 @@ mod tests { let target = sub.join("a.txt"); let shared = SharedScope::new(Scope::writable(dir.path()).unwrap()); - let fs = LocalWorkdir::with_shared_scope(shared.clone(), dir.path().to_path_buf()); + let fs = LocalWorkdirSession::with_shared_scope(shared.clone(), dir.path().to_path_buf()); // Write succeeds initially. fs.write(&target, b"first").unwrap(); @@ -1449,7 +1573,7 @@ mod tests { let target = dir.path().join("a.txt"); let shared = SharedScope::new(Scope::writable(dir.path()).unwrap()); - let fs1 = LocalWorkdir::with_shared_scope(shared.clone(), dir.path().to_path_buf()); + let fs1 = LocalWorkdirSession::with_shared_scope(shared.clone(), dir.path().to_path_buf()); let fs2 = fs1.clone(); // fs1 writes; both clones see the file. @@ -1488,7 +1612,7 @@ mod tests { ) .unwrap(); let workdir = make_fs(&dir); - let glob = Workdir::glob( + let glob = WorkdirSession::glob( &workdir, GlobRequest { pattern: "**/*.rs".into(), @@ -1499,7 +1623,7 @@ mod tests { .await .unwrap(); assert_eq!(glob.paths, [WorkdirPath::new("src/main.rs").unwrap()]); - let grep = Workdir::grep( + let grep = WorkdirSession::grep( &workdir, GrepRequest { pattern: "NEEDLE".into(), @@ -1520,7 +1644,7 @@ mod tests { assert_eq!(grep.match_count, 1); assert!(grep.output.contains("src/main.rs")); assert!(!grep.output.contains(dir.path().to_string_lossy().as_ref())); - let handle = Workdir::start_command( + let handle = WorkdirSession::start_command( &workdir, CommandRequest { command: "pwd && printf provider-command".into(), @@ -1530,7 +1654,7 @@ mod tests { ) .await .unwrap(); - let output = Workdir::command_output( + let output = WorkdirSession::command_output( &workdir, CommandOutputRequest { handle, @@ -1554,7 +1678,7 @@ mod tests { async fn completed_command_output_can_be_read_in_bounded_unicode_pages() { let dir = TempDir::new().unwrap(); let workdir = make_fs(&dir); - let handle = Workdir::start_command( + let handle = WorkdirSession::start_command( &workdir, CommandRequest { command: "printf 'aéz'".into(), @@ -1564,7 +1688,7 @@ mod tests { ) .await .unwrap(); - let first = Workdir::command_output( + let first = WorkdirSession::command_output( &workdir, CommandOutputRequest { handle: handle.clone(), @@ -1578,7 +1702,7 @@ mod tests { assert_eq!(first.content, "aé"); assert_eq!(first.next_cursor, Some(2)); - let second = Workdir::command_output( + let second = WorkdirSession::command_output( &workdir, CommandOutputRequest { handle: handle.clone(), @@ -1592,7 +1716,7 @@ mod tests { assert_eq!(second.content, "z"); assert_eq!(second.next_cursor, None); assert!(matches!( - Workdir::command_status(&workdir, handle).await, + WorkdirSession::command_status(&workdir, handle).await, Err(WorkdirError::UnknownCommand(_)) )); } @@ -1601,7 +1725,7 @@ mod tests { async fn provider_cancels_active_command() { let dir = TempDir::new().unwrap(); let workdir = make_fs(&dir); - let handle = Workdir::start_command( + let handle = WorkdirSession::start_command( &workdir, CommandRequest { command: "sleep 30".into(), @@ -1612,7 +1736,7 @@ mod tests { .await .unwrap(); assert_eq!( - Workdir::command_status(&workdir, handle.clone()) + WorkdirSession::command_status(&workdir, handle.clone()) .await .unwrap(), CommandStatus::Running @@ -1620,7 +1744,7 @@ mod tests { let waiting_workdir = workdir.clone(); let waiting_handle = handle.clone(); let waiter = tokio::spawn(async move { - Workdir::command_output( + WorkdirSession::command_output( &waiting_workdir, CommandOutputRequest { handle: waiting_handle, @@ -1632,7 +1756,7 @@ mod tests { .await }); tokio::task::yield_now().await; - Workdir::cancel_command(&workdir, handle.clone()) + WorkdirSession::cancel_command(&workdir, handle.clone()) .await .unwrap(); let waiter_error = tokio::time::timeout(Duration::from_secs(1), waiter) @@ -1642,7 +1766,7 @@ mod tests { .unwrap_err(); assert!(matches!(waiter_error, WorkdirError::UnknownCommand(_))); assert!(matches!( - Workdir::command_status(&workdir, handle).await, + WorkdirSession::command_status(&workdir, handle).await, Err(WorkdirError::UnknownCommand(_)) )); } diff --git a/crates/worker-runtime/src/worker_backend.rs b/crates/worker-runtime/src/worker_backend.rs index 17242635..fcf6b49d 100644 --- a/crates/worker-runtime/src/worker_backend.rs +++ b/crates/worker-runtime/src/worker_backend.rs @@ -37,7 +37,7 @@ use session_store::{CombinedStore, FsWorkerStore}; use tokio::runtime::Runtime; #[cfg(feature = "ws-server")] use tokio::sync::broadcast; -use workdir::{LocalWorkdir, WorkdirCapabilities, WorkdirHandle}; +use workdir::{LocalWorkdirSession, Workdir, WorkdirSessionCapabilities, WorkdirSessionHandle}; #[cfg(feature = "ws-server")] use worker::ipc::protocol_session::{live_log_entry_event, subscribe_worker_protocol_session}; @@ -357,18 +357,18 @@ async fn fetch_profile_source_archive_http( ) } -fn runtime_local_workdir( - binding_id: &str, +fn runtime_local_workdir_session( + workdir_id: &str, root: &Path, cwd: &Path, scope: manifest::SharedScope, -) -> WorkdirHandle { - Arc::new(LocalWorkdir::materialized_bound( - Some(binding_id.to_owned()), +) -> WorkdirSessionHandle { + Arc::new(LocalWorkdirSession::materialized_bound( + Workdir::new(workdir_id), root.to_path_buf(), cwd.to_path_buf(), scope, - WorkdirCapabilities::ALL, + WorkdirSessionCapabilities::ALL, )) } @@ -448,14 +448,14 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { .await .map_err(|err| format!("failed to create Worker from profile: {err}"))?; if let Some(binding) = request.working_directory.as_ref() { - worker.bind_workdir(Some(runtime_local_workdir( + worker.bind_workdir_session(Some(runtime_local_workdir_session( &binding.working_directory.id, binding.root(), binding.cwd(), worker.scope().clone(), ))); } else { - worker.bind_workdir(None); + worker.bind_workdir_session(None); } let runtime_base = self.runtime_base_dir()?; @@ -543,14 +543,14 @@ impl RuntimeWorkerFactory for ProfileRuntimeWorkerFactory { Err(err) => return Err(format!("failed to restore Worker from metadata: {err}")), }; if let Some(binding) = request.working_directory.as_ref() { - worker.bind_workdir(Some(runtime_local_workdir( + worker.bind_workdir_session(Some(runtime_local_workdir_session( &binding.working_directory.id, binding.root(), binding.cwd(), worker.scope().clone(), ))); } else { - worker.bind_workdir(None); + worker.bind_workdir_session(None); } let runtime_base = self.runtime_base_dir()?; @@ -1613,23 +1613,23 @@ mod tests { } #[test] - fn runtime_rebind_preserves_working_directory_id_on_a_fresh_provider() { + fn restore_opens_a_fresh_session_for_the_same_workdir_identity() { let root = tempfile::tempdir().unwrap(); - let spawned = runtime_local_workdir( + let spawned = runtime_local_workdir_session( "working-directory-42", root.path(), root.path(), manifest::SharedScope::new(Scope::writable(root.path()).unwrap()), ); - let restored = runtime_local_workdir( + let restored = runtime_local_workdir_session( "working-directory-42", root.path(), root.path(), manifest::SharedScope::new(Scope::writable(root.path()).unwrap()), ); - assert_eq!(spawned.binding_id(), Some("working-directory-42")); - assert_eq!(restored.binding_id(), Some("working-directory-42")); + assert_eq!(spawned.workdir().id().as_str(), "working-directory-42"); + assert_eq!(restored.workdir().id().as_str(), "working-directory-42"); assert!(!Arc::ptr_eq(&spawned, &restored)); } diff --git a/crates/worker/src/compact/worker.rs b/crates/worker/src/compact/worker.rs index 76c965ad..4eb6149a 100644 --- a/crates/worker/src/compact/worker.rs +++ b/crates/worker/src/compact/worker.rs @@ -27,8 +27,8 @@ use llm_engine::interceptor::{Interceptor, PreRequestAction, PreToolAction, Tool use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput, ToolResult}; use serde::Deserialize; #[cfg(test)] -use workdir::LocalWorkdir; -use workdir::{ReadRequest, WorkdirHandle, WorkdirPath}; +use workdir::LocalWorkdirSession; +use workdir::{ReadRequest, WorkdirPath, WorkdirSessionHandle}; use crate::compact::usage_tracker::UsageTracker; use crate::fs_view::ReadRequirement; @@ -327,7 +327,7 @@ fn truncate_to_token_budget(text: &mut String, max_tokens: u64) -> bool { } struct MarkReadRequiredTool { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, ctx: Arc>, } @@ -342,12 +342,12 @@ impl Tool for MarkReadRequiredTool { ToolError::InvalidArgument(format!("invalid mark_read_required input: {e}")) })?; - // Read through the shared Workdir so scope and I/O errors surface the + // Read through the shared WorkdirSession so scope and I/O errors surface the // same way the regular `read_file` tool does. let path = WorkdirPath::new(params.file_path.to_string_lossy()) .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; let result = self - .workdir + .session .read(ReadRequest { path, offset: params.offset.unwrap_or(0), @@ -454,7 +454,7 @@ impl Tool for WriteSummaryTool { } pub(crate) fn mark_read_required_tool( - workdir: WorkdirHandle, + session: WorkdirSessionHandle, ctx: Arc>, ) -> ToolDefinition { Arc::new(move || { @@ -464,7 +464,7 @@ pub(crate) fn mark_read_required_tool( .description(MARK_DESCRIPTION) .input_schema(schema_value); let tool: Arc = Arc::new(MarkReadRequiredTool { - workdir: workdir.clone(), + session: session.clone(), ctx: ctx.clone(), }); (meta, tool) @@ -635,9 +635,9 @@ mod tests { use super::*; use manifest::Scope; - fn make_fs(tmp: &std::path::Path) -> WorkdirHandle { + fn make_fs(tmp: &std::path::Path) -> WorkdirSessionHandle { let scope = Scope::writable(tmp.to_path_buf()).unwrap(); - Arc::new(LocalWorkdir::new(scope, tmp.to_path_buf())) + Arc::new(LocalWorkdirSession::new(scope, tmp.to_path_buf())) } fn make_usage(input: u64) -> llm_engine::timeline::event::UsageEvent { @@ -732,7 +732,7 @@ mod tests { let ctx = Arc::new(Mutex::new(CompactWorkerContext::with_budget(1_000))); let tool: Arc = Arc::new(MarkReadRequiredTool { - workdir: make_fs(tmp.path()), + session: make_fs(tmp.path()), ctx: ctx.clone(), }); let input = serde_json::json!({ "file_path": path.file_name().unwrap().to_str().unwrap() }) @@ -754,7 +754,7 @@ mod tests { let ctx = Arc::new(Mutex::new(CompactWorkerContext::with_budget(100))); let tool: Arc = Arc::new(MarkReadRequiredTool { - workdir: make_fs(tmp.path()), + session: make_fs(tmp.path()), ctx: ctx.clone(), }); let input = serde_json::json!({ "file_path": path.file_name().unwrap().to_str().unwrap() }) diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index b1530b02..426ceafe 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -222,6 +222,26 @@ impl WorkerController { } async fn spawn_inner( + worker: Worker, + runtime_base: &Path, + runtime_managed: bool, + ) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error> + where + C: LlmClient + Clone + 'static, + St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static, + { + let session = worker.workdir_session().cloned(); + let result = Self::spawn_initialized(worker, runtime_base, runtime_managed).await; + if result.is_err() + && let Some(session) = session + && let Err(error) = session.close().await + { + tracing::warn!(%error, "Workdir session close after controller startup failure failed"); + } + result + } + + async fn spawn_initialized( mut worker: Worker, runtime_base: &Path, runtime_managed: bool, @@ -562,7 +582,7 @@ fn wire_event_bridges_on_engine( /// Register the builtin file-manipulation tools, optional memory tools, /// and the Worker-orchestration tools (SpawnWorker + comm) on the Worker's -/// Engine. Returns the Workdir handle used to attach a `WorkerFsView` to +/// Engine. Returns the WorkdirSession handle used to attach a `WorkerFsView` to /// the shared state. async fn register_worker_tools( worker: &mut Worker, @@ -570,7 +590,7 @@ async fn register_worker_tools( spawner_socket: PathBuf, runtime_base: PathBuf, spawned_registry: Arc, -) -> std::io::Result> +) -> std::io::Result> where C: LlmClient + Clone + 'static, St: Store + WorkerMetadataStore + Clone + 'static, @@ -578,7 +598,7 @@ where // Worker-immutable snapshots taken before the mutable worker borrow // below so the worker borrow doesn't conflict with reads on `worker`. let scope_handle = worker.scope().clone(); - let worker_workdir = worker.workdir().cloned(); + let worker_workdir = worker.workdir_session().cloned(); 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(); @@ -1174,17 +1194,17 @@ async fn controller_loop( } } - if let Some(workdir) = worker.workdir() - && let Err(error) = workdir.shutdown().await - { - tracing::warn!(%error, "Workdir provider shutdown failed"); - } - // Background memory jobs own extract/consolidate workers after a - // turn completes. Join them before the controller task exits so - // staging writes and consolidation cleanups are not abandoned. + // turn completes. Join them before closing the Workdir session so no + // Worker-owned task can outlive its operation attachment. worker.wait_for_memory_jobs().await; + if let Some(session) = worker.workdir_session() + && let Err(error) = session.close().await + { + tracing::warn!(%error, "Workdir session close failed"); + } + // Report upward that this Worker is stopping before the controller // task exits. Awaited (not fire-and-forget): after `shutdown_tx.send` // the process may exit quickly, and a spawned task would be killed diff --git a/crates/worker/src/fs_view.rs b/crates/worker/src/fs_view.rs index acca643d..c1105b06 100644 --- a/crates/worker/src/fs_view.rs +++ b/crates/worker/src/fs_view.rs @@ -1,6 +1,6 @@ //! Worker 視点のファイルシステム操作。 //! -//! `Workdir` の上に「Worker が読み取りたい / 列挙したい」操作を集約する軽い wrapper。 +//! `WorkdirSession` の上に「Worker が読み取りたい / 列挙したい」操作を集約する軽い wrapper。 //! //! - `ReadRequirement` と `render_auto_read` — compact worker が `mark_read_required` //! で nominate したファイルを再読し、`[Auto-read file: ...]` system message に @@ -16,8 +16,10 @@ use llm_engine::Item; use tools::ToolsError; use tracing::warn; #[cfg(test)] -use workdir::LocalWorkdir; -use workdir::{EntryKind, ListRequest, ReadRequest, StatRequest, WorkdirHandle, WorkdirPath}; +use workdir::LocalWorkdirSession; +use workdir::{ + EntryKind, ListRequest, ReadRequest, StatRequest, WorkdirPath, WorkdirSessionHandle, +}; /// 補完候補1件の最大数。`list_file_completions` がこの値を超えたら打ち切り。 const COMPLETION_LIMIT: usize = 100; @@ -38,10 +40,10 @@ pub struct ReadRequirement { pub limit: Option, } -/// Worker から見えるファイルシステム操作の入口。Clone は cheap（`Workdir` 内 `Arc`）。 +/// Worker から見えるファイルシステム操作の入口。Clone は cheap（`WorkdirSession` 内 `Arc`）。 #[derive(Debug, Clone)] pub struct WorkerFsView { - workdir: WorkdirHandle, + session: WorkdirSessionHandle, } /// `list_file_completions` が返す候補1件。 @@ -54,10 +56,10 @@ pub struct FileCandidate { } /// `resolve_file_ref` の失敗理由。Worker 側で Alert に振り分けるために -/// Workdir / 内部判定の両方を区別できるよう保持する。 +/// WorkdirSession / 内部判定の両方を区別できるよう保持する。 #[derive(Debug)] pub enum ResolveError { - /// Path resolution / scope check failed via `Workdir`. + /// Path resolution / scope check failed via `WorkdirSession`. Fs(ToolsError), /// File contents are not valid UTF-8 (binary / non-text). Binary { path: PathBuf }, @@ -77,11 +79,11 @@ impl std::fmt::Display for ResolveError { impl std::error::Error for ResolveError {} impl WorkerFsView { - pub fn new(workdir: WorkdirHandle) -> Self { - Self { workdir } + pub fn new(session: WorkdirSessionHandle) -> Self { + Self { session } } - pub fn workdir(&self) -> &WorkdirHandle { - &self.workdir + pub fn session(&self) -> &WorkdirSessionHandle { + &self.session } pub async fn render_auto_read(&self, requirements: &[ReadRequirement]) -> Vec { @@ -95,7 +97,7 @@ impl WorkerFsView { } }; match self - .workdir + .session .read(ReadRequest { path: path.clone(), offset: req.offset.unwrap_or(0), @@ -128,7 +130,7 @@ impl WorkerFsView { .map_err(ToolsError::from) .map_err(ResolveError::Fs)?; let stat = self - .workdir + .session .stat(StatRequest { path: logical.clone(), }) @@ -137,7 +139,7 @@ impl WorkerFsView { .map_err(ResolveError::Fs)?; if stat.kind == EntryKind::Directory { let result = self - .workdir + .session .list(ListRequest { path: logical.clone(), limit: DIR_FILE_REF_ENTRY_LIMIT, @@ -175,7 +177,7 @@ impl WorkerFsView { return Ok(Item::system_message(text)); } let result = self - .workdir + .session .read(ReadRequest { path: logical.clone(), offset: 0, @@ -216,7 +218,7 @@ impl WorkerFsView { return Vec::new(); }; let Ok(result) = self - .workdir + .session .list(ListRequest { path: parent, limit: COMPLETION_LIMIT, @@ -285,8 +287,8 @@ mod tests { use std::sync::Arc; use tempfile::TempDir; - fn fs_for(dir: &TempDir) -> WorkdirHandle { - Arc::new(LocalWorkdir::new( + fn fs_for(dir: &TempDir) -> WorkdirSessionHandle { + Arc::new(LocalWorkdirSession::new( Scope::writable(dir.path()).unwrap(), dir.path().to_path_buf(), )) @@ -417,7 +419,8 @@ mod tests { }], }; let scope = Scope::from_config(&cfg).unwrap(); - let fs: WorkdirHandle = Arc::new(LocalWorkdir::new(scope, dir.path().to_path_buf())); + let fs: WorkdirSessionHandle = + Arc::new(LocalWorkdirSession::new(scope, dir.path().to_path_buf())); let view = WorkerFsView::new(fs); let item = view.resolve_file_ref("docs", 4096).await.unwrap(); @@ -493,7 +496,7 @@ mod tests { std::fs::create_dir(&inner).unwrap(); std::fs::write(outer.path().join("secret.txt"), "nope").unwrap(); let scope = Scope::writable(&inner).unwrap(); - let fs: WorkdirHandle = Arc::new(LocalWorkdir::new(scope, inner.clone())); + let fs: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::new(scope, inner.clone())); let view = WorkerFsView::new(fs); // Absolute path outside of scope. @@ -579,7 +582,8 @@ mod tests { }], }; let scope = Scope::from_config(&cfg).unwrap(); - let fs: WorkdirHandle = Arc::new(LocalWorkdir::new(scope, dir.path().to_path_buf())); + let fs: WorkdirSessionHandle = + Arc::new(LocalWorkdirSession::new(scope, dir.path().to_path_buf())); let view = WorkerFsView::new(fs); let cands = view.list_file_completions("").await; diff --git a/crates/worker/src/shared_state.rs b/crates/worker/src/shared_state.rs index 793acb0a..a10c4778 100644 --- a/crates/worker/src/shared_state.rs +++ b/crates/worker/src/shared_state.rs @@ -23,7 +23,7 @@ pub struct WorkerSharedState { pub greeting: protocol::Greeting, pub status: RwLock, /// Worker-from-the-inside view of the filesystem. Set once in - /// `WorkerController::start` after the local Workdir provider is + /// `WorkerController::start` after the local WorkdirSession provider is /// materialised, and read from the IPC server layer to answer /// `ListCompletions` queries without going through the controller. It is /// unset only in unit tests that construct `WorkerSharedState` directly. diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index de7fa371..a0498d3c 100644 --- a/crates/worker/src/worker.rs +++ b/crates/worker/src/worker.rs @@ -76,7 +76,7 @@ use protocol::{ use tokio::net::UnixStream; use tokio::sync::broadcast; use tokio::task::JoinHandle; -use workdir::{LocalWorkdir, WorkdirCapabilities, WorkdirHandle}; +use workdir::{LocalWorkdirSession, WorkdirSessionCapabilities, WorkdirSessionHandle}; const RESTORE_RECONCILIATION_REACHABILITY_TIMEOUT: Duration = Duration::from_millis(500); @@ -643,14 +643,14 @@ pub struct Worker { /// Explicit local filesystem authority, or `None` for Workers with no /// local cwd and no filesystem/Bash tool surface. filesystem_authority: WorkerFilesystemAuthority, - /// Live Workdir provider derived once from the Worker–Workdir binding. + /// Live WorkdirSession provider derived once from the Worker–Workdir binding. /// Local tools, file views, and compaction workers clone this handle. - workdir: Option, + workdir_session: Option, /// Path-free workspace identity/client context injected by Runtime/host. /// This never grants local filesystem authority. workspace_context: WorkerWorkspaceContext, /// Shared, atomically-swappable view of the Worker's resolved scope. - /// Cloned into local Workdir providers used by builtin tools, fs_view, + /// Cloned into local WorkdirSession providers used by builtin tools, fs_view, /// and compaction so updates propagate at the next permission check. scope: SharedScope, /// Filesystem authority this Worker may pass to spawned children. Direct tools @@ -827,7 +827,7 @@ impl Worker worker_metadata_writer: None, segment_state: self.segment_state.clone(), filesystem_authority: self.filesystem_authority.clone(), - workdir: self.workdir.clone(), + workdir_session: self.workdir_session.clone(), workspace_context: self.workspace_context.clone(), scope: self.scope.clone(), delegation_scope: self.delegation_scope.clone(), @@ -1016,7 +1016,7 @@ impl Worker { let delegation_scope = DelegationScope::from_config(&manifest.delegation_scope).map_err(WorkerError::Scope)?; let scope = SharedScope::new(scope); - let workdir = workdir_from_authority(&filesystem_authority, &scope); + let workdir_session = workdir_session_from_authority(&filesystem_authority, &scope); let mut worker = Self { manifest, engine: Some(worker), @@ -1024,7 +1024,7 @@ impl Worker { worker_metadata_writer: None, segment_state: SegmentState::new(session_id, segment_id, 0), filesystem_authority, - workdir, + workdir_session, workspace_context, scope, delegation_scope, @@ -1139,15 +1139,15 @@ impl Worker { self.filesystem_authority.as_local() } - pub fn workdir(&self) -> Option<&WorkdirHandle> { - self.workdir.as_ref() + pub fn workdir_session(&self) -> Option<&WorkdirSessionHandle> { + self.workdir_session.as_ref() } /// Replace the constructor fallback with the provider binding resolved by /// the owning Runtime. Runtime calls this before the Worker controller is /// spawned, so tools only ever observe the Runtime-bound handle. - pub fn bind_workdir(&mut self, workdir: Option) { - self.workdir = workdir; + pub fn bind_workdir_session(&mut self, workdir_session: Option) { + self.workdir_session = workdir_session; } /// Path-free workspace identity, if Runtime/host associated this Worker @@ -2014,7 +2014,7 @@ impl Worker { /// unresolved placeholder stays in the flattened user message so the LLM /// still sees the intent. async fn resolve_file_refs(&self, segments: &[Segment]) -> Vec { - let Some(workdir) = self.workdir.clone() else { + let Some(workdir) = self.workdir_session.clone() else { for seg in segments { if let Segment::FileRef { path } = seg { self.alert( @@ -2792,9 +2792,9 @@ impl Worker { ))); // Build an independent compact worker. It clones the main Worker's - // provider handle, so compact-time reads use the same Workdir instance. + // provider handle, so compact-time reads use the same WorkdirSession instance. // No-workdir Workers deliberately omit compact-time filesystem tools. - let workdir = self.workdir.clone(); + let workdir = self.workdir_session.clone(); let summary_tracker = tools::Tracker::new(); let summary_client: Box = self.build_compactor_client()?; let summary_system_prompt = self @@ -3858,7 +3858,7 @@ where worker.set_cache_key(Some(segment_id.to_string())); let worker_metadata_writer = Some(worker_metadata_writer_for_store(&store)); let scope = SharedScope::new(common.scope); - let workdir = workdir_from_authority(&common.filesystem_authority, &scope); + let workdir_session = workdir_session_from_authority(&common.filesystem_authority, &scope); let mut worker = Self { manifest, @@ -3867,7 +3867,7 @@ where worker_metadata_writer, segment_state: SegmentState::new(session_id, segment_id, 0), filesystem_authority: common.filesystem_authority, - workdir, + workdir_session, workspace_context: common.workspace_context, scope, delegation_scope: common.delegation_scope, @@ -3967,7 +3967,7 @@ where worker.set_cache_key(Some(segment_id.to_string())); let worker_metadata_writer = Some(worker_metadata_writer_for_store(&store)); let scope = SharedScope::new(common.scope); - let workdir = workdir_from_authority(&common.filesystem_authority, &scope); + let workdir_session = workdir_session_from_authority(&common.filesystem_authority, &scope); let mut worker = Self { manifest, @@ -3976,7 +3976,7 @@ where worker_metadata_writer, segment_state: SegmentState::new(session_id, segment_id, 0), filesystem_authority: common.filesystem_authority, - workdir, + workdir_session, workspace_context: common.workspace_context, scope, delegation_scope: common.delegation_scope, @@ -4259,7 +4259,7 @@ where let task_feature = TaskFeature::from_history(&state.history); let worker_metadata_writer = Some(worker_metadata_writer_for_store(&store)); let scope = SharedScope::new(common.scope); - let workdir = workdir_from_authority(&common.filesystem_authority, &scope); + let workdir_session = workdir_session_from_authority(&common.filesystem_authority, &scope); let mut worker = Self { manifest, @@ -4268,7 +4268,7 @@ where worker_metadata_writer, segment_state: SegmentState::new(session_id, segment_id, state.entries_count), filesystem_authority: common.filesystem_authority, - workdir, + workdir_session, workspace_context: common.workspace_context, scope, delegation_scope: common.delegation_scope, @@ -4934,17 +4934,17 @@ pub enum WorkerError { }, } -fn workdir_from_authority( +fn workdir_session_from_authority( authority: &WorkerFilesystemAuthority, scope: &SharedScope, -) -> Option { +) -> Option { authority.as_local().map(|local| { - Arc::new(LocalWorkdir::materialized( + Arc::new(LocalWorkdirSession::materialized( local.root.clone(), local.cwd.clone(), scope.clone(), - WorkdirCapabilities::ALL, - )) as WorkdirHandle + WorkdirSessionCapabilities::ALL, + )) as WorkdirSessionHandle }) } diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 5008fb6a..7d09ad0b 100644 --- a/crates/worker/tests/controller_test.rs +++ b/crates/worker/tests/controller_test.rs @@ -11,7 +11,10 @@ use llm_engine::llm_client::{ClientError, LlmClient, Request}; use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use session_store::{CombinedStore, FsWorkerStore}; use session_store::{FsStore, LogEntry}; -use workdir::{CommandRequest, LocalWorkdir, WorkdirCapabilities, WorkdirError, WorkdirHandle}; +use workdir::{ + CommandRequest, LocalWorkdirSession, Workdir, WorkdirError, WorkdirSessionCapabilities, + WorkdirSessionHandle, +}; use worker::{ Event, Method, Worker, WorkerController, WorkerFilesystemAuthority, WorkerHandle, @@ -215,16 +218,16 @@ async fn spawn_controller(worker: Worker) -> WorkerHandle } #[tokio::test] -async fn shutdown_closes_bound_workdir_commands() { +async fn shutdown_closes_bound_workdir_session() { let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await; - let workdir: WorkdirHandle = Arc::new(LocalWorkdir::materialized_bound( - Some("controller-test-workdir".to_owned()), + let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound( + Workdir::new("controller-test-workdir"), pwd.clone(), pwd, worker.scope().clone(), - WorkdirCapabilities::ALL, + WorkdirSessionCapabilities::ALL, )); - let command = workdir + let command = session .start_command(CommandRequest { command: "sleep 30".to_owned(), timeout_secs: 60, @@ -232,7 +235,7 @@ async fn shutdown_closes_bound_workdir_commands() { }) .await .unwrap(); - worker.bind_workdir(Some(Arc::clone(&workdir))); + worker.bind_workdir_session(Some(Arc::clone(&session))); let runtime_base = tempfile::tempdir().unwrap(); let (handle, shutdown_rx) = WorkerController::spawn(worker, runtime_base.path()) @@ -245,8 +248,40 @@ async fn shutdown_closes_bound_workdir_commands() { .expect("controller shutdown signal should remain open"); assert!(matches!( - workdir.command_status(command).await, - Err(WorkdirError::UnknownCommand(_)) + session.command_status(command).await, + Err(WorkdirError::Unavailable(_)) + )); +} + +#[tokio::test] +async fn controller_startup_failure_closes_bound_workdir_session() { + let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await; + let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound( + Workdir::new("controller-startup-failure-workdir"), + pwd.clone(), + pwd, + worker.scope().clone(), + WorkdirSessionCapabilities::ALL, + )); + worker.bind_workdir_session(Some(Arc::clone(&session))); + let runtime_base = tempfile::tempdir().unwrap(); + let invalid_runtime_base = runtime_base.path().join("not-a-directory"); + std::fs::write(&invalid_runtime_base, "file").unwrap(); + + assert!( + WorkerController::spawn(worker, &invalid_runtime_base) + .await + .is_err() + ); + assert!(matches!( + session + .start_command(CommandRequest { + command: "printf unreachable".to_owned(), + timeout_secs: 5, + output_limit: 1024, + }) + .await, + Err(WorkdirError::Unavailable(_)) )); }