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

workdir: add network-capable operation boundary

Browse Source
This commit is contained in:
Hare
2026-08-03 16:14:01 +09:00
parent 0fa36395e7
commit ddadc830ac
32 changed files with 3590 additions and 3241 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/Cargo.toml
+1
View File
@@ -26,6 +26,7 @@ tokio = { workspace = true, features = ["fs", "io-util", "macros", "net", "proce
toml = { workspace = true }
tracing = { workspace = true }
tools = { workspace = true }
workdir = { workspace = true }
minijinja = "2.19.0"
chrono = "0.4"
include_dir = "0.7.4"
crates/worker/src/compact/worker.rs
+32 -18
View File
@@ -26,10 +26,14 @@ use llm_engine::Item;
use llm_engine::interceptor::{Interceptor, PreRequestAction, PreToolAction, ToolCallInfo};
use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput, ToolResult};
use serde::Deserialize;
use tools::ScopedFs;
#[cfg(test)]
use workdir::LocalWorkdir;
use workdir::{ReadRequest, WorkdirHandle, WorkdirPath};
use crate::compact::usage_tracker::UsageTracker;
use crate::fs_view::{ReadRequirement, slice_lines};
use crate::fs_view::ReadRequirement;
#[cfg(test)]
use crate::fs_view::slice_lines;
use crate::session_reference::{
ReadDetail, ReadOptions, ReadSelector, SearchOptions, SessionReferenceView, ToolPart,
};
@@ -323,7 +327,7 @@ fn truncate_to_token_budget(text: &mut String, max_tokens: u64) -> bool {
}
struct MarkReadRequiredTool {
fs: ScopedFs,
workdir: WorkdirHandle,
ctx: Arc<Mutex<CompactWorkerContext>>,
}
@@ -338,14 +342,22 @@ impl Tool for MarkReadRequiredTool {
ToolError::InvalidArgument(format!("invalid mark_read_required input: {e}"))
})?;
// Read the file through the shared ScopedFs so scope and I/O
// errors surface the same way the regular `read_file` tool does.
let bytes = self
.fs
.read_bytes(&params.file_path)
// Read through the shared Workdir so scope and I/O errors surface the
// same way the regular `read_file` tool does.
let path = WorkdirPath::new(params.file_path.to_string_lossy())
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
let result = self
.workdir
.read(ReadRequest {
path,
offset: params.offset.unwrap_or(0),
limit: params.limit.unwrap_or(usize::MAX),
max_bytes: 4 * 1024 * 1024,
})
.await
.map_err(|e| ToolError::ExecutionFailed(format!("read failed: {e}")))?;
let text = String::from_utf8_lossy(&bytes);
let slice = slice_lines(&text, params.offset.unwrap_or(0), params.limit);
let text = String::from_utf8_lossy(&result.bytes);
let slice = text.as_ref();
let estimated_tokens = estimate_tokens(slice.len());
let mut guard = self.ctx.lock().expect("compact worker context poisoned");
@@ -442,7 +454,7 @@ impl Tool for WriteSummaryTool {
}
pub(crate) fn mark_read_required_tool(
fs: ScopedFs,
workdir: WorkdirHandle,
ctx: Arc<Mutex<CompactWorkerContext>>,
) -> ToolDefinition {
Arc::new(move || {
@@ -452,7 +464,7 @@ pub(crate) fn mark_read_required_tool(
.description(MARK_DESCRIPTION)
.input_schema(schema_value);
let tool: Arc<dyn Tool> = Arc::new(MarkReadRequiredTool {
fs: fs.clone(),
workdir: workdir.clone(),
ctx: ctx.clone(),
});
(meta, tool)
@@ -623,9 +635,9 @@ mod tests {
use super::*;
use manifest::Scope;
fn make_fs(tmp: &std::path::Path) -> ScopedFs {
fn make_fs(tmp: &std::path::Path) -> WorkdirHandle {
let scope = Scope::writable(tmp.to_path_buf()).unwrap();
ScopedFs::new(scope, tmp.to_path_buf())
Arc::new(LocalWorkdir::new(scope, tmp.to_path_buf()))
}
fn make_usage(input: u64) -> llm_engine::timeline::event::UsageEvent {
@@ -720,10 +732,11 @@ mod tests {
let ctx = Arc::new(Mutex::new(CompactWorkerContext::with_budget(1_000)));
let tool: Arc<dyn Tool> = Arc::new(MarkReadRequiredTool {
fs: make_fs(tmp.path()),
workdir: make_fs(tmp.path()),
ctx: ctx.clone(),
});
let input = serde_json::json!({ "file_path": path.to_str().unwrap() }).to_string();
let input = serde_json::json!({ "file_path": path.file_name().unwrap().to_str().unwrap() })
.to_string();
let out = tool.execute(&input, Default::default()).await.unwrap();
assert!(out.summary.starts_with("Marked"));
@@ -741,10 +754,11 @@ mod tests {
let ctx = Arc::new(Mutex::new(CompactWorkerContext::with_budget(100)));
let tool: Arc<dyn Tool> = Arc::new(MarkReadRequiredTool {
fs: make_fs(tmp.path()),
workdir: make_fs(tmp.path()),
ctx: ctx.clone(),
});
let input = serde_json::json!({ "file_path": path.to_str().unwrap() }).to_string();
let input = serde_json::json!({ "file_path": path.file_name().unwrap().to_str().unwrap() })
.to_string();
let res = tool.execute(&input, Default::default()).await;
assert!(matches!(res, Err(ToolError::ExecutionFailed(_))));
crates/worker/src/controller.rs
+30 -23
View File
@@ -88,23 +88,25 @@ impl WorkerHandle {
(event, entry_rx)
}
pub fn completion_entries(
pub async fn completion_entries(
&self,
kind: protocol::CompletionKind,
prefix: &str,
) -> Vec<protocol::CompletionEntry> {
match kind {
protocol::CompletionKind::File => self
.shared_state
.fs_view()
.map(|view| view.list_file_completions(prefix))
.unwrap_or_default()
.into_iter()
.map(|c| protocol::CompletionEntry {
value: c.path,
is_dir: c.is_dir,
})
.collect(),
protocol::CompletionKind::File => {
let Some(view) = self.shared_state.fs_view() else {
return Vec::new();
};
view.list_file_completions(prefix)
.await
.into_iter()
.map(|candidate| protocol::CompletionEntry {
value: candidate.path,
is_dir: candidate.is_dir,
})
.collect()
}
}
}
@@ -574,7 +576,7 @@ fn wire_event_bridges_on_engine<C, St>(
/// Register the builtin file-manipulation tools, optional memory tools,
/// and the Worker-orchestration tools (SpawnWorker + comm) on the Worker's
/// Engine. Returns the `ScopedFs` clone used to attach a `WorkerFsView` to
/// Engine. Returns the Workdir handle used to attach a `WorkerFsView` to
/// the shared state.
async fn register_worker_tools<C, St>(
worker: &mut Worker<C, St>,
@@ -582,7 +584,7 @@ async fn register_worker_tools<C, St>(
spawner_socket: PathBuf,
runtime_base: PathBuf,
spawned_registry: Arc<SpawnedWorkerRegistry>,
) -> std::io::Result<Option<tools::ScopedFs>>
) -> std::io::Result<Option<workdir::WorkdirHandle>>
where
C: LlmClient + Clone + 'static,
St: Store + WorkerMetadataStore + Clone + 'static,
@@ -590,6 +592,7 @@ where
// Worker-immutable snapshots taken before the mutable worker borrow
// below so the worker borrow doesn't conflict with reads on `worker`.
let scope_handle = worker.scope().clone();
let worker_workdir = worker.workdir().cloned();
let local_filesystem = worker.local_working_directory().cloned();
let local_workspace_root = local_filesystem.as_ref().map(|local| local.root.clone());
let task_feature = worker.task_feature();
@@ -603,21 +606,19 @@ where
let worker_metadata_store = worker.store().clone();
let self_parent_socket = worker.callback_socket().cloned();
// The Worker's SharedScope is the single source of truth for every
// ScopedFs when local filesystem authority exists. No-workdir Workers
// deliberately skip constructing/registering filesystem and Bash tools.
let (fs_for_view, tracker) = if let Some(local) = local_filesystem.as_ref() {
let fs = tools::ScopedFs::with_shared_scope(scope_handle.clone(), local.cwd.clone());
// Resolve the existing Worker–Workdir binding into the domain provider.
// Tools only consume the provider handle; they do not own its root, cwd,
// scope, or lifecycle. No-workdir Workers expose no local tools.
let (workdir_for_view, tracker) = if let Some(workdir) = worker_workdir {
let tracker = tools::Tracker::new();
let fs_for_view = fs.clone();
worker
.engine_mut()
.register_tools(tools::core_builtin_tools(
fs,
workdir.clone(),
tracker.clone(),
bash_output_dir,
));
(Some(fs_for_view), Some(tracker))
(Some(workdir), Some(tracker))
} else {
(None, None)
};
@@ -788,7 +789,7 @@ where
if let Some(tracker) = tracker {
worker.attach_tracker(tracker);
}
Ok(fs_for_view)
Ok(workdir_for_view)
}
/// Idle/Paused event loop. Each iteration either fires a staged
@@ -1187,6 +1188,12 @@ 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
// turn completes. Join them before the controller task exits so
// staging writes and consolidation cleanups are not abandoned.
crates/worker/src/fs_view.rs
+219 -304
View File
@@ -1,6 +1,6 @@
//! Worker 視点のファイルシステム操作。
//!
//! `ScopedFs` の上に「Worker が読み取りたい / 列挙したい」操作を集約する軽い wrapper。
//! `Workdir` の上に「Worker が読み取りたい / 列挙したい」操作を集約する軽い wrapper。
//!
//! - `ReadRequirement` と `render_auto_read` — compact worker が `mark_read_required`
//! で nominate したファイルを再読し、`[Auto-read file: ...]` system message に
@@ -13,16 +13,19 @@
use std::path::{Path, PathBuf};
use llm_engine::Item;
use manifest::Scope;
use tools::scoped_fs::first_symlink;
use tools::{ScopedFs, ToolsError};
use tools::ToolsError;
use tracing::warn;
#[cfg(test)]
use workdir::LocalWorkdir;
use workdir::{EntryKind, ListRequest, ReadRequest, StatRequest, WorkdirHandle, WorkdirPath};
/// 補完候補1件の最大数。`list_file_completions` がこの値を超えたら打ち切り。
const COMPLETION_LIMIT: usize = 100;
/// submit-time directory FileRef の shallow listing で返す最大 entry 数。
/// TUI completion と同じ浅い一覧という意味論に揃えるため、同じ上限を使う。
const DIR_FILE_REF_ENTRY_LIMIT: usize = COMPLETION_LIMIT;
/// Provider-side bound for auto-read and submit-time referenced-file reads.
const AUTO_READ_BYTE_LIMIT: usize = 4 * 1024 * 1024;
/// Compact worker が `mark_read_required` で nominate した「次セッション開始時に
/// 自動で再読すべきファイル」のエントリ。
@@ -35,10 +38,10 @@ pub struct ReadRequirement {
pub limit: Option<usize>,
}
/// Worker から見えるファイルシステム操作の入口。Clone は cheap（`ScopedFs` 内 `Arc`）。
/// Worker から見えるファイルシステム操作の入口。Clone は cheap（`Workdir` 内 `Arc`）。
#[derive(Debug, Clone)]
pub struct WorkerFsView {
fs: ScopedFs,
workdir: WorkdirHandle,
}
/// `list_file_completions` が返す候補1件。
@@ -51,10 +54,10 @@ pub struct FileCandidate {
}
/// `resolve_file_ref` の失敗理由。Worker 側で Alert に振り分けるために
/// ScopedFs / 内部判定の両方を区別できるよう保持する。
/// Workdir / 内部判定の両方を区別できるよう保持する。
#[derive(Debug)]
pub enum ResolveError {
/// Path resolution / scope check failed via `ScopedFs`.
/// Path resolution / scope check failed via `Workdir`.
Fs(ToolsError),
/// File contents are not valid UTF-8 (binary / non-text).
Binary { path: PathBuf },
@@ -74,142 +77,172 @@ impl std::fmt::Display for ResolveError {
impl std::error::Error for ResolveError {}
impl WorkerFsView {
pub fn new(fs: ScopedFs) -> Self {
Self { fs }
pub fn new(workdir: WorkdirHandle) -> Self {
Self { workdir }
}
pub fn workdir(&self) -> &WorkdirHandle {
&self.workdir
}
pub fn fs(&self) -> &ScopedFs {
&self.fs
}
/// `requirements` の各エントリを `ScopedFs` 経由で再読し、
/// `[Auto-read file: <path>:<range>]\n<body>` 形式の system message に変換する。
/// 読み取り失敗（NotFound / OutOfScope 等）は warn で記録してスキップする
/// — compact 全体を落とさないため。
pub fn render_auto_read(&self, requirements: &[ReadRequirement]) -> Vec<Item> {
pub async fn render_auto_read(&self, requirements: &[ReadRequirement]) -> Vec<Item> {
let mut out = Vec::with_capacity(requirements.len());
for req in requirements {
match self.fs.read_bytes(&req.path) {
Ok(bytes) => {
let text = String::from_utf8_lossy(&bytes).into_owned();
let body = slice_lines(&text, req.offset.unwrap_or(0), req.limit);
let path = match WorkdirPath::new(req.path.to_string_lossy()) {
Ok(path) => path,
Err(error) => {
warn!(path = %req.path.display(), %error, "invalid auto-read path");
continue;
}
};
match self
.workdir
.read(ReadRequest {
path: path.clone(),
offset: req.offset.unwrap_or(0),
limit: req.limit.unwrap_or(usize::MAX),
max_bytes: AUTO_READ_BYTE_LIMIT,
})
.await
{
Ok(result) => {
let body = String::from_utf8_lossy(&result.bytes);
let range = format_range(req.offset, req.limit);
out.push(Item::system_message(format!(
"[Auto-read file: {}{range}]\n{body}",
req.path.display()
"[Auto-read file: {path}{range}]\n{body}"
)));
}
Err(e) => {
warn!(
path = %req.path.display(),
error = %e,
"auto-read target could not be read; skipping",
);
Err(error) => {
warn!(path = %path, %error, "auto-read target could not be read; skipping")
}
}
}
out
}
/// `path` を ScopedFs 経由で解決し、submit 時の `Segment::FileRef`
/// attachment 用 system message を返す。
///
/// - `path` は relative なら cwd 相対、absolute なら absolute として解釈
/// - 通常ディレクトリは浅い entry listing として `[Dir: <path>]\n<body>` に展開する
/// - ディレクトリ listing は hidden / gitignore を特別扱いせず、scope 上 readable な
/// 直下 entry だけを最大 `DIR_FILE_REF_ENTRY_LIMIT` 件返す
/// - ファイル本文またはディレクトリ listing 本文が `max_bytes` を超える場合は切り詰める
/// - 非 UTF-8 (バイナリ) は `ResolveError::Binary` で拒否
/// - スコープ外 / NotFound / symlink directory 等は `ResolveError::Fs` で返す
pub fn resolve_file_ref(&self, path: &str, max_bytes: usize) -> Result<Item, ResolveError> {
let p = Path::new(path);
let abs = if p.is_absolute() {
p.to_path_buf()
} else {
self.fs.cwd().join(p)
};
// 通常ディレクトリだけを FileRef listing として扱う。symlink を含むパスは
// `ScopedFs::read_bytes` に委ね、既存の symlink 診断
// (`SymlinkTargetIsDirectory` / `SymlinkOutOfScope` 等) を保つ。
if first_symlink(&abs).is_none() {
let scope = self.fs.scope();
if !scope.is_readable(&abs) {
return Err(ResolveError::Fs(ToolsError::OutOfScope(abs)));
}
let meta = metadata_for_file_ref(&abs).map_err(ResolveError::Fs)?;
if meta.is_dir() {
return render_dir_file_ref(path, &abs, max_bytes, scope.as_ref());
pub async fn resolve_file_ref(
&self,
path: &str,
max_bytes: usize,
) -> Result<Item, ResolveError> {
let logical = WorkdirPath::new(path)
.map_err(ToolsError::from)
.map_err(ResolveError::Fs)?;
let stat = self
.workdir
.stat(StatRequest {
path: logical.clone(),
})
.await
.map_err(ToolsError::from)
.map_err(ResolveError::Fs)?;
if stat.kind == EntryKind::Directory {
let result = self
.workdir
.list(ListRequest {
path: logical.clone(),
limit: DIR_FILE_REF_ENTRY_LIMIT,
})
.await
.map_err(ToolsError::from)
.map_err(ResolveError::Fs)?;
let listing = result
.entries
.into_iter()
.map(|entry| match entry.kind {
EntryKind::Directory => format!("{}/", entry.path),
EntryKind::Symlink => format!("{}@", entry.path),
_ => entry.path.to_string(),
})
.collect::<Vec<_>>()
.join("\n");
let suffix = format!(
"\n[{} readable entries total, {} bytes total]{}",
result.total_entries,
result.total_bytes,
if result.truncated {
"\n[...listing truncated; use Glob for more]"
} else {
""
}
);
let header = format!("[Dir: {logical}]\n");
let listing_budget = max_bytes.saturating_sub(header.len() + suffix.len());
let (bounded_listing, truncated) = truncate_utf8_bytes(&listing, listing_budget);
let mut text = format!("{header}{bounded_listing}{suffix}");
if truncated {
text.push_str("\n[...directory attachment truncated; use Glob or Read for more]");
}
return Ok(Item::system_message(text));
}
let bytes = self.fs.read_bytes(&abs).map_err(ResolveError::Fs)?;
let total = bytes.len();
let (body_bytes, truncated) = if total > max_bytes {
(&bytes[..max_bytes], true)
} else {
(bytes.as_slice(), false)
};
let body = std::str::from_utf8(body_bytes)
.map_err(|_| ResolveError::Binary { path: abs.clone() })?;
let mut text = format!("[File: {path}]\n{body}");
if truncated {
let result = self
.workdir
.read(ReadRequest {
path: logical.clone(),
offset: 0,
limit: usize::MAX,
max_bytes,
})
.await
.map_err(ToolsError::from)
.map_err(ResolveError::Fs)?;
let total = stat.size;
let end = result.bytes.len().min(max_bytes);
let body = std::str::from_utf8(&result.bytes[..end]).map_err(|_| ResolveError::Binary {
path: PathBuf::from(logical.as_str()),
})?;
let mut text = format!("[File: {logical}]\n{body}");
if end < result.bytes.len() || result.truncated {
text.push_str(&format!(
"\n[...truncated, {total} bytes total — use read_file for the rest]"
"\n[...truncated, {total} bytes total — use Read for the rest]"
));
}
Ok(Item::system_message(text))
}
/// `prefix` にマッチするファイル / ディレクトリを scope 内で浅く列挙する。
///
/// - `prefix` が空 or `cwd` 相対のときは cwd 直下を見る
/// - `prefix` が末尾 `/` のときはそのディレクトリ直下を全列挙
/// - 末尾が名前部分のときは、その名前を starts_with でフィルタ
/// - scope 上 readable なエントリのみ返す
/// - ディレクトリ → ファイル の順、各グループ内は名前昇順
/// - 上限 `COMPLETION_LIMIT` 件で打ち切り（深い列挙はしない）
pub fn list_file_completions(&self, prefix: &str) -> Vec<FileCandidate> {
let cwd = self.fs.cwd();
let scope = self.fs.scope();
let (dir, name_prefix, is_absolute) = split_prefix(prefix, cwd);
let read_dir = match std::fs::read_dir(&dir) {
Ok(rd) => rd,
Err(_) => return Vec::new(),
pub async fn list_file_completions(&self, prefix: &str) -> Vec<FileCandidate> {
let prefix_path = Path::new(prefix);
let (parent, needle) = if prefix.ends_with('/') {
(prefix_path, String::new())
} else {
(
prefix_path.parent().unwrap_or_else(|| Path::new("")),
prefix_path
.file_name()
.map(|name| name.to_string_lossy().into_owned())
.unwrap_or_default(),
)
};
let mut out = Vec::new();
for entry in read_dir.flatten() {
let file_name = entry.file_name();
let name = file_name.to_string_lossy();
if !name.starts_with(&name_prefix) {
continue;
}
let path = entry.path();
if !scope.is_readable(&path) {
continue;
}
let is_dir = entry.file_type().map(|t| t.is_dir()).unwrap_or(false);
let display = if is_absolute {
path.display().to_string()
} else {
path.strip_prefix(cwd)
.map(|p| p.display().to_string())
.unwrap_or_else(|_| path.display().to_string())
};
out.push(FileCandidate {
path: display,
is_dir,
});
}
let Ok(parent) = WorkdirPath::new(parent.to_string_lossy()) else {
return Vec::new();
};
let Ok(result) = self
.workdir
.list(ListRequest {
path: parent,
limit: COMPLETION_LIMIT,
})
.await
else {
return Vec::new();
};
let mut out = result
.entries
.into_iter()
.filter_map(|entry| {
let name = Path::new(entry.path.as_str())
.file_name()?
.to_string_lossy();
name.starts_with(&needle).then_some(FileCandidate {
path: entry.path.to_string(),
is_dir: entry.kind == EntryKind::Directory,
})
})
.collect::<Vec<_>>();
out.sort_by(|a, b| match (a.is_dir, b.is_dir) {
(true, false) => std::cmp::Ordering::Less,
(false, true) => std::cmp::Ordering::Greater,
_ => a.path.cmp(&b.path),
});
out.truncate(COMPLETION_LIMIT);
out
}
}
@@ -224,85 +257,6 @@ pub fn slice_lines(text: &str, offset: usize, limit: Option<usize>) -> String {
lines[start..end].join("\n")
}
#[derive(Debug, Clone, PartialEq, Eq)]
struct DirListingEntry {
display: String,
kind_rank: u8,
}
fn metadata_for_file_ref(path: &Path) -> Result<std::fs::Metadata, ToolsError> {
std::fs::metadata(path).map_err(|e| match e.kind() {
std::io::ErrorKind::NotFound => ToolsError::NotFound(path.to_path_buf()),
_ => ToolsError::io(path, e),
})
}
fn render_dir_file_ref(
original_path: &str,
abs: &Path,
max_bytes: usize,
scope: &Scope,
) -> Result<Item, ResolveError> {
let read_dir = std::fs::read_dir(abs).map_err(|e| ResolveError::Fs(ToolsError::io(abs, e)))?;
let mut entries = Vec::new();
for entry in read_dir {
let entry = entry.map_err(|e| ResolveError::Fs(ToolsError::io(abs, e)))?;
let path = entry.path();
if !scope.is_readable(&path) {
continue;
}
let file_type = match entry.file_type() {
Ok(ft) => ft,
Err(e) => return Err(ResolveError::Fs(ToolsError::io(&path, e))),
};
let mut display = entry.file_name().to_string_lossy().into_owned();
let kind_rank = if file_type.is_dir() {
display.push('/');
0
} else if file_type.is_symlink() {
display.push('@');
1
} else {
2
};
entries.push(DirListingEntry { display, kind_rank });
}
entries.sort_by(|a, b| {
a.kind_rank
.cmp(&b.kind_rank)
.then_with(|| a.display.cmp(&b.display))
});
let total_entries = entries.len();
let entry_truncated = total_entries > DIR_FILE_REF_ENTRY_LIMIT;
let body = if total_entries == 0 {
"(empty directory)".to_string()
} else {
entries
.iter()
.take(DIR_FILE_REF_ENTRY_LIMIT)
.map(|e| e.display.as_str())
.collect::<Vec<_>>()
.join("\n")
};
let body_total_bytes = body.len();
let (body, byte_truncated) = truncate_utf8_bytes(&body, max_bytes);
let mut text = format!("[Dir: {original_path}]\n{body}");
if entry_truncated || byte_truncated {
text.push('\n');
text.push_str(&dir_listing_truncation_hint(
entry_truncated,
byte_truncated,
total_entries,
body_total_bytes,
));
}
Ok(Item::system_message(text))
}
fn truncate_utf8_bytes(s: &str, max_bytes: usize) -> (&str, bool) {
if s.len() <= max_bytes {
return (s, false);
@@ -314,26 +268,6 @@ fn truncate_utf8_bytes(s: &str, max_bytes: usize) -> (&str, bool) {
(&s[..end], true)
}
fn dir_listing_truncation_hint(
entry_truncated: bool,
byte_truncated: bool,
total_entries: usize,
body_total_bytes: usize,
) -> String {
match (entry_truncated, byte_truncated) {
(true, true) => format!(
"[...truncated, {total_entries} readable entries total; first {DIR_FILE_REF_ENTRY_LIMIT} entries were {body_total_bytes} bytes before byte cap — use Glob for more]"
),
(true, false) => {
format!("[...truncated, {total_entries} readable entries total — use Glob for more]")
}
(false, true) => {
format!("[...truncated, {body_total_bytes} bytes total — use Glob or Read for more]")
}
(false, false) => String::new(),
}
}
fn format_range(offset: Option<usize>, limit: Option<usize>) -> String {
match (offset, limit) {
(None, None) => String::new(),
@@ -343,41 +277,19 @@ fn format_range(offset: Option<usize>, limit: Option<usize>) -> String {
}
}
fn split_prefix(prefix: &str, cwd: &Path) -> (PathBuf, String, bool) {
let is_absolute = Path::new(prefix).is_absolute();
let p = Path::new(prefix);
let (parent, name) = if prefix.is_empty() || prefix.ends_with('/') {
(p.to_path_buf(), String::new())
} else {
let parent = p.parent().map(|p| p.to_path_buf()).unwrap_or_default();
let name = p
.file_name()
.map(|n| n.to_string_lossy().into_owned())
.unwrap_or_default();
(parent, name)
};
let dir = if is_absolute {
parent
} else if parent.as_os_str().is_empty() {
cwd.to_path_buf()
} else {
cwd.join(parent)
};
(dir, name, is_absolute)
}
#[cfg(test)]
mod tests {
use super::*;
use llm_engine::ContentPart;
use manifest::{Permission, Scope, ScopeConfig, ScopeRule};
use std::sync::Arc;
use tempfile::TempDir;
fn fs_for(dir: &TempDir) -> ScopedFs {
ScopedFs::new(
fn fs_for(dir: &TempDir) -> WorkdirHandle {
Arc::new(LocalWorkdir::new(
Scope::writable(dir.path()).unwrap(),
dir.path().to_path_buf(),
)
))
}
fn touch(path: &Path, content: &str) {
@@ -397,26 +309,28 @@ mod tests {
text
}
#[test]
fn slice_lines_handles_offset_and_limit() {
#[tokio::test]
async fn slice_lines_handles_offset_and_limit() {
let text = "a\nb\nc\nd";
assert_eq!(slice_lines(text, 0, None), "a\nb\nc\nd");
assert_eq!(slice_lines(text, 1, Some(2)), "b\nc");
assert_eq!(slice_lines(text, 10, None), "");
}
#[test]
fn render_auto_read_emits_system_messages_with_range_label() {
#[tokio::test]
async fn render_auto_read_emits_system_messages_with_range_label() {
let dir = TempDir::new().unwrap();
let file = dir.path().join("hello.txt");
std::fs::write(&file, "alpha\nbeta\ngamma\n").unwrap();
let view = WorkerFsView::new(fs_for(&dir));
let items = view.render_auto_read(&[ReadRequirement {
path: file.clone(),
offset: Some(1),
limit: Some(1),
}]);
let items = view
.render_auto_read(&[ReadRequirement {
path: PathBuf::from("hello.txt"),
offset: Some(1),
limit: Some(1),
}])
.await;
assert_eq!(items.len(), 1);
let rendered = format!("{:?}", items[0]);
@@ -426,35 +340,35 @@ mod tests {
assert!(!rendered.contains("alpha"));
}
#[test]
fn resolve_file_ref_emits_system_message_with_path_header() {
#[tokio::test]
async fn resolve_file_ref_emits_system_message_with_path_header() {
let dir = TempDir::new().unwrap();
std::fs::write(dir.path().join("hello.txt"), "hello world").unwrap();
let view = WorkerFsView::new(fs_for(&dir));
let item = view.resolve_file_ref("hello.txt", 1024).unwrap();
let item = view.resolve_file_ref("hello.txt", 1024).await.unwrap();
let text = format!("{item:?}");
assert!(text.contains("[File: hello.txt]"));
assert!(text.contains("hello world"));
assert!(!text.contains("truncated"));
}
#[test]
fn resolve_file_ref_truncates_with_hint_when_over_cap() {
#[tokio::test]
async fn resolve_file_ref_truncates_with_hint_when_over_cap() {
let dir = TempDir::new().unwrap();
let body = "x".repeat(2048);
std::fs::write(dir.path().join("big.txt"), &body).unwrap();
let view = WorkerFsView::new(fs_for(&dir));
let item = view.resolve_file_ref("big.txt", 256).unwrap();
let item = view.resolve_file_ref("big.txt", 256).await.unwrap();
let text = format!("{item:?}");
assert!(text.contains("[File: big.txt]"));
assert!(text.contains("truncated"));
assert!(text.contains("2048 bytes total"));
}
#[test]
fn resolve_file_ref_lists_directory_shallow_entries() {
#[tokio::test]
async fn resolve_file_ref_lists_directory_shallow_entries() {
let dir = TempDir::new().unwrap();
std::fs::create_dir_all(dir.path().join("docs/sub")).unwrap();
touch(&dir.path().join("docs/.hidden"), "hidden");
@@ -465,7 +379,7 @@ mod tests {
);
let view = WorkerFsView::new(fs_for(&dir));
let item = view.resolve_file_ref("docs", 4096).unwrap();
let item = view.resolve_file_ref("docs", 4096).await.unwrap();
let text = system_text(&item);
assert!(text.starts_with("[Dir: docs]\n"));
assert!(text.contains("sub/"));
@@ -481,8 +395,8 @@ mod tests {
);
}
#[test]
fn resolve_file_ref_directory_listing_filters_unreadable_entries() {
#[tokio::test]
async fn resolve_file_ref_directory_listing_filters_unreadable_entries() {
let dir = TempDir::new().unwrap();
let docs = dir.path().join("docs");
let secret = docs.join("secret");
@@ -503,25 +417,25 @@ mod tests {
}],
};
let scope = Scope::from_config(&cfg).unwrap();
let fs = ScopedFs::new(scope, dir.path().to_path_buf());
let fs: WorkdirHandle = Arc::new(LocalWorkdir::new(scope, dir.path().to_path_buf()));
let view = WorkerFsView::new(fs);
let item = view.resolve_file_ref("docs", 4096).unwrap();
let item = view.resolve_file_ref("docs", 4096).await.unwrap();
let text = system_text(&item);
assert!(text.contains("visible.txt"));
assert!(!text.contains("secret"));
assert!(!text.contains("hidden.txt"));
}
#[test]
fn resolve_file_ref_directory_listing_uses_upload_byte_cap() {
#[tokio::test]
async fn resolve_file_ref_directory_listing_uses_upload_byte_cap() {
let dir = TempDir::new().unwrap();
std::fs::create_dir(dir.path().join("docs")).unwrap();
touch(&dir.path().join("docs/very-long-file-name.txt"), "");
touch(&dir.path().join("docs/another-long-file-name.txt"), "");
let view = WorkerFsView::new(fs_for(&dir));
let item = view.resolve_file_ref("docs", 10).unwrap();
let item = view.resolve_file_ref("docs", 10).await.unwrap();
let text = system_text(&item);
assert!(text.starts_with("[Dir: docs]\n"));
assert!(text.contains("truncated"));
@@ -529,8 +443,8 @@ mod tests {
assert!(text.contains("use Glob or Read for more"));
}
#[test]
fn resolve_file_ref_directory_listing_uses_completion_entry_limit() {
#[tokio::test]
async fn resolve_file_ref_directory_listing_uses_completion_entry_limit() {
let dir = TempDir::new().unwrap();
std::fs::create_dir(dir.path().join("docs")).unwrap();
for i in 0..(DIR_FILE_REF_ENTRY_LIMIT + 5) {
@@ -538,7 +452,7 @@ mod tests {
}
let view = WorkerFsView::new(fs_for(&dir));
let item = view.resolve_file_ref("docs", 4096).unwrap();
let item = view.resolve_file_ref("docs", 4096).await.unwrap();
let text = system_text(&item);
assert!(text.contains("105 readable entries total"));
assert!(text.contains("file-099.txt"));
@@ -547,8 +461,8 @@ mod tests {
}
#[cfg(unix)]
#[test]
fn resolve_file_ref_directory_listing_marks_readable_symlink_entries() {
#[tokio::test]
async fn resolve_file_ref_directory_listing_marks_readable_symlink_entries() {
use std::os::unix::fs::symlink;
let dir = TempDir::new().unwrap();
@@ -557,92 +471,95 @@ mod tests {
symlink("target.txt", dir.path().join("docs/link.txt")).unwrap();
let view = WorkerFsView::new(fs_for(&dir));
let item = view.resolve_file_ref("docs", 4096).unwrap();
let item = view.resolve_file_ref("docs", 4096).await.unwrap();
let text = system_text(&item);
assert!(text.contains("link.txt@"));
}
#[test]
fn resolve_file_ref_rejects_binary_with_binary_error() {
#[tokio::test]
async fn resolve_file_ref_rejects_binary_with_binary_error() {
let dir = TempDir::new().unwrap();
std::fs::write(dir.path().join("blob.bin"), [0xff, 0xfe, 0x00, 0x80]).unwrap();
let view = WorkerFsView::new(fs_for(&dir));
let err = view.resolve_file_ref("blob.bin", 1024).unwrap_err();
let err = view.resolve_file_ref("blob.bin", 1024).await.unwrap_err();
assert!(matches!(err, ResolveError::Binary { .. }));
}
#[test]
fn resolve_file_ref_returns_fs_error_for_out_of_scope() {
#[tokio::test]
async fn resolve_file_ref_returns_fs_error_for_out_of_scope() {
let outer = TempDir::new().unwrap();
let inner = outer.path().join("scoped");
std::fs::create_dir(&inner).unwrap();
std::fs::write(outer.path().join("secret.txt"), "nope").unwrap();
let scope = Scope::writable(&inner).unwrap();
let fs = ScopedFs::new(scope, inner.clone());
let fs: WorkdirHandle = Arc::new(LocalWorkdir::new(scope, inner.clone()));
let view = WorkerFsView::new(fs);
// Absolute path outside of scope.
let outside = outer.path().join("secret.txt");
let err = view
.resolve_file_ref(outside.to_str().unwrap(), 1024)
.await
.unwrap_err();
assert!(matches!(err, ResolveError::Fs(_)));
}
#[test]
fn render_auto_read_skips_unreadable_targets() {
#[tokio::test]
async fn render_auto_read_skips_unreadable_targets() {
let dir = TempDir::new().unwrap();
let view = WorkerFsView::new(fs_for(&dir));
let items = view.render_auto_read(&[ReadRequirement {
path: dir.path().join("missing.txt"),
offset: None,
limit: None,
}]);
let items = view
.render_auto_read(&[ReadRequirement {
path: dir.path().join("missing.txt"),
offset: None,
limit: None,
}])
.await;
assert!(items.is_empty());
}
#[test]
fn list_file_completions_lists_pwd_when_prefix_empty() {
#[tokio::test]
async fn list_file_completions_lists_pwd_when_prefix_empty() {
let dir = TempDir::new().unwrap();
touch(&dir.path().join("alpha.rs"), "");
touch(&dir.path().join("beta.rs"), "");
std::fs::create_dir(dir.path().join("subdir")).unwrap();
let view = WorkerFsView::new(fs_for(&dir));
let cands = view.list_file_completions("");
let cands = view.list_file_completions("").await;
// ディレクトリ first
let names: Vec<&str> = cands.iter().map(|c| c.path.as_str()).collect();
assert_eq!(names, vec!["subdir", "alpha.rs", "beta.rs"]);
assert!(cands[0].is_dir);
}
#[test]
fn list_file_completions_filters_by_name_prefix() {
#[tokio::test]
async fn list_file_completions_filters_by_name_prefix() {
let dir = TempDir::new().unwrap();
touch(&dir.path().join("alpha.rs"), "");
touch(&dir.path().join("beta.rs"), "");
let view = WorkerFsView::new(fs_for(&dir));
let cands = view.list_file_completions("al");
let cands = view.list_file_completions("al").await;
assert_eq!(cands.len(), 1);
assert_eq!(cands[0].path, "alpha.rs");
}
#[test]
fn list_file_completions_descends_into_subdir_with_trailing_slash() {
#[tokio::test]
async fn list_file_completions_descends_into_subdir_with_trailing_slash() {
let dir = TempDir::new().unwrap();
touch(&dir.path().join("sub/x.rs"), "");
touch(&dir.path().join("sub/y.rs"), "");
let view = WorkerFsView::new(fs_for(&dir));
let cands = view.list_file_completions("sub/");
let cands = view.list_file_completions("sub/").await;
let names: Vec<&str> = cands.iter().map(|c| c.path.as_str()).collect();
assert_eq!(names, vec!["sub/x.rs", "sub/y.rs"]);
}
#[test]
fn list_file_completions_filters_out_non_readable_under_scope() {
#[tokio::test]
async fn list_file_completions_filters_out_non_readable_under_scope() {
let dir = TempDir::new().unwrap();
let secret = dir.path().join("secret");
std::fs::create_dir(&secret).unwrap();
@@ -662,25 +579,23 @@ mod tests {
}],
};
let scope = Scope::from_config(&cfg).unwrap();
let fs = ScopedFs::new(scope, dir.path().to_path_buf());
let fs: WorkdirHandle = Arc::new(LocalWorkdir::new(scope, dir.path().to_path_buf()));
let view = WorkerFsView::new(fs);
let cands = view.list_file_completions("");
let cands = view.list_file_completions("").await;
let names: Vec<&str> = cands.iter().map(|c| c.path.as_str()).collect();
assert!(names.contains(&"visible.rs"));
assert!(!names.contains(&"secret"));
}
#[test]
fn list_file_completions_supports_absolute_prefix() {
#[tokio::test]
async fn list_file_completions_rejects_absolute_prefix() {
let dir = TempDir::new().unwrap();
touch(&dir.path().join("a.rs"), "");
let view = WorkerFsView::new(fs_for(&dir));
let prefix = format!("{}/", dir.path().display());
let cands = view.list_file_completions(&prefix);
assert_eq!(cands.len(), 1);
assert!(cands[0].path.starts_with('/'));
assert!(cands[0].path.ends_with("a.rs"));
let cands = view.list_file_completions(&prefix).await;
assert!(cands.is_empty());
}
}
crates/worker/src/ipc/protocol_session.rs
+1 -1
View File
@@ -65,7 +65,7 @@ pub async fn dispatch_worker_protocol_method(
) -> Option<Event> {
match method {
Method::ListCompletions { kind, prefix } => {
let entries = handle.completion_entries(kind, &prefix);
let entries = handle.completion_entries(kind, &prefix).await;
Some(Event::Completions { kind, entries })
}
method => {
crates/worker/src/shared_state.rs
+4 -5
View File
@@ -23,11 +23,10 @@ pub struct WorkerSharedState {
pub greeting: protocol::Greeting,
pub status: RwLock<WorkerStatus>,
/// Worker-from-the-inside view of the filesystem. Set once in
/// `WorkerController::start` after the `ScopedFs` is materialised, and
/// read from the IPC server layer to answer `ListCompletions`
/// queries without going through the controller. `None` until set
/// (only relevant for unit tests that build a `WorkerSharedState`
/// directly without spinning up a controller).
/// `WorkerController::start` after the local Workdir provider is
/// materialised, and read from the IPC server layer to answer
/// `ListCompletions` queries without going through the controller. It is
/// unset only in unit tests that construct `WorkerSharedState` directly.
fs_view: OnceLock<WorkerFsView>,
}
crates/worker/src/worker.rs
+70 -31
View File
@@ -76,6 +76,7 @@ use protocol::{
use tokio::net::UnixStream;
use tokio::sync::broadcast;
use tokio::task::JoinHandle;
use workdir::{LocalWorkdir, WorkdirCapabilities, WorkdirHandle};
const RESTORE_RECONCILIATION_REACHABILITY_TIMEOUT: Duration = Duration::from_millis(500);
@@ -741,13 +742,15 @@ pub struct Worker<C: LlmClient, St: Store> {
/// Explicit local filesystem authority, or `None` for Workers with no
/// local cwd and no filesystem/Bash tool surface.
filesystem_authority: WorkerFilesystemAuthority,
/// Live Workdir provider derived once from the Worker–Workdir binding.
/// Local tools, file views, and compaction workers clone this handle.
workdir: Option<WorkdirHandle>,
/// Path-free workspace identity/client context injected by Runtime/host.
/// This never grants local filesystem authority.
workspace_context: WorkerWorkspaceContext,
/// Shared, atomically-swappable view of the Worker's resolved scope.
/// Cloned out to `ScopedFs` instances (builtin tools, fs_view,
/// compact worker) so scope updates propagate to every consumer
/// at the next permission check.
/// Cloned into local Workdir providers used by builtin tools, fs_view,
/// and compaction so updates propagate at the next permission check.
scope: SharedScope,
/// Filesystem authority this Worker may pass to spawned children. Direct tools
/// continue to use `scope`; SpawnWorker validates requested child scope here.
@@ -923,6 +926,7 @@ impl<C: LlmClient + Clone + 'static, St: Store + Clone + 'static> Worker<C, St>
worker_metadata_writer: None,
segment_state: self.segment_state.clone(),
filesystem_authority: self.filesystem_authority.clone(),
workdir: self.workdir.clone(),
workspace_context: self.workspace_context.clone(),
scope: self.scope.clone(),
delegation_scope: self.delegation_scope.clone(),
@@ -1110,6 +1114,8 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
let prompts = PromptCatalog::builtins_only()?;
let delegation_scope =
DelegationScope::from_config(&manifest.delegation_scope).map_err(WorkerError::Scope)?;
let scope = SharedScope::new(scope);
let workdir = workdir_from_authority(&filesystem_authority, &scope);
let mut worker = Self {
manifest,
engine: Some(worker),
@@ -1117,8 +1123,9 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
worker_metadata_writer: None,
segment_state: SegmentState::new(session_id, segment_id, 0),
filesystem_authority,
workdir,
workspace_context,
scope: SharedScope::new(scope),
scope,
delegation_scope,
hook_builder: HookRegistryBuilder::new(),
interceptor_installed: false,
@@ -1231,6 +1238,17 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
self.filesystem_authority.as_local()
}
pub fn workdir(&self) -> Option<&WorkdirHandle> {
self.workdir.as_ref()
}
/// Replace the constructor fallback with the provider binding resolved by
/// the owning Runtime. Runtime calls this before the Worker controller is
/// spawned, so tools only ever observe the Runtime-bound handle.
pub fn bind_workdir(&mut self, workdir: Option<WorkdirHandle>) {
self.workdir = workdir;
}
/// Path-free workspace identity, if Runtime/host associated this Worker
/// with a workspace.
pub fn workspace_id(&self) -> Option<&WorkspaceId> {
@@ -2063,7 +2081,7 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
// Resolve `@<path>` file refs to system messages stashed for the
// WorkerInterceptor to attach right after the user message. Resolution
// failures are non-fatal alerts.
let attachments = self.resolve_file_refs(&input);
let attachments = self.resolve_file_refs(&input).await;
let flattened = self.flatten_segments(&input);
if !attachments.is_empty() {
*self
@@ -2094,8 +2112,8 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
/// directory) surface as `AlertLevel::Warn` Alerts and are skipped — the
/// unresolved placeholder stays in the flattened user message so the LLM
/// still sees the intent.
fn resolve_file_refs(&self, segments: &[Segment]) -> Vec<SystemItem> {
let Some(local) = self.local_working_directory() else {
async fn resolve_file_refs(&self, segments: &[Segment]) -> Vec<SystemItem> {
let Some(workdir) = self.workdir.clone() else {
for seg in segments {
if let Segment::FileRef { path } = seg {
self.alert(
@@ -2107,16 +2125,16 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
}
return Vec::new();
};
let view = crate::fs_view::WorkerFsView::new(tools::ScopedFs::with_shared_scope(
self.scope.clone(),
local.cwd.clone(),
));
let view = crate::fs_view::WorkerFsView::new(workdir);
let mut out = Vec::new();
for seg in segments {
let Segment::FileRef { path } = seg else {
continue;
};
match view.resolve_file_ref(path, self.manifest.engine.file_upload.max_bytes) {
match view
.resolve_file_ref(path, self.manifest.engine.file_upload.max_bytes)
.await
{
Ok(item) => {
// `resolve_file_ref` returns an `Item::system_message`
// whose text already carries the `[File: <path>]` or
@@ -2872,13 +2890,10 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
auto_read_budget,
)));
// Build an independent compact worker. When the main Worker has local
// filesystem authority, compact-time reads go through the same scope
// and cwd policy. No-workdir Workers deliberately omit compact-time
// filesystem tools as well.
let scoped_fs = self
.local_working_directory()
.map(|local| tools::ScopedFs::with_shared_scope(self.scope.clone(), local.cwd.clone()));
// Build an independent compact worker. It clones the main Worker's
// provider handle, so compact-time reads use the same Workdir instance.
// No-workdir Workers deliberately omit compact-time filesystem tools.
let workdir = self.workdir.clone();
let summary_tracker = tools::Tracker::new();
let summary_client: Box<dyn LlmClient> = self.build_compactor_client()?;
let summary_system_prompt = self
@@ -2916,9 +2931,9 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
// Tools: read_file (shared scope, fresh tracker), bounded session
// history exploration, and compact-specific tools that populate `ctx`.
let compact_target_items = Arc::new(items_to_summarise.clone());
if let Some(scoped_fs) = scoped_fs.clone() {
summary_worker.register_tool(tools::read_tool(scoped_fs.clone(), summary_tracker));
summary_worker.register_tool(mark_read_required_tool(scoped_fs, ctx.clone()));
if let Some(workdir) = workdir.clone() {
summary_worker.register_tool(tools::read_tool(workdir.clone(), summary_tracker));
summary_worker.register_tool(mark_read_required_tool(workdir, ctx.clone()));
}
summary_worker.register_tool(search_session_log_tool(compact_target_items.clone()));
summary_worker.register_tool(read_session_items_tool(compact_target_items));
@@ -2998,12 +3013,13 @@ impl<C: LlmClient, St: Store> Worker<C, St> {
// logged and skipped inside `render_auto_read` rather than
// aborting compaction — a missing / moved file should not fail
// the whole compact.
let auto_read_messages = scoped_fs
.clone()
.map(|scoped_fs| {
WorkerFsView::new(scoped_fs).render_auto_read(&final_ctx.read_required)
})
.unwrap_or_default();
let auto_read_messages = if let Some(workdir) = workdir {
WorkerFsView::new(workdir)
.render_auto_read(&final_ctx.read_required)
.await
} else {
Vec::new()
};
// Reference list as a single system message; omitted when empty.
let reference_message = (!final_ctx.references.is_empty()).then(|| {
@@ -3940,6 +3956,8 @@ where
apply_worker_manifest(&mut worker, &manifest.engine);
worker.set_cache_key(Some(segment_id.to_string()));
let worker_metadata_writer = Some(worker_metadata_writer_for_store(&store));
let scope = SharedScope::new(common.scope);
let workdir = workdir_from_authority(&common.filesystem_authority, &scope);
let mut worker = Self {
manifest,
@@ -3948,8 +3966,9 @@ where
worker_metadata_writer,
segment_state: SegmentState::new(session_id, segment_id, 0),
filesystem_authority: common.filesystem_authority,
workdir,
workspace_context: common.workspace_context,
scope: SharedScope::new(common.scope),
scope,
delegation_scope: common.delegation_scope,
hook_builder: HookRegistryBuilder::new(),
interceptor_installed: false,
@@ -4046,6 +4065,8 @@ where
apply_worker_manifest(&mut worker, &manifest.engine);
worker.set_cache_key(Some(segment_id.to_string()));
let worker_metadata_writer = Some(worker_metadata_writer_for_store(&store));
let scope = SharedScope::new(common.scope);
let workdir = workdir_from_authority(&common.filesystem_authority, &scope);
let mut worker = Self {
manifest,
@@ -4054,8 +4075,9 @@ where
worker_metadata_writer,
segment_state: SegmentState::new(session_id, segment_id, 0),
filesystem_authority: common.filesystem_authority,
workdir,
workspace_context: common.workspace_context,
scope: SharedScope::new(common.scope),
scope,
delegation_scope: common.delegation_scope,
hook_builder: HookRegistryBuilder::new(),
interceptor_installed: false,
@@ -4335,6 +4357,8 @@ where
let extract_pointer = memory::extract::fold_pointer(&state.extensions);
let task_feature = TaskFeature::from_history(&state.history);
let worker_metadata_writer = Some(worker_metadata_writer_for_store(&store));
let scope = SharedScope::new(common.scope);
let workdir = workdir_from_authority(&common.filesystem_authority, &scope);
let mut worker = Self {
manifest,
@@ -4343,8 +4367,9 @@ where
worker_metadata_writer,
segment_state: SegmentState::new(session_id, segment_id, state.entries_count),
filesystem_authority: common.filesystem_authority,
workdir,
workspace_context: common.workspace_context,
scope: SharedScope::new(common.scope),
scope,
delegation_scope: common.delegation_scope,
hook_builder: HookRegistryBuilder::new(),
interceptor_installed: false,
@@ -5008,6 +5033,20 @@ pub enum WorkerError {
},
}
fn workdir_from_authority(
authority: &WorkerFilesystemAuthority,
scope: &SharedScope,
) -> Option<WorkdirHandle> {
authority.as_local().map(|local| {
Arc::new(LocalWorkdir::materialized(
local.root.clone(),
local.cwd.clone(),
scope.clone(),
WorkdirCapabilities::ALL,
)) as WorkdirHandle
})
}
/// Bundle of resources that every high-level Worker constructor needs:
/// filesystem authority, path-free workspace context, scope, an LLM client, the prompt catalog,
/// and (optionally) a parsed system-prompt template. Built once by
crates/worker/tests/controller_test.rs
+37
View File
@@ -11,6 +11,7 @@ use llm_engine::llm_client::{ClientError, LlmClient, Request};
use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
use session_store::{CombinedStore, FsWorkerStore};
use session_store::{FsStore, LogEntry};
use workdir::{CommandRequest, LocalWorkdir, WorkdirCapabilities, WorkdirError, WorkdirHandle};
use worker::{
Event, Method, Worker, WorkerController, WorkerFilesystemAuthority, WorkerHandle,
@@ -213,6 +214,42 @@ async fn spawn_controller(worker: Worker<MockClient, TestStore>) -> WorkerHandle
handle
}
#[tokio::test]
async fn shutdown_closes_bound_workdir_commands() {
let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await;
let workdir: WorkdirHandle = Arc::new(LocalWorkdir::materialized_bound(
Some("controller-test-workdir".to_owned()),
pwd.clone(),
pwd,
worker.scope().clone(),
WorkdirCapabilities::ALL,
));
let command = workdir
.start_command(CommandRequest {
command: "sleep 30".to_owned(),
timeout_secs: 60,
output_limit: 1024,
})
.await
.unwrap();
worker.bind_workdir(Some(Arc::clone(&workdir)));
let runtime_base = tempfile::tempdir().unwrap();
let (handle, shutdown_rx) = WorkerController::spawn(worker, runtime_base.path())
.await
.unwrap();
handle.send(Method::Shutdown).await.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), shutdown_rx)
.await
.expect("controller should shut down")
.expect("controller shutdown signal should remain open");
assert!(matches!(
workdir.command_status(command).await,
Err(WorkdirError::UnknownCommand(_))
));
}
async fn wait_for_status(handle: &WorkerHandle, status: WorkerStatus) {
let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(2);
loop {