workdir: separate identity from worker session

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