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

feat: spill long bash output to worker temp storage

Browse Source
This commit is contained in:
Hare
2026-08-31 16:52:27 +09:00
parent 10264b4019
commit 62eaefb1fa
22 changed files with 664 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker/examples/worker_protocol.rs
+3 -1
View File
@@ -47,7 +47,9 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
let worker = worker::Worker::from_manifest_toml(&toml, store).await?;
let runtime_tmp = tempfile::tempdir()?;
let (handle, _shutdown_rx) = WorkerController::spawn(worker, runtime_tmp.path()).await?;
let bash_output_dir = runtime_tmp.path().join("bash-output");
let (handle, _shutdown_rx) =
WorkerController::spawn(worker, runtime_tmp.path(), &bash_output_dir).await?;
// Check initial status via shared state
println!("[shared_state] {}", handle.shared_state.status_json());
crates/worker/src/bootstrap.rs
+62 -7
View File
@@ -17,9 +17,28 @@ use manifest::WorkerManifest;
#[derive(Debug, Clone)]
pub enum WorkerBootstrapLayout {
/// A direct Worker rooted below the supplied runtime base directory.
Direct { runtime_base: PathBuf },
Direct {
runtime_base: PathBuf,
bash_output_dir: PathBuf,
},
/// A runtime-managed Worker with an exact persisted run directory.
RuntimeManagedRun { run_dir: PathBuf },
RuntimeManagedRun {
run_dir: PathBuf,
bash_output_dir: PathBuf,
},
}
/// Return the temporary Bash spill directory owned by a stable Worker identity.
///
/// The directory deliberately lives outside session/run-generation storage so a
/// restarted controller for the same Worker keeps the same readable artifact
/// boundary.
pub fn bash_output_dir_for_worker_id(worker_id: impl std::fmt::Display) -> PathBuf {
std::env::temp_dir()
.join("yoi")
.join("workers")
.join(worker_id.to_string())
.join("bash-output")
}
/// Construction and controller inputs that are stable for one Worker launch.
@@ -204,12 +223,29 @@ where
{
let cleanup_session = worker.workdir_session().cloned();
let controller = match layout {
WorkerBootstrapLayout::Direct { runtime_base } => {
WorkerController::spawn_with_transport(worker, &runtime_base, transport).await
WorkerBootstrapLayout::Direct {
runtime_base,
bash_output_dir,
} => {
WorkerController::spawn_with_transport(
worker,
&runtime_base,
&bash_output_dir,
transport,
)
.await
}
WorkerBootstrapLayout::RuntimeManagedRun { run_dir } => {
WorkerController::spawn_runtime_managed_run_with_transport(worker, &run_dir, transport)
.await
WorkerBootstrapLayout::RuntimeManagedRun {
run_dir,
bash_output_dir,
} => {
WorkerController::spawn_runtime_managed_run_with_transport(
worker,
&run_dir,
&bash_output_dir,
transport,
)
.await
}
};
@@ -227,3 +263,22 @@ where
}
}
}
#[cfg(test)]
mod tests {
use super::bash_output_dir_for_worker_id;
#[test]
fn bash_output_directory_is_stable_per_worker_below_system_temp() {
let path = bash_output_dir_for_worker_id("019c1234-worker");
assert_eq!(
path,
std::env::temp_dir()
.join("yoi")
.join("workers")
.join("019c1234-worker")
.join("bash-output")
);
}
}
crates/worker/src/controller.rs
+52 -23
View File
@@ -222,6 +222,7 @@ impl WorkerController {
pub async fn spawn<C, St>(
worker: Worker<C, St>,
runtime_base: &Path,
bash_output_dir: &Path,
) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error>
where
C: LlmClient + Clone + 'static,
@@ -230,6 +231,7 @@ impl WorkerController {
Self::spawn_inner(
worker,
runtime_base,
bash_output_dir,
false,
None,
WorkerControllerTransport::UnixSocket,
@@ -242,24 +244,9 @@ impl WorkerController {
pub async fn spawn_with_transport<C, St>(
worker: Worker<C, St>,
runtime_base: &Path,
bash_output_dir: &Path,
transport: WorkerControllerTransport,
) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error>
where
C: LlmClient + Clone + 'static,
St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static,
{
Self::spawn_inner(worker, runtime_base, false, None, transport).await
}
/// Spawn a Worker owned by `worker-runtime`.
///
/// The controller still uses an ephemeral directory for Unix sockets and
/// tool spill artifacts, but does not write legacy pid/status/manifest
/// liveness projections.
pub async fn spawn_runtime_managed<C, St>(
worker: Worker<C, St>,
runtime_base: &Path,
) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error>
where
C: LlmClient + Clone + 'static,
St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static,
@@ -267,6 +254,33 @@ impl WorkerController {
Self::spawn_inner(
worker,
runtime_base,
bash_output_dir,
false,
None,
transport,
)
.await
}
/// Spawn a Worker owned by `worker-runtime`.
///
/// The controller uses an ephemeral directory for Unix sockets while tool
/// spill artifacts use the separately supplied Worker-owned temporary path.
/// Runtime-managed Workers do not write legacy pid/status/manifest liveness
/// projections.
pub async fn spawn_runtime_managed<C, St>(
worker: Worker<C, St>,
runtime_base: &Path,
bash_output_dir: &Path,
) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error>
where
C: LlmClient + Clone + 'static,
St: Store + WorkerMetadataStore + Clone + Send + Sync + 'static,
{
Self::spawn_inner(
worker,
runtime_base,
bash_output_dir,
true,
None,
WorkerControllerTransport::UnixSocket,
@@ -278,6 +292,7 @@ impl WorkerController {
pub async fn spawn_runtime_managed_run<C, St>(
worker: Worker<C, St>,
run_dir: &Path,
bash_output_dir: &Path,
) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error>
where
C: LlmClient + Clone + 'static,
@@ -286,6 +301,7 @@ impl WorkerController {
Self::spawn_runtime_managed_run_with_transport(
worker,
run_dir,
bash_output_dir,
WorkerControllerTransport::UnixSocket,
)
.await
@@ -296,6 +312,7 @@ impl WorkerController {
pub async fn spawn_runtime_managed_run_with_transport<C, St>(
worker: Worker<C, St>,
run_dir: &Path,
bash_output_dir: &Path,
transport: WorkerControllerTransport,
) -> Result<(WorkerHandle, ShutdownReceiver), std::io::Error>
where
@@ -305,12 +322,21 @@ impl WorkerController {
let parent = run_dir
.parent()
.ok_or_else(|| std::io::Error::other("run path has no parent"))?;
Self::spawn_inner(worker, parent, true, Some(run_dir), transport).await
Self::spawn_inner(
worker,
parent,
bash_output_dir,
true,
Some(run_dir),
transport,
)
.await
}
async fn spawn_inner<C, St>(
worker: Worker<C, St>,
runtime_base: &Path,
bash_output_dir: &Path,
runtime_managed: bool,
runtime_run: Option<&Path>,
transport: WorkerControllerTransport,
@@ -323,6 +349,7 @@ impl WorkerController {
let result = Self::spawn_initialized(
worker,
runtime_base,
bash_output_dir,
runtime_managed,
runtime_run,
transport,
@@ -340,6 +367,7 @@ impl WorkerController {
async fn spawn_initialized<C, St>(
mut worker: Worker<C, St>,
runtime_base: &Path,
bash_output_dir: &Path,
runtime_managed: bool,
runtime_run: Option<&Path>,
transport: WorkerControllerTransport,
@@ -397,11 +425,11 @@ impl WorkerController {
worker.attach_internal_worker_registry(spawned_registry.clone());
worker.attach_working_event_tx(working_event_tx.clone());
// Bash spills long outputs to a per-worker subdir under the runtime
// dir. Push a recursive `allow(Read)` for that path into the
// Worker's runtime scope so the agent can `Read` saved files
// without polluting the workspace.
let bash_output_dir = runtime_dir.path().join("bash-output");
// Bash spill artifacts are owned by the stable Worker identity rather
// than a controller session/run generation. Push a recursive
// `allow(Read)` for the exact tool output path into the Worker's shared
// runtime scope so the Workdir session and system prompt stay aligned.
let bash_output_dir = bash_output_dir.to_path_buf();
std::fs::create_dir_all(&bash_output_dir).map_err(|e| {
std::io::Error::other(format!(
"create bash output dir {}: {e}",
@@ -880,7 +908,7 @@ where
.register_tools(tools::core_builtin_tools(
workdir.clone(),
tracker.clone(),
bash_output_dir,
bash_output_dir.clone(),
));
if feature_config.image.enabled && model_supports_image_attachments(&spawner_manifest.model)
{
@@ -1103,6 +1131,7 @@ where
spawner_workspace_context,
parent_notifications,
runtime_base.clone(),
bash_output_dir.clone(),
spawner_workspace_root,
source_workdir_session,
spawned_registry.clone(),
crates/worker/src/entrypoint.rs
+2
View File
@@ -634,10 +634,12 @@ async fn run_cli_inner(cli: Cli) -> ExitCode {
return ExitCode::FAILURE;
}
};
let bash_output_dir = crate::bash_output_dir_for_worker_id(&worker_name);
let started = match start_worker_controller(
worker,
WorkerBootstrapLayout::Direct {
runtime_base: runtime_base.clone(),
bash_output_dir,
},
WorkerControllerTransport::UnixSocket,
)
crates/worker/src/lib.rs
+1 -1
View File
@@ -27,7 +27,7 @@ mod worker;
pub use bootstrap::{
BootstrappedWorker, PreparedWorker, WorkerBootstrap, WorkerBootstrapError,
WorkerBootstrapLayout, start_worker_controller,
WorkerBootstrapLayout, bash_output_dir_for_worker_id, start_worker_controller,
};
pub use compact::token_counter::{EstimateSource, SplitPoint, TokenEstimate};
pub use controller::{ShutdownReceiver, WorkerController, WorkerControllerTransport, WorkerHandle};
crates/worker/src/spawn/tool.rs
+50 -8
View File
@@ -22,7 +22,7 @@ use manifest::{
use serde::Deserialize;
use tokio::sync::mpsc;
use workdir::{
WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule,
WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule, WorkdirPath,
WorkdirSessionHandle,
};
@@ -258,9 +258,10 @@ pub struct SubWorkerSpawnTool {
spawner_name: String,
workspace_context: crate::worker::WorkerWorkspaceContext,
parent_notifications: ParentNotificationTarget,
/// Runtime-owned root used only for bounded Internal Worker tool artifacts such as Bash spill
/// output. It is not an Internal Worker identity or catalog location.
/// Runtime-owned root used for Internal Worker controller state.
runtime_base: PathBuf,
/// Parent Worker-owned temporary root used for bounded Bash spill output.
bash_output_dir: PathBuf,
/// Inherited runtime workspace root for Profile/project/Ticket/workflow/
/// memory context. SubWorkerSpawn `cwd` must not affect this value.
workspace_root: PathBuf,
@@ -292,6 +293,7 @@ impl SubWorkerSpawnTool {
workspace_context: crate::worker::WorkerWorkspaceContext,
parent_notifications: ParentNotificationTarget,
runtime_base: PathBuf,
bash_output_dir: PathBuf,
workspace_root: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>,
registry: Arc<SpawnedWorkerRegistry>,
@@ -304,6 +306,7 @@ impl SubWorkerSpawnTool {
workspace_context,
parent_notifications,
runtime_base,
bash_output_dir,
workspace_root,
source_workdir_session,
registry,
@@ -367,7 +370,22 @@ impl Tool for SubWorkerSpawnTool {
.reserve_internal_name(input.name.clone())
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
let workdir_rules = parse_workdir_scope(&input.scope)?;
let mut workdir_rules = parse_workdir_scope(&input.scope)?;
let child_bash_output_dir = self.bash_output_dir.join("sub-workers").join(&input.name);
tokio::fs::create_dir_all(&child_bash_output_dir)
.await
.map_err(|error| {
ToolError::ExecutionFailed(format!(
"create Internal Worker Bash output directory {}: {error}",
child_bash_output_dir.display()
))
})?;
workdir_rules.push(WorkdirDelegationRule {
target: WorkdirPath::new_scoped(child_bash_output_dir.to_string_lossy())
.map_err(|error| ToolError::ExecutionFailed(error.to_string()))?,
permission: WorkdirDelegationPermission::Read,
recursive: true,
});
let source_workdir_session =
require_active_workdir_session(self.source_workdir_session.as_ref())?;
let delegation_request = workdir_delegation_request(input.cwd.as_deref(), workdir_rules)?;
@@ -464,14 +482,22 @@ impl Tool for SubWorkerSpawnTool {
.await
.map_err(|error| ToolError::ExecutionFailed(format!("build Internal Worker: {error}")))?;
child.bind_workdir_session(Some(workdir_delegation.scoped_session.clone()));
child
.add_scope_rules([ScopeRule {
target: child_bash_output_dir.clone(),
permission: manifest::Permission::Read,
recursive: true,
}])
.map_err(|error| {
ToolError::ExecutionFailed(format!(
"grant Internal Worker Bash output scope: {error}"
))
})?;
let child_scope = child.scope().clone();
let child_registry = SpawnedWorkerRegistry::new_internal(input.name.clone(), child_scope);
register_worker_tools(
&mut child,
self.runtime_base
.join("internal-workers")
.join(&input.name)
.join("bash-output"),
child_bash_output_dir,
self.runtime_base.clone(),
child_registry.clone(),
None,
@@ -883,6 +909,7 @@ pub(crate) fn sub_worker_spawn_tool(
workspace_context: crate::worker::WorkerWorkspaceContext,
parent_notifications: ParentNotificationTarget,
runtime_base: PathBuf,
bash_output_dir: PathBuf,
workspace_root: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>,
registry: Arc<SpawnedWorkerRegistry>,
@@ -894,6 +921,7 @@ pub(crate) fn sub_worker_spawn_tool(
workspace_context,
parent_notifications,
runtime_base,
bash_output_dir,
workspace_root,
source_workdir_session,
registry,
@@ -907,6 +935,7 @@ fn sub_worker_spawn_tool_impl(
workspace_context: crate::worker::WorkerWorkspaceContext,
parent_notifications: ParentNotificationTarget,
runtime_base: PathBuf,
bash_output_dir: PathBuf,
workspace_root: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>,
registry: Arc<SpawnedWorkerRegistry>,
@@ -938,6 +967,7 @@ fn sub_worker_spawn_tool_impl(
workspace_context.clone(),
parent_notifications.clone(),
runtime_base.clone(),
bash_output_dir.clone(),
workspace_root.clone(),
source_workdir_session.clone(),
registry.clone(),
@@ -1082,12 +1112,17 @@ extract_threshold = 4000
async fn reviewer_profile_write_scope_exposes_command_tools_and_notifies_parent_controller() {
let runtime = TempDir::new().unwrap();
let workspace_root = runtime.path().join("project");
let bash_output_dir = runtime.path().join("bash-output");
let available_profiles = write_project_profile_registry(
&workspace_root,
Some("reviewer"),
&[("reviewer", "reviewer.toml", INTERNAL_REVIEWER_PROFILE)],
);
let mut manifest = parent_manifest(&workspace_root, None);
manifest
.scope
.allow
.push(abs_rule(&bash_output_dir, Permission::Read));
manifest.delegation_scope = ScopeConfig {
allow: vec![abs_rule(&workspace_root, Permission::Write)],
deny: Vec::new(),
@@ -1118,6 +1153,7 @@ extract_threshold = 4000
workspace_context,
ParentNotificationTarget::Controller(parent_method_tx.downgrade()),
runtime.path().to_path_buf(),
bash_output_dir.clone(),
workspace_root.clone(),
Some(source_workdir_session),
registry.clone(),
@@ -1167,6 +1203,12 @@ extract_threshold = 4000
.await
.expect("spawn project reviewer as Internal Worker");
assert!(output.summary.contains("internal worker `reviewer-child`"));
assert!(
bash_output_dir
.join("sub-workers")
.join("reviewer-child")
.is_dir()
);
assert!(spawner_scope.snapshot().is_writable(&workspace_root));
let record = registry
.get_internal("reviewer-child")
crates/worker/tests/compact_events_test.rs
+2 -1
View File
@@ -861,7 +861,8 @@ async fn controller_compact_method_emits_start_and_done() {
]);
let worker = make_worker_with_manifest(POST_RUN_MANIFEST_TOML, client).await;
let runtime_tmp = tempfile::tempdir().unwrap();
let (handle, _shutdown) = WorkerController::spawn(worker, runtime_tmp.path())
let bash_output_dir = runtime_tmp.path().join("bash-output");
let (handle, _shutdown) = WorkerController::spawn(worker, runtime_tmp.path(), &bash_output_dir)
.await
.unwrap();
let mut rx = handle.subscribe();
crates/worker/tests/controller_test.rs
+40 -6
View File
@@ -276,12 +276,38 @@ async fn spawn_controller(worker: Worker<MockClient, TestStore>) -> WorkerHandle
let tmp = tempfile::tempdir().unwrap();
let runtime_base = tmp.path().to_owned();
std::mem::forget(tmp);
let (handle, _shutdown_rx) = WorkerController::spawn(worker, &runtime_base)
let bash_output_dir = runtime_base.join("bash-output");
let (handle, _shutdown_rx) = WorkerController::spawn(worker, &runtime_base, &bash_output_dir)
.await
.unwrap();
handle
}
#[tokio::test]
async fn controller_grants_read_scope_for_exact_bash_output_directory() {
let worker = make_worker(MockClient::new(simple_text_events())).await;
let shared_scope = worker.scope().clone();
let runtime_base = tempfile::tempdir().unwrap();
let worker_tmp = tempfile::tempdir().unwrap();
let bash_output_dir = worker_tmp.path().join("worker-1").join("bash-output");
let (handle, shutdown_rx) =
WorkerController::spawn(worker, runtime_base.path(), &bash_output_dir)
.await
.unwrap();
assert!(bash_output_dir.is_dir());
assert!(shared_scope.snapshot().allow_rules().iter().any(|rule| {
rule.target == bash_output_dir
&& rule.permission == manifest::Permission::Read
&& rule.recursive
}));
assert!(!handle.runtime_dir.path().join("bash-output").exists());
handle.send(Method::Shutdown).await.unwrap();
shutdown_rx.await.unwrap();
}
#[tokio::test]
async fn shutdown_closes_bound_workdir_session() {
let (mut worker, pwd) = make_worker_with_pwd(MockClient::new(simple_text_events())).await;
@@ -297,6 +323,7 @@ async fn shutdown_closes_bound_workdir_session() {
command: "sleep 30".to_owned(),
timeout_secs: 60,
output_limit: 1024,
spill_dir: None,
tool_call_id: None,
})
.await
@@ -304,9 +331,11 @@ async fn shutdown_closes_bound_workdir_session() {
worker.bind_workdir_session(Some(Arc::clone(&session)));
let runtime_base = tempfile::tempdir().unwrap();
let (handle, shutdown_rx) = WorkerController::spawn(worker, runtime_base.path())
.await
.unwrap();
let bash_output_dir = runtime_base.path().join("bash-output");
let (handle, shutdown_rx) =
WorkerController::spawn(worker, runtime_base.path(), &bash_output_dir)
.await
.unwrap();
handle.send(Method::Shutdown).await.unwrap();
tokio::time::timeout(std::time::Duration::from_secs(5), shutdown_rx)
.await
@@ -338,6 +367,7 @@ async fn controller_projects_workdir_command_events_and_snapshot_state() {
command: "printf ready; sleep 0.3; printf done".to_owned(),
timeout_secs: 5,
output_limit: 1024,
spill_dir: None,
tool_call_id: Some("tool-command-1".into()),
})
.await
@@ -445,6 +475,7 @@ async fn controller_refreshes_command_snapshot_after_high_output_provider_lag()
.to_owned(),
timeout_secs: 10,
output_limit: 1024,
spill_dir: None,
tool_call_id: Some("tool-high-output".into()),
})
.await
@@ -508,8 +539,9 @@ async fn controller_startup_failure_closes_bound_workdir_session() {
let invalid_runtime_base = runtime_base.path().join("not-a-directory");
std::fs::write(&invalid_runtime_base, "file").unwrap();
let bash_output_dir = runtime_base.path().join("bash-output");
assert!(
WorkerController::spawn(worker, &invalid_runtime_base)
WorkerController::spawn(worker, &invalid_runtime_base, &bash_output_dir)
.await
.is_err()
);
@@ -519,6 +551,7 @@ async fn controller_startup_failure_closes_bound_workdir_session() {
command: "printf unreachable".to_owned(),
timeout_secs: 5,
output_limit: 1024,
spill_dir: None,
tool_call_id: None,
})
.await,
@@ -863,7 +896,8 @@ permission = "write"
let client = MockClient::new(simple_text_events());
let worker = make_worker_with_pwd_and_manifest(client, manifest).await.0;
let tmp = tempfile::tempdir().unwrap();
let result = WorkerController::spawn(worker, tmp.path()).await;
let bash_output_dir = tmp.path().join("bash-output");
let result = WorkerController::spawn(worker, tmp.path(), &bash_output_dir).await;
assert!(
result.is_ok(),
"feature exposure must not imply delegation authority"