From 21bd089a23de0494429c907a660bd8acffb8cfa2 Mon Sep 17 00:00:00 2001 From: Hare Date: Wed, 19 Aug 2026 09:45:56 +0900 Subject: [PATCH] feat: delegate subworker access through workdir sessions --- Cargo.lock | 1 + crates/worker/Cargo.toml | 1 + crates/worker/src/controller.rs | 44 +-- .../src/feature/builtin/manage_workdir.rs | 94 ++++- crates/worker/src/spawn/registry.rs | 10 + crates/worker/src/spawn/tool.rs | 344 ++++++++---------- crates/workspace-server/src/server.rs | 73 +++- 7 files changed, 347 insertions(+), 220 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 95ae1d04..ed14f7e5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -6067,6 +6067,7 @@ dependencies = [ "config-source", "dotenv", "flow", + "fs-operation", "fs4", "futures", "futures-util", diff --git a/crates/worker/Cargo.toml b/crates/worker/Cargo.toml index 27e3dae0..23af9dd8 100644 --- a/crates/worker/Cargo.toml +++ b/crates/worker/Cargo.toml @@ -33,6 +33,7 @@ config-source = { path = "../config-source" } include_dir = "0.7.4" fs4 = { workspace = true, features = ["sync"] } flow = { path = "../flow" } +fs-operation = { workspace = true } libc = { workspace = true } schemars = { workspace = true } ticket = { workspace = true } diff --git a/crates/worker/src/controller.rs b/crates/worker/src/controller.rs index e952a46b..5c96edbb 100644 --- a/crates/worker/src/controller.rs +++ b/crates/worker/src/controller.rs @@ -681,18 +681,20 @@ 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 feature_config = worker.manifest().feature.clone(); - if feature_config.manage_workdir.enabled { - if let Some(existing) = worker.workdir_session().cloned() { - existing.close().await.map_err(std::io::Error::other)?; - } + if feature_config.manage_workdir.enabled && worker.workdir_session().is_none() { let workspace_client = worker.workspace_client_handle(); - worker.bind_workdir_session(Some( + worker.bind_workdir_session(Some(workdir::delegation_capable_session( crate::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession::handle( workspace_client, ), - )); + ))); + } + if feature_config.sub_worker.enabled + && let Some(existing) = worker.workdir_session().cloned() + && !existing.is_delegation_capable() + { + worker.bind_workdir_session(Some(workdir::delegation_capable_session(existing))); } let worker_workdir = worker.workdir_session().cloned(); let local_filesystem = worker.local_working_directory().cloned(); @@ -844,6 +846,7 @@ where } let host_worker_observation_provider = worker.worker_observation_provider(); + let source_workdir_session = worker.workdir_session().cloned(); { let workspace_client = worker.workspace_client_handle(); let engine = worker.engine_mut(); @@ -902,27 +905,18 @@ where Arc, > = Vec::new(); - // Worker-orchestration tools (SubWorkerSpawn + three control tools) share - // the Worker-scoped `SpawnedWorkerRegistry` (also consumed by the main - // loop's `WorkerEvent` handler). Expose them only behind the explicit - // profile feature and require delegation authority up front so enabling - // the surface cannot imply broad child scope by accident. + // Worker-orchestration tools derive child filesystem authority from the + // active provider-backed Workdir session. The tool remains registered + // without one so invocation fails deterministically until the parent + // attaches a Workdir. if feature_config.sub_worker.enabled { let spawner_cwd = local_filesystem .as_ref() .map(|local| local.cwd.clone()) - .ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "worker spawn tools require local Worker filesystem authority", - ) - })?; - let spawner_workspace_root = local_workspace_root.clone().ok_or_else(|| { - std::io::Error::new( - std::io::ErrorKind::InvalidInput, - "worker spawn tools require local Worker filesystem authority", - ) - })?; + .unwrap_or_else(|| PathBuf::from("/")); + let spawner_workspace_root = local_workspace_root + .clone() + .unwrap_or_else(|| PathBuf::from("/")); engine.register_tool(sub_worker_spawn_tool( spawner_name.clone(), spawner_workspace_context, @@ -930,9 +924,9 @@ where runtime_base.clone(), spawner_workspace_root, spawner_cwd.clone(), + source_workdir_session, spawned_registry.clone(), spawner_manifest, - scope_handle, prompts, )); observation_providers.push(Arc::new( diff --git a/crates/worker/src/feature/builtin/manage_workdir.rs b/crates/worker/src/feature/builtin/manage_workdir.rs index 1f592124..57cf359b 100644 --- a/crates/worker/src/feature/builtin/manage_workdir.rs +++ b/crates/worker/src/feature/builtin/manage_workdir.rs @@ -16,7 +16,8 @@ use serde_json::json; use workdir::http::{WorkdirSessionOperation, WorkdirSessionOperationResult}; use workdir::workspace::{ WorkingDirectoryDetailResponse as WorkdirDetailResponse, - WorkingDirectoryListResponse as WorkdirListResponse, + WorkingDirectoryListResponse as WorkdirListResponse, WorkspaceWorkdirSessionFence, + WorkspaceWorkdirSessionOperationRequest, }; use workdir::{ CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest, @@ -153,6 +154,7 @@ struct WorkspaceHttpWorkdirBackend { pub struct WorkspaceAttachedWorkdirSession { client: Arc, workdir: Workdir, + expected_session_fence: Option, } impl WorkspaceAttachedWorkdirSession { @@ -160,6 +162,7 @@ impl WorkspaceAttachedWorkdirSession { Arc::new(Self { client, workdir: Workdir::new("workspace-attachment"), + expected_session_fence: None, }) } @@ -176,7 +179,11 @@ impl WorkspaceAttachedWorkdirSession { "/api/w/{}/workers/self/workdir-session/operations", encode_path_segment(workspace_id) ), - serde_json::to_string(&operation).map_err(|error| { + serde_json::to_string(&WorkspaceWorkdirSessionOperationRequest { + expected_session_fence: self.expected_session_fence.clone(), + operation, + }) + .map_err(|error| { WorkdirError::Transport(format!( "failed to encode Workspace Workdir operation: {error}" )) @@ -224,6 +231,43 @@ impl WorkdirSession for WorkspaceAttachedWorkdirSession { WorkdirSessionCapabilities::ALL } + async fn capture_delegation_source(&self) -> Result { + let expected_session_fence = if let Some(fence) = &self.expected_session_fence { + fence.clone() + } else { + let workspace_id = self.client.workspace_id().ok_or_else(|| { + WorkdirError::Unavailable("Workspace identity is unavailable".to_string()) + })?; + let response = self + .client + .execute(WorkspaceRequest { + method: WorkspaceRequestMethod::Get, + path: format!( + "/api/w/{}/workers/self/workdir-session/fence", + encode_path_segment(workspace_id) + ), + body: None, + }) + .map_err(|error| { + WorkdirError::Unavailable(format!( + "failed to capture Workdir attachment fence: {error}" + )) + })?; + let fence: WorkspaceWorkdirSessionFence = serde_json::from_str(&response.body) + .map_err(|error| { + WorkdirError::Unavailable(format!( + "invalid Workdir attachment fence response: {error}" + )) + })?; + fence.value + }; + Ok(Arc::new(Self { + client: self.client.clone(), + workdir: self.workdir.clone(), + expected_session_fence: Some(expected_session_fence), + })) + } + async fn stat(&self, request: StatRequest) -> Result { match self.operate(WorkdirSessionOperation::Stat(request))? { WorkdirSessionOperationResult::Stat(result) => Ok(result), @@ -1002,11 +1046,55 @@ mod tests { ); let body: serde_json::Value = serde_json::from_str(requests[0].body.as_deref().unwrap()).unwrap(); - assert_eq!(body["operation"], "stat"); + assert_eq!(body["operation"]["operation"], "stat"); + assert!(body.get("expected_session_fence").is_none()); assert!(body.get("runtime_id").is_none()); assert!(body.get("session_id").is_none()); } + #[tokio::test] + async fn delegated_attached_session_carries_captured_fence_on_operations() { + let client = Arc::new(RecordingWorkspaceClient::new(vec![ + response(json!({"value": "attachment-fence"})), + response(json!({ + "operation": "stat", + "result": {"path": "visible.txt", "kind": "file", "size": 8} + })), + ])); + let parent = workdir::delegation_capable_session(WorkspaceAttachedWorkdirSession::handle( + client.clone(), + )); + let delegation = parent + .delegate(workdir::WorkdirDelegationRequest { + rules: vec![workdir::WorkdirDelegationRule { + target: workdir::WorkdirPath::new("visible.txt").unwrap(), + permission: workdir::WorkdirDelegationPermission::Read, + recursive: false, + }], + cwd: workdir::WorkdirPath::new("visible.txt").unwrap(), + }) + .await + .unwrap(); + delegation + .scoped_session + .stat(StatRequest { + path: workdir::WorkdirPath::new("visible.txt").unwrap(), + }) + .await + .unwrap(); + + let requests = client.requests(); + assert_eq!(requests.len(), 2); + assert_eq!( + requests[0].path, + "/api/w/workspace%2Ftest/workers/self/workdir-session/fence" + ); + let body: serde_json::Value = + serde_json::from_str(requests[1].body.as_deref().unwrap()).unwrap(); + assert_eq!(body["expected_session_fence"], "attachment-fence"); + assert_eq!(body["operation"]["operation"], "stat"); + } + #[test] fn invalid_or_extra_inputs_are_rejected_before_workspace_request() { let client = Arc::new(RecordingWorkspaceClient::new(Vec::new())); diff --git a/crates/worker/src/spawn/registry.rs b/crates/worker/src/spawn/registry.rs index 58c514b0..fd9b461b 100644 --- a/crates/worker/src/spawn/registry.rs +++ b/crates/worker/src/spawn/registry.rs @@ -19,6 +19,7 @@ use session_store::{ WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerStoreError, }; use tracing::warn; +use workdir::WorkdirDelegation; use crate::internal_worker::InternalWorkerSessionHandle; use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord}; @@ -28,6 +29,9 @@ use crate::runtime::worker_allocation; pub(crate) struct InternalSpawnedWorkerRecord { pub worker_name: String, pub scope_delegated: Vec, + pub workdir_delegation: Arc, + #[cfg(test)] + pub installed_tools: Arc<[String]>, pub session: InternalWorkerSessionHandle, scope_reclaimed: Arc, } @@ -36,11 +40,16 @@ impl InternalSpawnedWorkerRecord { pub(crate) fn new( worker_name: String, scope_delegated: Vec, + workdir_delegation: WorkdirDelegation, + #[cfg(test)] installed_tools: Vec, session: InternalWorkerSessionHandle, ) -> Self { Self { worker_name, scope_delegated, + workdir_delegation: Arc::new(workdir_delegation), + #[cfg(test)] + installed_tools: installed_tools.into(), session, scope_reclaimed: Arc::new(AtomicBool::new(false)), } @@ -247,6 +256,7 @@ impl SpawnedWorkerRegistry { if !record.claim_scope_reclaim() { return Ok(false); } + record.workdir_delegation.release(); let result = if let Some(parent_scope) = &self.parent_scope { parent_scope .update(|current| current.with_removed_deny_rules(delegated_write_rules(record))) diff --git a/crates/worker/src/spawn/tool.rs b/crates/worker/src/spawn/tool.rs index 26dae685..93ec63f2 100644 --- a/crates/worker/src/spawn/tool.rs +++ b/crates/worker/src/spawn/tool.rs @@ -10,16 +10,21 @@ use std::sync::Arc; use arc_swap::ArcSwap; use async_trait::async_trait; +use fs_operation::FsPath; use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput}; use manifest::{ - CompactionConfigPartial, DelegationScope, EngineManifestConfig, FileUploadLimitsPartial, - Permission, PermissionConfigPartial, ProfileDiscovery, ProfileError, ProfileRegistry, - ProfileRegistrySource, ProfileResolveOptions, ProfileResolver, ProfileSelector, Scope, - ScopeConfig, ScopeRule, SessionConfigPartial, SharedScope, ToolOutputLimitsPartial, - WorkerManifest, WorkerManifestConfig, WorkerMetaConfig, + CompactionConfigPartial, EngineManifestConfig, FileUploadLimitsPartial, Permission, + PermissionConfigPartial, ProfileDiscovery, ProfileError, ProfileRegistry, + ProfileRegistrySource, ProfileResolveOptions, ProfileResolver, ProfileSelector, ScopeConfig, + ScopeRule, SessionConfigPartial, ToolOutputLimitsPartial, WorkerManifest, WorkerManifestConfig, + WorkerMetaConfig, }; use serde::Deserialize; use tokio::sync::mpsc; +use workdir::{ + WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule, + WorkdirSessionHandle, +}; use crate::PromptCatalogSource; use crate::controller::register_worker_tools; @@ -270,6 +275,8 @@ pub struct SubWorkerSpawnTool { /// Directory the spawned SubWorker's tools should use when the LLM did not /// override it. Defaults to the spawner's cwd. spawner_cwd: PathBuf, + /// Active provider-backed Workdir session from which child leases are captured. + source_workdir_session: Option, /// Parent-owned in-memory registry shared by the five SubWorker tools. registry: Arc, /// Spawner's resolved Manifest. `profile = "inherit"` derives the @@ -279,18 +286,6 @@ pub struct SubWorkerSpawnTool { prompt_loader: PromptCatalogSource, /// Compact selector list shared by tool description and diagnostics. available_profiles: AvailableProfiles, - /// Spawner's runtime scope. After a successful spawn, the - /// `Permission::Write` rules in the delegated scope are revoked - /// from the spawner's in-memory view (a `deny(Write, target)` is - /// pushed on top, downgrading the spawner's effective access on - /// those paths to `Read`). Mirrors the worker-allocation's - /// `effective_write` semantics: Write is the only permission - /// tracked across Workers, so revocation only touches Write. - spawner_scope: SharedScope, - /// Filesystem scope this Worker is allowed to subdelegate to children. - /// This is intentionally separate from `spawner_scope`, which authorizes - /// the current Worker's own direct tools. - delegation_scope: DelegationScope, internal_client_override: Option>, } @@ -308,12 +303,11 @@ impl SubWorkerSpawnTool { runtime_base: PathBuf, workspace_root: PathBuf, spawner_cwd: PathBuf, + source_workdir_session: Option, registry: Arc, spawner_manifest: WorkerManifest, prompt_loader: PromptCatalogSource, available_profiles: AvailableProfiles, - spawner_scope: SharedScope, - delegation_scope: DelegationScope, ) -> Self { Self { spawner_name, @@ -322,12 +316,11 @@ impl SubWorkerSpawnTool { runtime_base, workspace_root, spawner_cwd, + source_workdir_session, registry, spawner_manifest, prompt_loader, available_profiles, - spawner_scope, - delegation_scope, internal_client_override: None, } } @@ -386,8 +379,16 @@ impl Tool for SubWorkerSpawnTool { .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; let scope_allow = parse_scope(&input.scope)?; - self.validate_delegation_scope(&scope_allow)?; - let child_cwd = validate_spawn_cwd(input.cwd.as_deref(), &scope_allow, &self.spawner_cwd)?; + let source_workdir_session = + require_active_workdir_session(self.source_workdir_session.as_ref())?; + let delegation_request = + self.workdir_delegation_request(input.cwd.as_deref(), &scope_allow)?; + let workdir_delegation = source_workdir_session + .delegate(delegation_request) + .await + .map_err(|error| { + ToolError::InvalidArgument(format!("delegate Workdir session: {error}")) + })?; let spawn_selector = parse_spawn_profile_selector(input.profile.as_deref()).map_err(|msg| { @@ -413,11 +414,14 @@ impl Tool for SubWorkerSpawnTool { allow: scope_allow.clone(), deny: Vec::new(), }; - let child_manifest = + let mut child_manifest = WorkerManifest::try_from(WorkerManifestConfig::builtin_defaults().merge(child_config)) .map_err(|error| { ToolError::ExecutionFailed(format!("resolve child manifest: {error}")) })?; + // Delegated children stay bound to their scoped session and cannot use + // Workspace attachment tools to replace it with parent-level authority. + child_manifest.feature.manage_workdir.enabled = false; let reviewer_capability = input.review.as_ref().map(|review| { ( review.ticket_id.clone(), @@ -458,8 +462,7 @@ impl Tool for SubWorkerSpawnTool { self.workspace_context.clone() }; let store = EphemeralSessionStore::default(); - let filesystem_authority = - WorkerFilesystemAuthority::local(self.workspace_root.clone(), child_cwd.clone()); + let filesystem_authority = WorkerFilesystemAuthority::None; let mut child = Worker::, EphemeralSessionStore>::from_internal_manifest_with_context( child_manifest, store.clone(), @@ -472,6 +475,7 @@ 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())); let child_scope = child.scope().clone(); let child_registry = SpawnedWorkerRegistry::new_internal(input.name.clone(), child_scope); register_worker_tools( @@ -488,23 +492,14 @@ impl Tool for SubWorkerSpawnTool { .map_err(|error| { ToolError::ExecutionFailed(format!("install Internal Worker features: {error}")) })?; - // Transfer delegated Write authority before the child accepts its first turn. This closes - // the parallel-tool window where parent and child could otherwise both write the same path. - // The machine-wide allocation remains owned by the parent Worker; no fake child PID/socket - // identity is introduced. - let revoke_write: Vec = scope_allow - .iter() - .filter(|rule| rule.permission == Permission::Write) - .cloned() + #[cfg(test)] + let installed_tools = child + .engine() + .tool_server_handle() + .tool_definitions_sorted() + .into_iter() + .map(|definition| definition.name) .collect(); - if !revoke_write.is_empty() { - self.spawner_scope - .update(|current| current.with_added_deny_rules(revoke_write.clone())) - .map_err(|error| { - ToolError::ExecutionFailed(format!("revoke spawner scope: {error}")) - })?; - } - let child_name = input.name.clone(); let registry = Arc::downgrade(&self.registry); let parent_notifications = self.parent_notifications.clone(); @@ -530,19 +525,9 @@ impl Tool for SubWorkerSpawnTool { })), ) .await; - let session = match session_result { - Ok(session) => session, - Err(error) => { - if !revoke_write.is_empty() { - let _ = self - .spawner_scope - .update(|current| current.with_removed_deny_rules(revoke_write.clone())); - } - return Err(ToolError::ExecutionFailed(format!( - "prepare Internal Worker session: {error}" - ))); - } - }; + let session = session_result.map_err(|error| { + ToolError::ExecutionFailed(format!("prepare Internal Worker session: {error}")) + })?; if let Some((ticket_id, capability_token)) = &reviewer_capability { let workspace_id = self.workspace_context.workspace_id().ok_or_else(|| { @@ -605,15 +590,13 @@ impl Tool for SubWorkerSpawnTool { let record = crate::spawn::registry::InternalSpawnedWorkerRecord::new( input.name.clone(), scope_allow, + workdir_delegation, + #[cfg(test)] + installed_tools, session.clone(), ); if let Err(error) = name_reservation.commit(record) { let _ = session.stop().await; - if !revoke_write.is_empty() { - let _ = self - .spawner_scope - .update(|current| current.with_removed_deny_rules(revoke_write)); - } return Err(ToolError::ExecutionFailed(format!( "register Internal Worker session: {error}" ))); @@ -636,27 +619,73 @@ impl Tool for SubWorkerSpawnTool { } impl SubWorkerSpawnTool { - fn validate_delegation_scope(&self, scope_allow: &[ScopeRule]) -> Result<(), ToolError> { - if self.delegation_scope.is_empty() && !scope_allow.is_empty() { - return Err(ToolError::InvalidArgument( - "SubWorkerSpawn requires delegation authority, but this Worker has no delegation scope grant; direct filesystem scope only authorizes this Worker's own tools".into(), - )); + fn workdir_delegation_request( + &self, + cwd: Option<&Path>, + scope_allow: &[ScopeRule], + ) -> Result { + let rules = scope_allow + .iter() + .map(|rule| { + Ok(WorkdirDelegationRule { + target: self.logical_workdir_path(&rule.target)?, + permission: match rule.permission { + Permission::Read => WorkdirDelegationPermission::Read, + Permission::Write => WorkdirDelegationPermission::Write, + }, + recursive: rule.recursive, + }) + }) + .collect::, ToolError>>()?; + let cwd = cwd.unwrap_or(&self.spawner_cwd); + if !cwd.is_absolute() { + return Err(ToolError::InvalidArgument(format!( + "cwd must be absolute, got `{}`", + cwd.display() + ))); } - for rule in scope_allow { - let allowed = self - .delegation_scope - .allows_rule(rule) - .map_err(|error| ToolError::InvalidArgument(error.to_string()))?; - if !allowed { - return Err(ToolError::InvalidArgument(format!( - "requested child scope {} {:?} is outside this Worker's delegation scope grant", - rule.target.display(), - rule.permission - ))); - } - } - Ok(()) + Ok(WorkdirDelegationRequest { + rules, + cwd: self.logical_workdir_path(cwd)?, + }) } + + fn logical_workdir_path(&self, path: &Path) -> Result { + let logical = if self.workspace_root == Path::new("/") { + path.strip_prefix(Path::new("/")) + } else { + path.strip_prefix(&self.workspace_root) + } + .map_err(|_| { + ToolError::InvalidArgument(format!( + "scope target `{}` is not a Workdir-owned logical path", + path.display() + )) + })?; + let logical = logical.to_str().ok_or_else(|| { + ToolError::InvalidArgument(format!( + "scope target `{}` is not valid UTF-8", + path.display() + )) + })?; + FsPath::new(logical).map_err(|error| { + ToolError::InvalidArgument(format!( + "scope target `{}` is not a valid logical Workdir path: {error}", + path.display() + )) + }) + } +} + +fn require_active_workdir_session( + session: Option<&WorkdirSessionHandle>, +) -> Result<&WorkdirSessionHandle, ToolError> { + session.ok_or_else(|| { + ToolError::InvalidArgument( + "SubWorkerSpawn requires an active Workdir session; attach a Workdir before delegating filesystem access" + .to_string(), + ) + }) } fn parse_scope(rules: &[ScopeRuleInput]) -> Result, ToolError> { @@ -681,63 +710,6 @@ fn parse_scope(rules: &[ScopeRuleInput]) -> Result, ToolError> { .collect() } -fn validate_spawn_cwd( - cwd: Option<&Path>, - scope_allow: &[ScopeRule], - default_cwd: &Path, -) -> Result { - let Some(cwd) = cwd else { - return Ok(default_cwd.to_path_buf()); - }; - if !cwd.is_absolute() { - return Err(ToolError::InvalidArgument(format!( - "SubWorkerSpawn.cwd must be absolute: {}", - cwd.display() - ))); - } - let metadata = std::fs::metadata(cwd).map_err(|e| { - if e.kind() == std::io::ErrorKind::NotFound { - ToolError::InvalidArgument(format!( - "SubWorkerSpawn.cwd does not exist: {}", - cwd.display() - )) - } else { - ToolError::InvalidArgument(format!( - "SubWorkerSpawn.cwd is not usable: {}: {e}", - cwd.display() - )) - } - })?; - if !metadata.is_dir() { - return Err(ToolError::InvalidArgument(format!( - "SubWorkerSpawn.cwd must be a directory: {}", - cwd.display() - ))); - } - let canonical = std::fs::canonicalize(cwd).map_err(|e| { - ToolError::InvalidArgument(format!( - "SubWorkerSpawn.cwd is not usable: {}: {e}", - cwd.display() - )) - })?; - let child_scope = Scope::from_config(&ScopeConfig { - allow: scope_allow.to_vec(), - deny: Vec::new(), - }) - .map_err(|e| { - ToolError::InvalidArgument(format!( - "requested child scope cannot validate SubWorkerSpawn.cwd: {e}" - )) - })?; - if !child_scope.is_readable(&canonical) { - return Err(ToolError::InvalidArgument(format!( - "SubWorkerSpawn.cwd {} is outside the child's delegated readable scope; cwd grants no authority, so add an explicit read or write scope rule covering it", - cwd.display() - ))); - } - Ok(canonical) -} - /// Serialise the internal manifest config that gets handed to the child /// Worker runtime process via the hidden `--spawn-config-json` flag. /// `WorkerManifestConfig`'s `Serialize` impl is the single source of truth for the @@ -944,9 +916,9 @@ pub(crate) fn sub_worker_spawn_tool( runtime_base: PathBuf, workspace_root: PathBuf, spawner_cwd: PathBuf, + source_workdir_session: Option, registry: Arc, spawner_manifest: WorkerManifest, - spawner_scope: SharedScope, prompts: Arc>, ) -> ToolDefinition { sub_worker_spawn_tool_impl( @@ -956,9 +928,9 @@ pub(crate) fn sub_worker_spawn_tool( runtime_base, workspace_root, spawner_cwd, + source_workdir_session, registry, spawner_manifest, - spawner_scope, prompts, ) } @@ -970,9 +942,9 @@ fn sub_worker_spawn_tool_impl( runtime_base: PathBuf, workspace_root: PathBuf, spawner_cwd: PathBuf, + source_workdir_session: Option, registry: Arc, spawner_manifest: WorkerManifest, - spawner_scope: SharedScope, prompts: Arc>, ) -> ToolDefinition { Arc::new(move || { @@ -1002,13 +974,11 @@ fn sub_worker_spawn_tool_impl( runtime_base.clone(), workspace_root.clone(), spawner_cwd.clone(), + source_workdir_session.clone(), registry.clone(), spawner_manifest.clone(), prompts.load_full().source(), available_profiles, - spawner_scope.clone(), - DelegationScope::from_config(&spawner_manifest.delegation_scope) - .expect("resolved Worker manifest has a valid delegation scope"), )); (meta, tool) }) @@ -1017,6 +987,7 @@ fn sub_worker_spawn_tool_impl( #[cfg(test)] mod tests { use super::*; + use manifest::{DelegationScope, Scope, SharedScope}; use std::pin::Pin; use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::time::Duration; @@ -1035,6 +1006,16 @@ mod tests { WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceResponse, }; + #[test] + fn missing_active_workdir_session_fails_deterministically() { + let error = require_active_workdir_session(None).unwrap_err(); + assert!(matches!( + error, + ToolError::InvalidArgument(message) + if message.contains("requires an active Workdir session") + )); + } + #[test] fn reviewer_handoff_requires_explicit_builtin_profile_and_read_only_scope() { let valid: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({ @@ -1132,6 +1113,15 @@ extract_threshold = 4000 let fail_requests = Arc::new(AtomicBool::new(false)); let prompt_loader = PromptCatalogSource::builtins_only(); let (parent_method_tx, mut parent_method_rx) = mpsc::channel(8); + let source_workdir_session = workdir::delegation_capable_session(Arc::new( + workdir::LocalWorkdirSession::materialized_bound( + workdir::Workdir::new("test-workdir"), + workspace_root.clone(), + workspace_root.clone(), + spawner_scope.clone(), + workdir::WorkdirSessionCapabilities::ALL, + ), + )); let tool = SubWorkerSpawnTool::new( "parent".into(), workspace_context, @@ -1139,12 +1129,11 @@ extract_threshold = 4000 runtime.path().to_path_buf(), workspace_root.clone(), workspace_root.clone(), + Some(source_workdir_session), registry.clone(), manifest.clone(), prompt_loader, available_profiles, - spawner_scope.clone(), - DelegationScope::from_config(&manifest.delegation_scope).unwrap(), ) .with_internal_client(Box::new(ScriptedInternalClient { calls: calls.clone(), @@ -1161,7 +1150,7 @@ extract_threshold = 4000 "task": "review immutable commit", "scope": [{ "target": workspace_root.clone(), - "permission": "write", + "permission": "read", "recursive": true }] }); @@ -1188,16 +1177,30 @@ extract_threshold = 4000 .await .expect("spawn project reviewer as Internal Worker"); assert!(output.summary.contains("internal worker `reviewer-child`")); - assert!(!spawner_scope.snapshot().is_writable(&workspace_root)); + assert!(spawner_scope.snapshot().is_writable(&workspace_root)); let record = registry .get_internal("reviewer-child") .expect("Internal reviewer registry record"); + assert!(record.installed_tools.iter().any(|name| name == "Read")); + for denied in ["Write", "Edit", "Bash"] { + assert!( + !record.installed_tools.iter().any(|name| name == denied), + "read-only child unexpectedly received {denied}: {:?}", + record.installed_tools + ); + } + assert!( + !record + .installed_tools + .iter() + .any(|name| matches!(name.as_str(), "WorkdirAttachSelf" | "WorkdirDetachSelf")) + ); assert_eq!( record.session.wait_until_idle().await, crate::internal_worker::InternalWorkerSessionStatus::Idle ); assert_eq!(calls.load(Ordering::SeqCst), 1); - assert!(observed_parent_write_revoked.load(Ordering::SeqCst)); + assert!(!observed_parent_write_revoked.load(Ordering::SeqCst)); assert!(observed_instruction_override.load(Ordering::SeqCst)); let completion = tokio::time::timeout(Duration::from_secs(1), parent_method_rx.recv()) .await @@ -1295,7 +1298,11 @@ extract_threshold = 4000 assert_eq!(calls.load(Ordering::SeqCst), 3); assert!( spawner_scope.snapshot().is_writable(&workspace_root), - "Failed terminal child must automatically reclaim its delegated write scope" + "Failed terminal child must release its delegated Workdir session" + ); + assert!( + !record.workdir_delegation.is_active(), + "failed child must revoke cloned scoped sessions" ); assert!(registry.get_internal("reviewer-child").is_some()); @@ -1315,7 +1322,7 @@ extract_threshold = 4000 ) .await .unwrap(); - assert!(!spawner_scope.snapshot().is_writable(&workspace_root)); + assert!(spawner_scope.snapshot().is_writable(&workspace_root)); drop(list); drop(send); drop(stop); @@ -1343,45 +1350,6 @@ extract_threshold = 4000 ); } - #[test] - fn spawn_worker_validate_cwd_requires_absolute_existing_directory_in_child_scope() { - let root = TempDir::new().unwrap(); - let child_cwd = root.path().join("child"); - std::fs::create_dir(&child_cwd).unwrap(); - let file_path = root.path().join("file.txt"); - std::fs::write(&file_path, "not a dir").unwrap(); - let outside = TempDir::new().unwrap(); - let missing = root.path().join("missing"); - let rules = vec![abs_rule(root.path(), Permission::Write)]; - - assert_eq!( - validate_spawn_cwd(None, &rules, root.path()).unwrap(), - root.path() - ); - assert_eq!( - validate_spawn_cwd(Some(&child_cwd), &rules, root.path()).unwrap(), - std::fs::canonicalize(&child_cwd).unwrap() - ); - - for (cwd, expected) in [ - (Path::new("relative"), "must be absolute"), - (missing.as_path(), "does not exist"), - (file_path.as_path(), "must be a directory"), - ( - outside.path(), - "outside the child's delegated readable scope", - ), - ] { - let err = validate_spawn_cwd(Some(cwd), &rules, root.path()).unwrap_err(); - match err { - ToolError::InvalidArgument(message) => { - assert!(message.contains(expected), "{message}") - } - other => panic!("expected InvalidArgument, got {other:?}"), - } - } - } - #[test] fn orchestration_delegation_allows_root_read_and_worktree_writes_not_root_writes() { let tmp = TempDir::new().unwrap(); diff --git a/crates/workspace-server/src/server.rs b/crates/workspace-server/src/server.rs index 9aacc7b4..01c5a983 100644 --- a/crates/workspace-server/src/server.rs +++ b/crates/workspace-server/src/server.rs @@ -49,7 +49,8 @@ use workdir::workspace::{ WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse, WorkingDirectoryDiagnostic, WorkingDirectoryDiagnosticSeverity, WorkingDirectoryListResponse as BrowserWorkingDirectoryListResponse, WorkingDirectoryOccupancy, - WorkingDirectoryStatusKind, WorkingDirectorySummary, + WorkingDirectoryStatusKind, WorkingDirectorySummary, WorkspaceWorkdirSessionFence, + WorkspaceWorkdirSessionOperationRequest, }; use worker::feature::builtin::{WorkerObservationSubject, WorkerObservationSubjectRef}; use worker_runtime::resource::{BackendResourceError, BackendResourceFetchRequest}; @@ -1523,6 +1524,10 @@ pub fn build_router(api: WorkspaceApi) -> Router { post(scoped_attach_current_worker_workdir) .delete(scoped_detach_current_worker_workdir), ) + .route( + "/api/w/{workspace_id}/workers/self/workdir-session/fence", + get(scoped_current_worker_workdir_session_fence), + ) .route( "/api/w/{workspace_id}/workers/self/workdir-session/operations", post(scoped_execute_current_worker_workdir_operation), @@ -5043,7 +5048,7 @@ async fn scoped_attach_current_worker_workdir( worker: worker.clone(), workdir_id: workdir_id.to_string(), role: "attachment".to_string(), - linked_at: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true), + linked_at: Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true), unlinked_at: None, })?; if let Err(error) = open_current_worker_workdir_session_locked(&api, &worker, &link).await { @@ -5086,19 +5091,55 @@ async fn scoped_detach_current_worker_workdir( })) } +async fn scoped_current_worker_workdir_session_fence( + State(api): State, + AxumPath(path): AxumPath, + headers: HeaderMap, +) -> ApiResult> { + validate_workspace_scope(&api, &path.workspace_id)?; + let worker = current_worker_identity(&api, &path.workspace_id, &headers)?; + let session_lock = current_worker_session_lock(&api, &worker); + let _session_guard = session_lock.lock().await; + let link = current_worker_active_attachment(&api, &worker)?; + Ok(Json(WorkspaceWorkdirSessionFence { + value: current_worker_workdir_session_fence(&link), + })) +} + +fn current_worker_workdir_session_fence(link: &WorkerWorkdirLinkRecord) -> String { + format!("v1:{}\0{}", link.workdir_id, link.linked_at) +} + +fn validate_current_worker_workdir_session_fence( + link: &WorkerWorkdirLinkRecord, + expected: Option<&str>, +) -> Result<()> { + if expected.is_some_and(|expected| expected != current_worker_workdir_session_fence(link)) { + Err(Error::WorkdirAttachmentConflict( + "delegated Workdir session attachment changed".to_string(), + )) + } else { + Ok(()) + } +} + async fn scoped_execute_current_worker_workdir_operation( State(api): State, AxumPath(path): AxumPath, headers: HeaderMap, - Json(operation): Json, + Json(request): Json, ) -> ApiResult> { validate_workspace_scope(&api, &path.workspace_id)?; let worker = current_worker_identity(&api, &path.workspace_id, &headers)?; let session_lock = current_worker_session_lock(&api, &worker); let _session_guard = session_lock.lock().await; let link = current_worker_active_attachment(&api, &worker)?; + validate_current_worker_workdir_session_fence( + &link, + request.expected_session_fence.as_deref(), + )?; let session = open_current_worker_workdir_session_locked(&api, &worker, &link).await?; - let result = execute_workdir_session_operation(&session, operation) + let result = execute_workdir_session_operation(&session, request.operation) .await .map_err(|error| Error::RuntimeOperationFailed { runtime_id: worker.runtime_id.clone(), @@ -16167,6 +16208,30 @@ mod tests { assert_eq!(response.status(), StatusCode::BAD_REQUEST); } + #[test] + fn delegated_workdir_session_fence_rejects_reattached_link() { + let first = WorkerWorkdirLinkRecord { + workspace_id: "workspace-a".to_string(), + worker: workdir::workspace::RuntimeWorkerRef::new("runtime-a", "worker-a"), + workdir_id: "workdir-a".to_string(), + role: "primary".to_string(), + linked_at: "2026-01-01T00:00:00Z".to_string(), + unlinked_at: None, + }; + let expected = current_worker_workdir_session_fence(&first); + assert!(validate_current_worker_workdir_session_fence(&first, None).is_ok()); + assert!(validate_current_worker_workdir_session_fence(&first, Some(&expected)).is_ok()); + + let reattached = WorkerWorkdirLinkRecord { + linked_at: "2026-01-01T00:00:01Z".to_string(), + ..first + }; + assert!(matches!( + validate_current_worker_workdir_session_fence(&reattached, Some(&expected)), + Err(Error::WorkdirAttachmentConflict(_)) + )); + } + #[tokio::test] async fn backend_workdir_session_proxy_executes_typed_operations() { use manifest::Scope;