refactor: broker SubWorker Workdir tools through parent

This commit is contained in:
2026-09-06 02:17:44 +09:00
parent 1239c638a5
commit 68f00bc948
17 changed files with 715 additions and 1117 deletions
+1
View File
@@ -118,6 +118,7 @@ impl Tool for BashTool {
command: params.command, command: params.command,
timeout_secs, timeout_secs,
output_limit: INLINE_BYTE_BUDGET, output_limit: INLINE_BYTE_BUDGET,
cwd: None,
spill_dir: Some(self.output_dir.clone()), spill_dir: Some(self.output_dir.clone()),
tool_call_id: Some(call_id.clone()), tool_call_id: Some(call_id.clone()),
}) })
+3 -41
View File
@@ -68,12 +68,10 @@ pub enum WorkdirSessionOperation {
CommandCancel(CommandHandle), CommandCancel(CommandHandle),
} }
/// Wire envelope for an operation and its optional provider-enforced child scope. /// Wire envelope for one provider operation.
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct WorkdirSessionOperationRequest { pub struct WorkdirSessionOperationRequest {
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub delegations: Vec<crate::WorkdirDelegationRequest>,
pub operation: WorkdirSessionOperation, pub operation: WorkdirSessionOperation,
} }
@@ -289,7 +287,7 @@ mod client {
use reqwest::{Client, StatusCode, Url}; use reqwest::{Client, StatusCode, Url};
use super::*; use super::*;
use crate::{Workdir, WorkdirSession, WorkdirSessionHandle}; use crate::{Workdir, WorkdirSession};
/// Provides a fresh bearer token for each Runtime request. Backend /// Provides a fresh bearer token for each Runtime request. Backend
/// implementations can mint short-lived capability tokens without making a /// implementations can mint short-lived capability tokens without making a
@@ -324,7 +322,6 @@ mod client {
workdir: Workdir, workdir: Workdir,
session_id: WorkdirSessionId, session_id: WorkdirSessionId,
capabilities: WorkdirSessionCapabilities, capabilities: WorkdirSessionCapabilities,
delegations: Vec<crate::WorkdirDelegationRequest>,
closed: AtomicBool, closed: AtomicBool,
} }
@@ -377,7 +374,6 @@ mod client {
workdir: Workdir::new(opened.workdir_id.as_str()), workdir: Workdir::new(opened.workdir_id.as_str()),
session_id: opened.session_id, session_id: opened.session_id,
capabilities: opened.capabilities, capabilities: opened.capabilities,
delegations: Vec::new(),
closed: AtomicBool::new(false), closed: AtomicBool::new(false),
}) })
} }
@@ -404,10 +400,7 @@ mod client {
"operations", "operations",
], ],
)?; )?;
let operation = WorkdirSessionOperationRequest { let operation = WorkdirSessionOperationRequest { operation };
delegations: self.delegations.clone(),
operation,
};
let response = self let response = self
.client .client
.post(url) .post(url)
@@ -436,37 +429,6 @@ mod client {
self.capabilities self.capabilities
} }
fn transports_delegation_context(&self) -> bool {
true
}
async fn capture_delegation_source(
&self,
request: &crate::WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
if self.closed.load(Ordering::Acquire) {
return Err(WorkdirError::SessionClosed);
}
let mut delegations = self.delegations.clone();
delegations.push(request.clone());
let candidate = Arc::new(Self {
client: self.client.clone(),
base_url: self.base_url.clone(),
authorization: self.authorization.clone(),
workdir: self.workdir.clone(),
session_id: self.session_id.clone(),
capabilities: self.capabilities,
delegations,
closed: AtomicBool::new(false),
});
candidate
.stat(StatRequest {
path: fs_operation::FsPath::new("").expect("empty Workdir path is valid"),
})
.await?;
Ok(candidate)
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> { async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
match self.operate(WorkdirSessionOperation::Stat(request)).await? { match self.operate(WorkdirSessionOperation::Stat(request)).await? {
WorkdirSessionOperationResult::Stat(result) => Ok(result), WorkdirSessionOperationResult::Stat(result) => Ok(result),
+5 -39
View File
@@ -5,10 +5,10 @@
//! bound to one Worker. Tools consume sessions; they do not own Workdir //! bound to one Worker. Tools consume sessions; they do not own Workdir
//! materialization or cleanup. //! materialization or cleanup.
mod delegation;
pub mod http; pub mod http;
mod local; mod local;
mod operation; mod operation;
mod scope;
pub mod workspace; pub mod workspace;
use std::path::{Path, PathBuf}; use std::path::{Path, PathBuf};
@@ -18,11 +18,6 @@ use async_trait::async_trait;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use tokio::sync::broadcast; use tokio::sync::broadcast;
pub use delegation::{
AppliedWorkdirDelegation, ReadOnlyWorkdirSession, WorkdirDelegation,
WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule,
apply_delegation_chain, delegation_capable_session,
};
pub use fs_operation::{ pub use fs_operation::{
ContentHash, EditRequest, EditResult, EntryKind, FsPath as WorkdirPath, GlobRequest, ContentHash, EditRequest, EditResult, EntryKind, FsPath as WorkdirPath, GlobRequest,
GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult, GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult,
@@ -32,6 +27,10 @@ pub use local::{
LocalWorkdirSession, SymlinkInfo, WorkdirSessionResource, direct_symlink, first_symlink, LocalWorkdirSession, SymlinkInfo, WorkdirSessionResource, direct_symlink, first_symlink,
}; };
pub use operation::*; pub use operation::*;
pub use scope::{
ReadOnlyWorkdirSession, WorkdirScopeLease, WorkdirToolBroker, WorkdirToolScope,
WorkdirToolScopePermission, WorkdirToolScopeRule,
};
/// Persistent, opaque identity of one materialized Workdir. /// Persistent, opaque identity of one materialized Workdir.
#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
@@ -148,39 +147,6 @@ pub trait WorkdirSession: std::fmt::Debug + Send + Sync {
fn workdir(&self) -> &Workdir; fn workdir(&self) -> &Workdir;
fn capabilities(&self) -> WorkdirSessionCapabilities; fn capabilities(&self) -> WorkdirSessionCapabilities;
fn is_delegation_capable(&self) -> bool {
false
}
/// Whether this session transports the delegation chain to another
/// provider boundary that will apply logical cwd/path resolution there.
fn transports_delegation_context(&self) -> bool {
false
}
/// Capture a provider-specific source for a delegated child session.
/// Remote providers use this boundary to pin attachment identity without
/// exposing transport handles or host paths.
async fn capture_delegation_source(
&self,
_request: &WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
Err(WorkdirError::Denied(
"workdir provider does not support delegated sessions".into(),
))
}
/// Attenuate this session into a revocable child lease. Only sessions
/// created with [`delegation_capable_session`] implement this operation.
async fn delegate(
&self,
_request: WorkdirDelegationRequest,
) -> Result<WorkdirDelegation, WorkdirError> {
Err(WorkdirError::Denied(
"workdir session is not delegation-capable".into(),
))
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError>; async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError>;
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError>; async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError>;
async fn write(&self, request: WriteRequest) -> Result<WriteResult, WorkdirError>; async fn write(&self, request: WriteRequest) -> Result<WriteResult, WorkdirError>;
+23 -69
View File
@@ -18,7 +18,7 @@ use std::sync::{Arc, Mutex as StdMutex};
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::{Duration, SystemTime, UNIX_EPOCH};
use async_trait::async_trait; use async_trait::async_trait;
use manifest::{Permission, Scope, ScopeConfig, ScopeRule, SharedScope}; use manifest::{Scope, SharedScope};
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use tokio::process::Command; use tokio::process::Command;
use tokio::sync::{Mutex, broadcast, watch}; use tokio::sync::{Mutex, broadcast, watch};
@@ -28,10 +28,8 @@ use crate::{
CommandEvent, CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandEvent, CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest,
CommandSnapshot, CommandStatus, CommandStream, CommandStreamSlice, EditRequest, EditResult, CommandSnapshot, CommandStatus, CommandStream, CommandStreamSlice, EditRequest, EditResult,
GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, ReadRequest, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, ReadRequest,
ReadResult, StatRequest, StatResult, Workdir, WorkdirDelegationPermission, ReadResult, StatRequest, StatResult, Workdir, WorkdirError, WorkdirPath, WorkdirSession,
WorkdirDelegationRequest, WorkdirError, WorkdirPath, WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability, WriteRequest, WriteResult,
WorkdirSessionCapabilities, WorkdirSessionCapability, WorkdirSessionHandle, WriteRequest,
WriteResult,
}; };
#[cfg(test)] #[cfg(test)]
use crate::{EntryKind, WriteOutcome}; use crate::{EntryKind, WriteOutcome};
@@ -558,69 +556,6 @@ impl WorkdirSession for LocalWorkdirSession {
self.inner.capabilities self.inner.capabilities
} }
async fn capture_delegation_source(
&self,
request: &WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
let host_rules = request
.rules
.iter()
.map(|rule| ScopeRule {
target: self.inner.root.join(rule.target.as_str()),
permission: match rule.permission {
WorkdirDelegationPermission::Read => Permission::Read,
WorkdirDelegationPermission::Write => Permission::Write,
},
recursive: rule.recursive,
})
.collect::<Vec<_>>();
for (logical, host) in request.rules.iter().zip(&host_rules) {
if logical.permission == WorkdirDelegationPermission::Write {
let resolved = Scope::resolved_target(host)
.map_err(|error| WorkdirError::Denied(error.to_string()))?;
if resolved != host.target {
return Err(WorkdirError::Denied(format!(
"write delegation target `{}` traverses a symlink",
logical.target
)));
}
}
}
let parent_scope = self.inner.scope.snapshot();
for rule in &host_rules {
if !parent_scope
.allows_rule(rule)
.map_err(|error| WorkdirError::Denied(error.to_string()))?
{
return Err(WorkdirError::Denied(format!(
"delegated provider scope `{}` exceeds the parent session",
rule.target.display()
)));
}
}
let child_scope = Scope::from_config(&ScopeConfig {
allow: host_rules,
deny: Vec::new(),
})
.map_err(|error| WorkdirError::Denied(error.to_string()))?;
let child_cwd = self.inner.root.join(request.cwd.as_str());
if !child_scope.is_readable(&child_cwd)
|| !std::fs::metadata(&child_cwd).is_ok_and(|metadata| metadata.is_dir())
{
return Err(WorkdirError::Denied(format!(
"delegated cwd `{}` is not a readable Workdir directory",
request.cwd
)));
}
Ok(Arc::new(LocalWorkdirSession::materialized_bound(
self.inner.workdir.clone(),
self.inner.root.clone(),
self.inner.root.clone(),
SharedScope::new(child_scope),
self.inner.capabilities,
)))
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> { async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
self.ensure_capability(WorkdirSessionCapability::Read)?; self.ensure_capability(WorkdirSessionCapability::Read)?;
let logical = request.path.clone(); let logical = request.path.clone();
@@ -694,9 +629,20 @@ impl WorkdirSession for LocalWorkdirSession {
{ {
return Err(WorkdirError::OutOfScope(spill_dir.to_path_buf())); return Err(WorkdirError::OutOfScope(spill_dir.to_path_buf()));
} }
let cwd = if let Some(logical_cwd) = request.cwd.as_ref() {
let cwd = self.resolve(logical_cwd);
let scope = self.inner.scope.snapshot();
if !scope.is_readable(&cwd)
|| !std::fs::metadata(&cwd).is_ok_and(|metadata| metadata.is_dir())
{
return Err(WorkdirError::OutOfScope(cwd));
}
cwd
} else {
self.inner.cwd.clone()
};
let id = self.inner.next_command_id.fetch_add(1, Ordering::Relaxed); let id = self.inner.next_command_id.fetch_add(1, Ordering::Relaxed);
let handle = CommandHandle(format!("command-{id}")); let handle = CommandHandle(format!("command-{id}"));
let cwd = self.inner.cwd.clone();
let (completion_tx, completion) = watch::channel(false); let (completion_tx, completion) = watch::channel(false);
let command_id = handle.0.clone(); let command_id = handle.0.clone();
let telemetry = self.inner.command_telemetry.clone(); let telemetry = self.inner.command_telemetry.clone();
@@ -1516,6 +1462,7 @@ mod tests {
command: "sleep 30".to_owned(), command: "sleep 30".to_owned(),
timeout_secs: 60, timeout_secs: 60,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: None, tool_call_id: None,
}, },
@@ -2043,6 +1990,7 @@ mod tests {
command: "pwd && printf provider-command".into(), command: "pwd && printf provider-command".into(),
timeout_secs: 5, timeout_secs: 5,
output_limit: 4096, output_limit: 4096,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: None, tool_call_id: None,
}, },
@@ -2141,6 +2089,7 @@ mod tests {
command: "printf hidden".into(), command: "printf hidden".into(),
timeout_secs: 5, timeout_secs: 5,
output_limit: 1, output_limit: 1,
cwd: None,
spill_dir: Some(spill.path().to_path_buf()), spill_dir: Some(spill.path().to_path_buf()),
tool_call_id: None, tool_call_id: None,
}, },
@@ -2178,6 +2127,7 @@ mod tests {
command: "i=0; while [ $i -lt 200 ]; do printf 'line-%03d\\n' \"$i\"; i=$((i+1)); done; printf 'FINAL-NEEDLE\\n'".into(), command: "i=0; while [ $i -lt 200 ]; do printf 'line-%03d\\n' \"$i\"; i=$((i+1)); done; printf 'FINAL-NEEDLE\\n'".into(),
timeout_secs: 5, timeout_secs: 5,
output_limit: 64, output_limit: 64,
cwd: None,
spill_dir: Some(spill.path().to_path_buf()), spill_dir: Some(spill.path().to_path_buf()),
tool_call_id: None, tool_call_id: None,
}, },
@@ -2224,6 +2174,7 @@ mod tests {
command: "printf 'aéz'".into(), command: "printf 'aéz'".into(),
timeout_secs: 5, timeout_secs: 5,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: None, tool_call_id: None,
}, },
@@ -2449,6 +2400,7 @@ mod tests {
command: "printf ready; printf warning >&2; sleep 0.2; printf done".into(), command: "printf ready; printf warning >&2; sleep 0.2; printf done".into(),
timeout_secs: 5, timeout_secs: 5,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: Some("tool-7".into()), tool_call_id: Some("tool-7".into()),
}, },
@@ -2553,6 +2505,7 @@ mod tests {
command: "sleep 30".into(), command: "sleep 30".into(),
timeout_secs: 1, timeout_secs: 1,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: None, tool_call_id: None,
}, },
@@ -2623,6 +2576,7 @@ mod tests {
command: "sleep 30".into(), command: "sleep 30".into(),
timeout_secs: 60, timeout_secs: 60,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: None, tool_call_id: None,
}, },
+4
View File
@@ -11,6 +11,10 @@ pub struct CommandRequest {
pub command: String, pub command: String,
pub timeout_secs: u64, pub timeout_secs: u64,
pub output_limit: usize, pub output_limit: usize,
/// Workdir-relative command directory. Providers validate it against the
/// active session before process start.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub cwd: Option<fs_operation::FsPath>,
/// Provider-local directory where complete output is retained when the /// Provider-local directory where complete output is retained when the
/// inline result exceeds `output_limit`. /// inline result exceeds `output_limit`.
pub spill_dir: Option<PathBuf>, pub spill_dir: Option<PathBuf>,
File diff suppressed because it is too large Load Diff
-10
View File
@@ -104,15 +104,5 @@ mod tests {
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] #[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)] #[serde(deny_unknown_fields)]
pub struct WorkspaceWorkdirSessionOperationRequest { pub struct WorkspaceWorkdirSessionOperationRequest {
#[serde(default, skip_serializing_if = "Option::is_none")]
pub expected_session_fence: Option<String>,
#[serde(default, skip_serializing_if = "Vec::is_empty")]
pub delegations: Vec<crate::WorkdirDelegationRequest>,
pub operation: crate::http::WorkdirSessionOperation, pub operation: crate::http::WorkdirSessionOperation,
} }
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub struct WorkspaceWorkdirSessionFence {
pub value: String,
}
+4 -88
View File
@@ -718,8 +718,7 @@ async fn run_workdir_session_operation(
.ok_or_else(RuntimeHttpWorkdirError::not_found)?; .ok_or_else(RuntimeHttpWorkdirError::not_found)?;
record.session.clone() record.session.clone()
}; };
let applied = workdir::apply_delegation_chain(source, request.delegations).await?; let session = source.as_ref();
let session = applied.scoped_session.as_ref();
let operation = request.operation; let operation = request.operation;
let result = match operation { let result = match operation {
@@ -2080,8 +2079,8 @@ mod tests {
use manifest::{Scope, SharedScope}; use manifest::{Scope, SharedScope};
use tower::ServiceExt; use tower::ServiceExt;
use workdir::{ use workdir::{
GrepOutputMode, GrepRequest, LocalWorkdirSession, ReadRequest, StatRequest, Workdir, GrepOutputMode, GrepRequest, LocalWorkdirSession, StatRequest, Workdir, WorkdirPath,
WorkdirPath, WorkdirSessionCapabilities, WorkdirSessionCapabilities,
}; };
#[test] #[test]
@@ -2502,16 +2501,6 @@ mod tests {
async fn workdir_session_operations_enforce_owner_and_close_terminally() { async fn workdir_session_operations_enforce_owner_and_close_terminally() {
let temp = tempfile::tempdir().expect("tempdir"); let temp = tempfile::tempdir().expect("tempdir");
std::fs::write(temp.path().join("hello.txt"), "hello").expect("write fixture"); std::fs::write(temp.path().join("hello.txt"), "hello").expect("write fixture");
#[cfg(unix)]
{
use std::os::unix::fs::symlink;
std::fs::create_dir(temp.path().join("granted")).expect("granted directory");
std::fs::write(temp.path().join("granted/visible"), "visible")
.expect("visible fixture");
std::fs::create_dir(temp.path().join("secret")).expect("secret directory");
std::fs::write(temp.path().join("secret/key"), "hidden").expect("secret fixture");
symlink("../secret/key", temp.path().join("granted/link")).expect("symlink fixture");
}
let scope = SharedScope::new(Scope::writable(temp.path()).expect("scope")); let scope = SharedScope::new(Scope::writable(temp.path()).expect("scope"));
let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound( let session: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::materialized_bound(
Workdir::new("wd-1"), Workdir::new("wd-1"),
@@ -2545,7 +2534,6 @@ mod tests {
expires_at: u64::MAX, expires_at: u64::MAX,
}; };
let operation = WorkdirSessionOperationRequest { let operation = WorkdirSessionOperationRequest {
delegations: Vec::new(),
operation: WorkdirSessionOperation::Stat(StatRequest { operation: WorkdirSessionOperation::Stat(StatRequest {
path: WorkdirPath::new("hello.txt").expect("logical path"), path: WorkdirPath::new("hello.txt").expect("logical path"),
}), }),
@@ -2562,7 +2550,6 @@ mod tests {
assert!(matches!(result, WorkdirSessionOperationResult::Stat(_))); assert!(matches!(result, WorkdirSessionOperationResult::Stat(_)));
let grep = WorkdirSessionOperationRequest { let grep = WorkdirSessionOperationRequest {
delegations: Vec::new(),
operation: WorkdirSessionOperation::Grep(GrepRequest { operation: WorkdirSessionOperation::Grep(GrepRequest {
pattern: "hello".into(), pattern: "hello".into(),
path: WorkdirPath::new("hello.txt").unwrap(), path: WorkdirPath::new("hello.txt").unwrap(),
@@ -2585,78 +2572,7 @@ mod tests {
) )
.await .await
.expect("grep direct file through provider operation"); .expect("grep direct file through provider operation");
match result { assert!(matches!(result, WorkdirSessionOperationResult::Grep(_)));
WorkdirSessionOperationResult::Grep(result) => {
assert_eq!(result.match_count, 1);
assert_eq!(result.matched_files, 1);
assert!(result.output.starts_with("hello.txt\n"));
assert!(result.output.contains("> 1 │ hello"));
}
other => panic!("unexpected workdir grep result: {other:?}"),
}
#[cfg(unix)]
{
let delegated_visible = WorkdirSessionOperationRequest {
delegations: vec![workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: WorkdirPath::new("granted").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: true,
}],
cwd: WorkdirPath::new("granted").unwrap(),
}],
operation: WorkdirSessionOperation::Read(ReadRequest {
path: WorkdirPath::new("visible").unwrap(),
offset: 0,
limit: 20,
max_bytes: 1024,
}),
};
let visible = run_workdir_session_operation(
State(state.clone()),
Path("session-1".to_string()),
Some(Extension(auth.clone())),
Ok(Json(delegated_visible)),
)
.await
.expect("non-root delegated cwd should resolve once")
.0;
assert!(matches!(
visible,
WorkdirSessionOperationResult::Read(result) if result.bytes == b"visible"
));
let delegated_read = WorkdirSessionOperationRequest {
delegations: vec![workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: WorkdirPath::new("granted").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: true,
}],
cwd: WorkdirPath::new("granted").unwrap(),
}],
operation: WorkdirSessionOperation::Read(ReadRequest {
path: WorkdirPath::new("link").unwrap(),
offset: 0,
limit: 20,
max_bytes: 1024,
}),
};
let error = run_workdir_session_operation(
State(state.clone()),
Path("session-1".to_string()),
Some(Extension(auth.clone())),
Ok(Json(delegated_read)),
)
.await
.expect_err("provider must reject delegated symlink escape");
assert_ne!(error.status, StatusCode::OK);
assert_eq!(
std::fs::read_to_string(temp.path().join("secret/key")).unwrap(),
"hidden"
);
}
let wrong_owner = RuntimeAuthContext { let wrong_owner = RuntimeAuthContext {
workspace_id: "workspace-b".to_string(), workspace_id: "workspace-b".to_string(),
+15 -9
View File
@@ -514,6 +514,7 @@ impl WorkerController {
runtime_base.to_path_buf(), runtime_base.to_path_buf(),
spawned_registry.clone(), spawned_registry.clone(),
Some(method_tx.downgrade()), Some(method_tx.downgrade()),
None,
) )
.await?; .await?;
if let Some(session) = fs_for_view.as_ref() { if let Some(session) = fs_for_view.as_ref() {
@@ -911,6 +912,7 @@ pub(crate) async fn register_worker_tools<C, St>(
runtime_base: PathBuf, runtime_base: PathBuf,
spawned_registry: Arc<SpawnedWorkerRegistry>, spawned_registry: Arc<SpawnedWorkerRegistry>,
parent_method_tx: Option<mpsc::WeakSender<Method>>, parent_method_tx: Option<mpsc::WeakSender<Method>>,
inherited_workdir_tool_broker: Option<workdir::WorkdirToolBroker>,
) -> std::io::Result<Option<workdir::WorkdirSessionHandle>> ) -> std::io::Result<Option<workdir::WorkdirSessionHandle>>
where where
C: LlmClient + Clone + 'static, C: LlmClient + Clone + 'static,
@@ -919,21 +921,26 @@ where
// Worker-immutable snapshots taken before the mutable worker borrow // Worker-immutable snapshots taken before the mutable worker borrow
// below so the worker borrow doesn't conflict with reads on `worker`. // below so the worker borrow doesn't conflict with reads on `worker`.
let feature_config = worker.manifest().feature.clone(); let feature_config = worker.manifest().feature.clone();
let mut workdir_tool_broker = inherited_workdir_tool_broker;
if feature_config.manage_workdir.enabled && worker.workdir_session().is_none() { if feature_config.manage_workdir.enabled && worker.workdir_session().is_none() {
let workspace_client = worker.workspace_client_handle(); let workspace_client = worker.workspace_client_handle();
worker.bind_workdir_session(Some(workdir::delegation_capable_session( let broker = workdir::WorkdirToolBroker::new(
crate::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession::handle( crate::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession::handle(
workspace_client, workspace_client,
), ),
))); );
} worker.bind_workdir_session(Some(broker.tool_session()));
if feature_config.sub_worker.enabled workdir_tool_broker = Some(broker);
} else if workdir_tool_broker.is_none()
&& let Some(existing) = worker.workdir_session().cloned() && let Some(existing) = worker.workdir_session().cloned()
&& !existing.is_delegation_capable()
{ {
worker.bind_workdir_session(Some(workdir::delegation_capable_session(existing))); let broker = workdir::WorkdirToolBroker::new(existing);
worker.bind_workdir_session(Some(broker.tool_session()));
workdir_tool_broker = Some(broker);
} }
let worker_workdir = worker.workdir_session().cloned(); let worker_workdir = workdir_tool_broker
.as_ref()
.map(workdir::WorkdirToolBroker::tool_session);
let local_filesystem = worker.local_working_directory().cloned(); let local_filesystem = worker.local_working_directory().cloned();
let local_workspace_root = local_filesystem.as_ref().map(|local| local.root.clone()); let local_workspace_root = local_filesystem.as_ref().map(|local| local.root.clone());
let task_feature = worker.task_feature(); let task_feature = worker.task_feature();
@@ -1157,7 +1164,6 @@ where
} }
let host_worker_observation_provider = worker.worker_observation_provider(); 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 workspace_client = worker.workspace_client_handle();
let engine = worker.engine_mut(); let engine = worker.engine_mut();
@@ -1199,7 +1205,7 @@ where
runtime_base.clone(), runtime_base.clone(),
bash_output_dir.clone(), bash_output_dir.clone(),
spawner_workspace_root, spawner_workspace_root,
source_workdir_session, workdir_tool_broker,
spawned_registry.clone(), spawned_registry.clone(),
spawner_manifest, spawner_manifest,
prompts, prompts,
@@ -12,7 +12,7 @@ use async_trait::async_trait;
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use serde_json::json; use serde_json::json;
use workdir::http::{WorkdirSessionOperation, WorkdirSessionOperationResult}; use workdir::http::{WorkdirSessionOperation, WorkdirSessionOperationResult};
use workdir::workspace::{WorkspaceWorkdirSessionFence, WorkspaceWorkdirSessionOperationRequest}; use workdir::workspace::WorkspaceWorkdirSessionOperationRequest;
use workdir::{ use workdir::{
CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest, CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest,
EditResult, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult, EditResult, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult,
@@ -156,8 +156,6 @@ struct WorkspaceHttpWorkdirBackend {
pub struct WorkspaceAttachedWorkdirSession { pub struct WorkspaceAttachedWorkdirSession {
client: Arc<dyn WorkspaceClient>, client: Arc<dyn WorkspaceClient>,
workdir: Workdir, workdir: Workdir,
expected_session_fence: Option<String>,
delegations: Vec<workdir::WorkdirDelegationRequest>,
} }
impl WorkspaceAttachedWorkdirSession { impl WorkspaceAttachedWorkdirSession {
@@ -165,8 +163,6 @@ impl WorkspaceAttachedWorkdirSession {
Arc::new(Self { Arc::new(Self {
client, client,
workdir: Workdir::new("workspace-attachment"), workdir: Workdir::new("workspace-attachment"),
expected_session_fence: None,
delegations: Vec::new(),
}) })
} }
@@ -183,16 +179,13 @@ impl WorkspaceAttachedWorkdirSession {
"/api/w/{}/workers/self/workdir-session/operations", "/api/w/{}/workers/self/workdir-session/operations",
encode_path_segment(workspace_id) encode_path_segment(workspace_id)
), ),
serde_json::to_string(&WorkspaceWorkdirSessionOperationRequest { serde_json::to_string(&WorkspaceWorkdirSessionOperationRequest { operation }).map_err(
expected_session_fence: self.expected_session_fence.clone(), |error| {
delegations: self.delegations.clone(),
operation,
})
.map_err(|error| {
WorkdirError::Transport(format!( WorkdirError::Transport(format!(
"failed to encode Workspace Workdir operation: {error}" "failed to encode Workspace Workdir operation: {error}"
)) ))
})?, },
)?,
); );
let response = self let response = self
.client .client
@@ -241,59 +234,6 @@ impl WorkdirSession for WorkspaceAttachedWorkdirSession {
WorkdirSessionCapabilities::ALL WorkdirSessionCapabilities::ALL
} }
fn transports_delegation_context(&self) -> bool {
true
}
async fn capture_delegation_source(
&self,
request: &workdir::WorkdirDelegationRequest,
) -> Result<WorkdirSessionHandle, WorkdirError> {
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
};
let mut delegations = self.delegations.clone();
delegations.push(request.clone());
let candidate = Arc::new(Self {
client: self.client.clone(),
workdir: self.workdir.clone(),
expected_session_fence: Some(expected_session_fence),
delegations,
});
candidate
.stat(StatRequest {
path: workdir::WorkdirPath::new("").expect("empty Workdir path is valid"),
})
.await?;
Ok(candidate)
}
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> { async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
match self.operate(WorkdirSessionOperation::Stat(request))? { match self.operate(WorkdirSessionOperation::Stat(request))? {
WorkdirSessionOperationResult::Stat(result) => Ok(result), WorkdirSessionOperationResult::Stat(result) => Ok(result),
@@ -1155,6 +1095,7 @@ mod tests {
command: "true".to_string(), command: "true".to_string(),
timeout_secs: 120, timeout_secs: 120,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: Some("/worker-local/bash-output".into()), spill_dir: Some("/worker-local/bash-output".into()),
tool_call_id: Some("call-1".to_string()), tool_call_id: Some("call-1".to_string()),
}) })
@@ -1178,83 +1119,6 @@ mod tests {
); );
} }
#[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": "", "kind": "directory", "size": 0}
})),
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("").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: false,
}],
cwd: workdir::WorkdirPath::new("").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(), 3);
assert_eq!(
requests[0].path,
"/api/w/workspace%2Ftest/workers/self/workdir-session/fence"
);
let body: serde_json::Value =
serde_json::from_str(requests[2].body.as_deref().unwrap()).unwrap();
assert_eq!(body["expected_session_fence"], "attachment-fence");
assert_eq!(body["operation"]["operation"], "stat");
assert_eq!(body["delegations"][0]["rules"][0]["target"], "");
}
#[tokio::test]
async fn attached_provider_rejection_happens_before_delegation_is_returned() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![
response(json!({"value": "attachment-fence"})),
response(json!({"error": "provider rejected delegated write target"})),
]));
let parent = workdir::delegation_capable_session(WorkspaceAttachedWorkdirSession::handle(
client.clone(),
));
let result = parent
.delegate(workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: workdir::WorkdirPath::new("linked-target").unwrap(),
permission: workdir::WorkdirDelegationPermission::Write,
recursive: true,
}],
cwd: workdir::WorkdirPath::new("linked-target").unwrap(),
})
.await;
assert!(result.is_err(), "provider rejection must fail before lease");
let requests = client.requests();
assert_eq!(requests.len(), 2);
let validation: serde_json::Value =
serde_json::from_str(requests[1].body.as_deref().unwrap()).unwrap();
assert_eq!(validation["operation"]["operation"], "stat");
assert_eq!(validation["delegations"].as_array().unwrap().len(), 1);
}
#[tokio::test] #[tokio::test]
async fn attached_session_preserves_typed_provider_validation_error() { async fn attached_session_preserves_typed_provider_validation_error() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![error_response( let client = Arc::new(RecordingWorkspaceClient::new(vec![error_response(
@@ -1298,73 +1162,52 @@ mod tests {
} }
#[tokio::test] #[tokio::test]
async fn nested_attached_session_preserves_full_delegation_chain() { async fn scoped_broker_operations_carry_no_child_context() {
let client = Arc::new(RecordingWorkspaceClient::new(vec![ let client = Arc::new(RecordingWorkspaceClient::new(vec![
response(json!({"value": "attachment-fence"})),
response(json!({ response(json!({
"operation": "stat", "operation": "stat",
"result": {"path": "", "kind": "directory", "size": 0} "result": {"path": "visible.txt", "kind": "file", "size": 8}
})), })),
response(json!({ response(json!({
"operation": "stat", "operation": "stat",
"result": {"path": "nested", "kind": "directory", "size": 0} "result": {"path": "visible.txt", "kind": "file", "size": 8}
})),
response(json!({
"operation": "stat",
"result": {"path": "nested/file", "kind": "file", "size": 1}
})), })),
])); ]));
let parent = workdir::delegation_capable_session(WorkspaceAttachedWorkdirSession::handle( let broker = workdir::WorkdirToolBroker::new(WorkspaceAttachedWorkdirSession::handle(
client.clone(), client.clone(),
)); ));
let outer = parent let scoped = broker
.delegate(workdir::WorkdirDelegationRequest { .scope(workdir::WorkdirToolScope {
rules: vec![workdir::WorkdirDelegationRule { rules: vec![workdir::WorkdirToolScopeRule {
target: workdir::WorkdirPath::new("").unwrap(), target: workdir::WorkdirPath::new("").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read, permission: workdir::WorkdirToolScopePermission::Read,
recursive: true, recursive: true,
}], }],
cwd: workdir::WorkdirPath::new("").unwrap(), cwd: workdir::WorkdirPath::new("").unwrap(),
command: false,
}) })
.await .await
.unwrap(); .unwrap();
let nested = outer scoped
.scoped_session
.delegate(workdir::WorkdirDelegationRequest {
rules: vec![workdir::WorkdirDelegationRule {
target: workdir::WorkdirPath::new("nested").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read,
recursive: true,
}],
cwd: workdir::WorkdirPath::new("nested").unwrap(),
})
.await
.unwrap();
nested
.scoped_session
.stat(StatRequest { .stat(StatRequest {
path: workdir::WorkdirPath::new("file").unwrap(), path: workdir::WorkdirPath::new("visible.txt").unwrap(),
}) })
.await .await
.unwrap(); .unwrap();
let requests = client.requests(); let requests = client.requests();
assert_eq!(requests.len(), 4); assert_eq!(requests.len(), 2);
let outer_validation: serde_json::Value = for request in requests {
serde_json::from_str(requests[1].body.as_deref().unwrap()).unwrap();
let nested_validation: serde_json::Value =
serde_json::from_str(requests[2].body.as_deref().unwrap()).unwrap();
assert_eq!(outer_validation["delegations"].as_array().unwrap().len(), 1);
assert_eq!( assert_eq!(
nested_validation["delegations"].as_array().unwrap().len(), request.path,
2 "/api/w/workspace%2Ftest/workers/self/workdir-session/operations"
); );
let body: serde_json::Value = let body: serde_json::Value =
serde_json::from_str(requests[3].body.as_deref().unwrap()).unwrap(); serde_json::from_str(request.body.as_deref().unwrap()).unwrap();
assert_eq!(body["delegations"].as_array().unwrap().len(), 2); assert!(body.get("delegations").is_none());
assert_eq!(body["delegations"][0]["rules"][0]["target"], ""); assert!(body.get("child").is_none());
assert_eq!(body["delegations"][1]["rules"][0]["target"], "nested"); assert!(body.get("expected_session_fence").is_none());
assert_eq!(body["operation"]["request"]["path"], "file"); }
} }
#[test] #[test]
+6 -2
View File
@@ -709,7 +709,7 @@ pub(crate) fn prepare_internal_worker_from_spec(
} }
Box::pin(prepare_internal_worker_session( Box::pin(prepare_internal_worker_session(
worker, store, visibility, None, None, worker, store, visibility, None, None, None,
)) ))
.await .await
}) })
@@ -746,13 +746,16 @@ pub(crate) async fn prepare_internal_worker_session(
visibility: InternalWorkerVisibility, visibility: InternalWorkerVisibility,
child_registry: Option<Arc<SpawnedWorkerRegistry>>, child_registry: Option<Arc<SpawnedWorkerRegistry>>,
on_turn_end: Option<Arc<dyn Fn(InternalWorkerSessionStatus) + Send + Sync>>, on_turn_end: Option<Arc<dyn Fn(InternalWorkerSessionStatus) + Send + Sync>>,
command_event_broker: Option<workdir::WorkdirToolBroker>,
) -> Result<InternalWorkerSessionHandle, InternalWorkerSessionError> { ) -> Result<InternalWorkerSessionHandle, InternalWorkerSessionError> {
let (event_tx, _event_rx) = broadcast::channel(256); let (event_tx, _event_rx) = broadcast::channel(256);
let sink = worker.sink(); let sink = worker.sink();
spawn_internal_log_event_bridge(sink.clone(), event_tx.clone()); spawn_internal_log_event_bridge(sink.clone(), event_tx.clone());
let alerter = Alerter::new(event_tx.clone()); let alerter = Alerter::new(event_tx.clone());
let in_flight = InFlightEvents::new(event_tx.clone()); let in_flight = InFlightEvents::new(event_tx.clone());
if let Some(session) = worker.workdir_session() { if let Some(broker) = command_event_broker.as_ref() {
wire_workdir_command_events(&broker.tool_session(), &in_flight);
} else if let Some(session) = worker.workdir_session() {
wire_workdir_command_events(session, &in_flight); wire_workdir_command_events(session, &in_flight);
} }
let actor_in_flight = in_flight.clone(); let actor_in_flight = in_flight.clone();
@@ -887,6 +890,7 @@ pub(crate) async fn spawn_prepared_internal_worker_session(
InternalWorkerVisibility::ServicePrivate, InternalWorkerVisibility::ServicePrivate,
None, None,
on_turn_end, on_turn_end,
None,
) )
.await?; .await?;
handle.send(input).await?; handle.send(input).await?;
+10 -9
View File
@@ -25,7 +25,7 @@ use session_store::{
}; };
use tokio::sync::broadcast; use tokio::sync::broadcast;
use tracing::warn; use tracing::warn;
use workdir::WorkdirDelegation; use workdir::WorkdirScopeLease;
use crate::internal_worker::{InternalWorkerSessionHandle, InternalWorkerVisibility}; use crate::internal_worker::{InternalWorkerSessionHandle, InternalWorkerVisibility};
use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord}; use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord};
@@ -68,7 +68,7 @@ pub(crate) struct SubWorkerStopSummary {
pub(crate) struct InternalSpawnedWorkerRecord { pub(crate) struct InternalSpawnedWorkerRecord {
pub worker_name: String, pub worker_name: String,
pub scope_delegated: Vec<ScopeRule>, pub scope_delegated: Vec<ScopeRule>,
pub workdir_delegation: Arc<WorkdirDelegation>, pub workdir_tool_scope: Arc<WorkdirScopeLease>,
#[cfg(test)] #[cfg(test)]
pub installed_tools: Arc<[String]>, pub installed_tools: Arc<[String]>,
pub session: InternalWorkerSessionHandle, pub session: InternalWorkerSessionHandle,
@@ -86,7 +86,7 @@ impl InternalSpawnedWorkerRecord {
pub(crate) fn new( pub(crate) fn new(
worker_name: String, worker_name: String,
scope_delegated: Vec<ScopeRule>, scope_delegated: Vec<ScopeRule>,
workdir_delegation: WorkdirDelegation, workdir_tool_scope: WorkdirScopeLease,
#[cfg(test)] installed_tools: Vec<String>, #[cfg(test)] installed_tools: Vec<String>,
session: InternalWorkerSessionHandle, session: InternalWorkerSessionHandle,
change_tracker: Option<tools::Tracker>, change_tracker: Option<tools::Tracker>,
@@ -94,7 +94,7 @@ impl InternalSpawnedWorkerRecord {
Self { Self {
worker_name, worker_name,
scope_delegated, scope_delegated,
workdir_delegation: Arc::new(workdir_delegation), workdir_tool_scope: Arc::new(workdir_tool_scope),
#[cfg(test)] #[cfg(test)]
installed_tools: installed_tools.into(), installed_tools: installed_tools.into(),
session, session,
@@ -690,7 +690,7 @@ impl SpawnedWorkerRegistry {
if !record.claim_scope_reclaim() { if !record.claim_scope_reclaim() {
return Ok(false); return Ok(false);
} }
record.workdir_delegation.release(); record.workdir_tool_scope.release();
let result = if let Some(parent_scope) = &self.parent_scope { let result = if let Some(parent_scope) = &self.parent_scope {
parent_scope parent_scope
.update(|current| current.with_removed_deny_rules(delegated_write_rules(record))) .update(|current| current.with_removed_deny_rules(delegated_write_rules(record)))
@@ -966,7 +966,7 @@ mod tests {
deny: Vec::new(), deny: Vec::new(),
}) })
.unwrap(); .unwrap();
let source = workdir::delegation_capable_session(Arc::new( let source = workdir::WorkdirToolBroker::new(Arc::new(
workdir::LocalWorkdirSession::materialized_bound( workdir::LocalWorkdirSession::materialized_bound(
workdir::Workdir::new("registry-test"), workdir::Workdir::new("registry-test"),
root.clone(), root.clone(),
@@ -976,13 +976,14 @@ mod tests {
), ),
)); ));
let delegation = source let delegation = source
.delegate(workdir::WorkdirDelegationRequest { .scope(workdir::WorkdirToolScope {
rules: vec![workdir::WorkdirDelegationRule { rules: vec![workdir::WorkdirToolScopeRule {
target: workdir::WorkdirPath::new("").unwrap(), target: workdir::WorkdirPath::new("").unwrap(),
permission: workdir::WorkdirDelegationPermission::Read, permission: workdir::WorkdirToolScopePermission::Read,
recursive: true, recursive: true,
}], }],
cwd: workdir::WorkdirPath::new("").unwrap(), cwd: workdir::WorkdirPath::new("").unwrap(),
command: false,
}) })
.await .await
.unwrap(); .unwrap();
+62 -160
View File
@@ -22,8 +22,7 @@ use manifest::{
use serde::Deserialize; use serde::Deserialize;
use tokio::sync::mpsc; use tokio::sync::mpsc;
use workdir::{ use workdir::{
WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule, WorkdirPath, WorkdirToolBroker, WorkdirToolScope, WorkdirToolScopePermission, WorkdirToolScopeRule,
WorkdirSessionHandle,
}; };
use crate::PromptCatalogSource; use crate::PromptCatalogSource;
@@ -64,6 +63,9 @@ struct SubWorkerSpawnInput {
/// spawner's explicit delegation authority; direct tool scope alone is not /// spawner's explicit delegation authority; direct tool scope alone is not
/// sufficient. Omit `recursive` for normal workspace/worktree delegation; it defaults to true. /// sufficient. Omit `recursive` for normal workspace/worktree delegation; it defaults to true.
scope: Vec<ScopeRuleInput>, scope: Vec<ScopeRuleInput>,
/// Explicitly grant command execution through the parent-owned Workdir tool broker.
#[serde(default)]
command: bool,
/// Binds an actual read-only builtin Reviewer child to the current Merge Request candidate. /// Binds an actual read-only builtin Reviewer child to the current Merge Request candidate.
/// Review capability material is generated by the trusted spawn layer. /// Review capability material is generated by the trusted spawn layer.
#[serde(default)] #[serde(default)]
@@ -267,8 +269,8 @@ pub struct SubWorkerSpawnTool {
workspace_root: PathBuf, workspace_root: PathBuf,
/// Directory the spawned SubWorker's tools should use when the LLM did not /// Directory the spawned SubWorker's tools should use when the LLM did not
/// override it. Defaults to the spawner's cwd. /// override it. Defaults to the spawner's cwd.
/// Active provider-backed Workdir session from which child leases are captured. /// Parent-owned broker for scoped Workdir tool execution.
source_workdir_session: Option<WorkdirSessionHandle>, workdir_tool_broker: Option<WorkdirToolBroker>,
/// Parent-owned in-memory registry shared by the five SubWorker tools. /// Parent-owned in-memory registry shared by the five SubWorker tools.
registry: Arc<SpawnedWorkerRegistry>, registry: Arc<SpawnedWorkerRegistry>,
/// Spawner's resolved Manifest. `profile = "inherit"` derives the /// Spawner's resolved Manifest. `profile = "inherit"` derives the
@@ -295,7 +297,7 @@ impl SubWorkerSpawnTool {
runtime_base: PathBuf, runtime_base: PathBuf,
bash_output_dir: PathBuf, bash_output_dir: PathBuf,
workspace_root: PathBuf, workspace_root: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>, workdir_tool_broker: Option<WorkdirToolBroker>,
registry: Arc<SpawnedWorkerRegistry>, registry: Arc<SpawnedWorkerRegistry>,
spawner_manifest: WorkerManifest, spawner_manifest: WorkerManifest,
prompt_loader: PromptCatalogSource, prompt_loader: PromptCatalogSource,
@@ -308,7 +310,7 @@ impl SubWorkerSpawnTool {
runtime_base, runtime_base,
bash_output_dir, bash_output_dir,
workspace_root, workspace_root,
source_workdir_session, workdir_tool_broker,
registry, registry,
spawner_manifest, spawner_manifest,
prompt_loader, prompt_loader,
@@ -341,6 +343,11 @@ fn validate_reviewer_handoff(input: &SubWorkerSpawnInput) -> Result<(), ToolErro
"Merge Request Reviewer SubWorkers must include writable delegated scope".to_string(), "Merge Request Reviewer SubWorkers must include writable delegated scope".to_string(),
)); ));
} }
if !input.command {
return Err(ToolError::InvalidArgument(
"Merge Request Reviewer SubWorkers require an explicit command grant".to_string(),
));
}
Ok(()) Ok(())
} }
@@ -370,7 +377,7 @@ impl Tool for SubWorkerSpawnTool {
.reserve_internal_name(input.name.clone()) .reserve_internal_name(input.name.clone())
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?; .map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
let mut workdir_rules = parse_workdir_scope(&input.scope)?; let workdir_rules = parse_workdir_scope(&input.scope)?;
let child_bash_output_dir = self.bash_output_dir.join("sub-workers").join(&input.name); let child_bash_output_dir = self.bash_output_dir.join("sub-workers").join(&input.name);
tokio::fs::create_dir_all(&child_bash_output_dir) tokio::fs::create_dir_all(&child_bash_output_dir)
.await .await
@@ -380,28 +387,15 @@ impl Tool for SubWorkerSpawnTool {
child_bash_output_dir.display() child_bash_output_dir.display()
)) ))
})?; })?;
let source_workdir_session = let workdir_tool_broker = require_workdir_tool_broker(self.workdir_tool_broker.as_ref())?;
require_active_workdir_session(self.source_workdir_session.as_ref())?; let tool_scope = workdir_tool_scope(input.cwd.as_deref(), workdir_rules, input.command)?;
let transports_delegation_context = source_workdir_session.transports_delegation_context(); let workdir_scope = workdir_tool_broker
// Provider-transported sessions resolve every delegation rule in the .scope(tool_scope)
// receiving Workdir namespace. The Bash spill directory instead belongs
// to this Worker host, so forwarding it would widen the request with a
// foreign absolute path and fail the provider's existing scope check.
if !transports_delegation_context {
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 delegation_request = workdir_delegation_request(input.cwd.as_deref(), workdir_rules)?;
let workdir_delegation = source_workdir_session
.delegate(delegation_request)
.await .await
.map_err(|error| { .map_err(|error| {
ToolError::InvalidArgument(format!("delegate Workdir session: {error}")) ToolError::InvalidArgument(format!("scope parent-owned Workdir tools: {error}"))
})?; })?;
let child_workdir_tool_broker = workdir_scope.broker();
let spawn_selector = let spawn_selector =
parse_spawn_profile_selector(input.profile.as_deref()).map_err(|msg| { parse_spawn_profile_selector(input.profile.as_deref()).map_err(|msg| {
@@ -490,7 +484,6 @@ impl Tool for SubWorkerSpawnTool {
) )
.await .await
.map_err(|error| ToolError::ExecutionFailed(format!("build Internal Worker: {error}")))?; .map_err(|error| ToolError::ExecutionFailed(format!("build Internal Worker: {error}")))?;
child.bind_workdir_session(Some(workdir_delegation.scoped_session.clone()));
child child
.add_scope_rules([ScopeRule { .add_scope_rules([ScopeRule {
target: child_bash_output_dir.clone(), target: child_bash_output_dir.clone(),
@@ -510,6 +503,7 @@ impl Tool for SubWorkerSpawnTool {
self.runtime_base.clone(), self.runtime_base.clone(),
child_registry.clone(), child_registry.clone(),
None, None,
Some(child_workdir_tool_broker.clone()),
) )
.await .await
.map_err(|error| { .map_err(|error| {
@@ -552,6 +546,7 @@ impl Tool for SubWorkerSpawnTool {
); );
parent_notifications.notify(message, true); parent_notifications.notify(message, true);
})), })),
Some(child_workdir_tool_broker.clone()),
) )
.await; .await;
let session = session_result.map_err(|error| { let session = session_result.map_err(|error| {
@@ -621,7 +616,7 @@ impl Tool for SubWorkerSpawnTool {
let record = crate::spawn::registry::InternalSpawnedWorkerRecord::new( let record = crate::spawn::registry::InternalSpawnedWorkerRecord::new(
input.name.clone(), input.name.clone(),
scope_allow, scope_allow,
workdir_delegation, workdir_scope,
#[cfg(test)] #[cfg(test)]
installed_tools, installed_tools,
session.clone(), session.clone(),
@@ -674,18 +669,18 @@ fn logical_workdir_path(value: &str, field: &str) -> Result<FsPath, ToolError> {
}) })
} }
fn parse_workdir_scope(rules: &[ScopeRuleInput]) -> Result<Vec<WorkdirDelegationRule>, ToolError> { fn parse_workdir_scope(rules: &[ScopeRuleInput]) -> Result<Vec<WorkdirToolScopeRule>, ToolError> {
if rules.is_empty() { if rules.is_empty() {
return Err(ToolError::InvalidArgument("scope must not be empty".into())); return Err(ToolError::InvalidArgument("scope must not be empty".into()));
} }
rules rules
.iter() .iter()
.map(|rule| { .map(|rule| {
Ok(WorkdirDelegationRule { Ok(WorkdirToolScopeRule {
target: logical_workdir_path(&rule.target, "scope.target")?, target: logical_workdir_path(&rule.target, "scope.target")?,
permission: match rule.permission { permission: match rule.permission {
PermissionInput::Read => WorkdirDelegationPermission::Read, PermissionInput::Read => WorkdirToolScopePermission::Read,
PermissionInput::Write => WorkdirDelegationPermission::Write, PermissionInput::Write => WorkdirToolScopePermission::Write,
}, },
recursive: rule.recursive, recursive: rule.recursive,
}) })
@@ -693,22 +688,24 @@ fn parse_workdir_scope(rules: &[ScopeRuleInput]) -> Result<Vec<WorkdirDelegation
.collect() .collect()
} }
fn workdir_delegation_request( fn workdir_tool_scope(
cwd: Option<&str>, cwd: Option<&str>,
rules: Vec<WorkdirDelegationRule>, rules: Vec<WorkdirToolScopeRule>,
) -> Result<WorkdirDelegationRequest, ToolError> { command: bool,
Ok(WorkdirDelegationRequest { ) -> Result<WorkdirToolScope, ToolError> {
Ok(WorkdirToolScope {
rules, rules,
cwd: logical_workdir_path(cwd.unwrap_or("."), "cwd")?, cwd: logical_workdir_path(cwd.unwrap_or("."), "cwd")?,
command,
}) })
} }
fn require_active_workdir_session( fn require_workdir_tool_broker(
session: Option<&WorkdirSessionHandle>, broker: Option<&WorkdirToolBroker>,
) -> Result<&WorkdirSessionHandle, ToolError> { ) -> Result<&WorkdirToolBroker, ToolError> {
session.ok_or_else(|| { broker.ok_or_else(|| {
ToolError::InvalidArgument( ToolError::InvalidArgument(
"SubWorkerSpawn requires an active Workdir session; attach a Workdir before delegating filesystem access" "SubWorkerSpawn requires parent-owned Workdir tools; attach a Workdir before granting filesystem access"
.to_string(), .to_string(),
) )
}) })
@@ -946,7 +943,7 @@ pub(crate) fn sub_worker_spawn_tool(
runtime_base: PathBuf, runtime_base: PathBuf,
bash_output_dir: PathBuf, bash_output_dir: PathBuf,
workspace_root: PathBuf, workspace_root: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>, workdir_tool_broker: Option<WorkdirToolBroker>,
registry: Arc<SpawnedWorkerRegistry>, registry: Arc<SpawnedWorkerRegistry>,
spawner_manifest: WorkerManifest, spawner_manifest: WorkerManifest,
prompts: Arc<ArcSwap<PromptCatalog>>, prompts: Arc<ArcSwap<PromptCatalog>>,
@@ -958,7 +955,7 @@ pub(crate) fn sub_worker_spawn_tool(
runtime_base, runtime_base,
bash_output_dir, bash_output_dir,
workspace_root, workspace_root,
source_workdir_session, workdir_tool_broker,
registry, registry,
spawner_manifest, spawner_manifest,
prompts, prompts,
@@ -972,7 +969,7 @@ fn sub_worker_spawn_tool_impl(
runtime_base: PathBuf, runtime_base: PathBuf,
bash_output_dir: PathBuf, bash_output_dir: PathBuf,
workspace_root: PathBuf, workspace_root: PathBuf,
source_workdir_session: Option<WorkdirSessionHandle>, workdir_tool_broker: Option<WorkdirToolBroker>,
registry: Arc<SpawnedWorkerRegistry>, registry: Arc<SpawnedWorkerRegistry>,
spawner_manifest: WorkerManifest, spawner_manifest: WorkerManifest,
prompts: Arc<ArcSwap<PromptCatalog>>, prompts: Arc<ArcSwap<PromptCatalog>>,
@@ -1004,7 +1001,7 @@ fn sub_worker_spawn_tool_impl(
runtime_base.clone(), runtime_base.clone(),
bash_output_dir.clone(), bash_output_dir.clone(),
workspace_root.clone(), workspace_root.clone(),
source_workdir_session.clone(), workdir_tool_broker.clone(),
registry.clone(), registry.clone(),
spawner_manifest.clone(), spawner_manifest.clone(),
prompts.load_full().source(), prompts.load_full().source(),
@@ -1037,12 +1034,12 @@ mod tests {
}; };
#[test] #[test]
fn missing_active_workdir_session_fails_deterministically() { fn missing_parent_workdir_tool_broker_fails_deterministically() {
let error = require_active_workdir_session(None).unwrap_err(); let error = require_workdir_tool_broker(None).unwrap_err();
assert!(matches!( assert!(matches!(
error, error,
ToolError::InvalidArgument(message) ToolError::InvalidArgument(message)
if message.contains("requires an active Workdir session") if message.contains("requires parent-owned Workdir tools")
)); ));
} }
@@ -1079,6 +1076,7 @@ mod tests {
let valid: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({ let valid: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
"name":"reviewer","task":"review","profile":"builtin:reviewer", "name":"reviewer","task":"review","profile":"builtin:reviewer",
"scope":[{"target":"work","permission":"write"}], "scope":[{"target":"work","permission":"write"}],
"command":true,
"review":{"ticket_id":"T1"} "review":{"ticket_id":"T1"}
})) }))
.unwrap(); .unwrap();
@@ -1173,7 +1171,7 @@ enabled = false
let fail_requests = Arc::new(AtomicBool::new(false)); let fail_requests = Arc::new(AtomicBool::new(false));
let prompt_loader = PromptCatalogSource::builtins_only(); let prompt_loader = PromptCatalogSource::builtins_only();
let (parent_method_tx, mut parent_method_rx) = mpsc::channel(8); let (parent_method_tx, mut parent_method_rx) = mpsc::channel(8);
let source_workdir_session = workdir::delegation_capable_session(Arc::new( let workdir_tool_broker = workdir::WorkdirToolBroker::new(Arc::new(
workdir::LocalWorkdirSession::materialized_bound( workdir::LocalWorkdirSession::materialized_bound(
workdir::Workdir::new("test-workdir"), workdir::Workdir::new("test-workdir"),
workspace_root.clone(), workspace_root.clone(),
@@ -1189,7 +1187,7 @@ enabled = false
runtime.path().to_path_buf(), runtime.path().to_path_buf(),
bash_output_dir.clone(), bash_output_dir.clone(),
workspace_root.clone(), workspace_root.clone(),
Some(source_workdir_session), Some(workdir_tool_broker),
registry.clone(), registry.clone(),
manifest.clone(), manifest.clone(),
prompt_loader, prompt_loader,
@@ -1212,7 +1210,8 @@ enabled = false
"target": ".", "target": ".",
"permission": "write", "permission": "write",
"recursive": true "recursive": true
}] }],
"command": true
}); });
assert!(spawner_scope.snapshot().is_writable(&workspace_root)); assert!(spawner_scope.snapshot().is_writable(&workspace_root));
@@ -1247,15 +1246,6 @@ enabled = false
let record = registry let record = registry
.get_internal("reviewer-child") .get_internal("reviewer-child")
.expect("Internal reviewer registry record"); .expect("Internal reviewer registry record");
let child_bash_output_dir = bash_output_dir.join("sub-workers").join("reviewer-child");
record
.workdir_delegation
.scoped_session
.stat(workdir::StatRequest {
path: WorkdirPath::new_scoped(child_bash_output_dir.to_string_lossy()).unwrap(),
})
.await
.expect("local child retains read scope for its Bash output directory");
for required in ["Read", "Write", "Edit", "Glob", "Grep", "Bash"] { for required in ["Read", "Write", "Edit", "Glob", "Grep", "Bash"] {
assert!( assert!(
record.installed_tools.iter().any(|name| name == required), record.installed_tools.iter().any(|name| name == required),
@@ -1371,7 +1361,7 @@ enabled = false
"Stopped terminal child must release its delegated Workdir session" "Stopped terminal child must release its delegated Workdir session"
); );
assert!( assert!(
!record.workdir_delegation.is_active(), !record.workdir_tool_scope.is_active(),
"stopped child must revoke cloned scoped sessions" "stopped child must revoke cloned scoped sessions"
); );
assert!(registry.get_internal("reviewer-child").is_some()); assert!(registry.get_internal("reviewer-child").is_some());
@@ -1426,7 +1416,7 @@ enabled = false
Arc::new(AvailableWorkspaceClient), Arc::new(AvailableWorkspaceClient),
); );
let remote_client = Arc::new(StrictRemoteWorkdirWorkspaceClient::default()); let remote_client = Arc::new(StrictRemoteWorkdirWorkspaceClient::default());
let source_workdir_session = workdir::delegation_capable_session( let workdir_tool_broker = workdir::WorkdirToolBroker::new(
WorkspaceAttachedWorkdirSession::handle(remote_client.clone()), WorkspaceAttachedWorkdirSession::handle(remote_client.clone()),
); );
let calls = Arc::new(AtomicUsize::new(0)); let calls = Arc::new(AtomicUsize::new(0));
@@ -1438,7 +1428,7 @@ enabled = false
runtime.path().to_path_buf(), runtime.path().to_path_buf(),
bash_output_dir.clone(), bash_output_dir.clone(),
workspace_root.clone(), workspace_root.clone(),
Some(source_workdir_session), Some(workdir_tool_broker),
registry.clone(), registry.clone(),
manifest, manifest,
PromptCatalogSource::builtins_only(), PromptCatalogSource::builtins_only(),
@@ -1478,51 +1468,12 @@ enabled = false
record.session.wait_until_idle().await, record.session.wait_until_idle().await,
crate::internal_worker::InternalWorkerSessionStatus::Idle crate::internal_worker::InternalWorkerSessionStatus::Idle
); );
assert!(record.installed_tools.iter().any(|tool| tool == "Write"));
assert!(!record.installed_tools.iter().any(|tool| tool == "Bash"));
assert_eq!(calls.load(Ordering::SeqCst), 1); assert_eq!(calls.load(Ordering::SeqCst), 1);
assert_eq!(
remote_client
.foreign_scope_rejections
.load(Ordering::SeqCst),
0
);
let child_bash_output_dir = bash_output_dir.join("sub-workers").join("remote-child");
assert!(child_bash_output_dir.is_dir());
for required in ["Read", "Write", "Edit", "Glob", "Grep", "Bash"] {
assert!( assert!(
record.installed_tools.iter().any(|name| name == required), remote_client.requests().is_empty(),
"remote write-scoped child is missing {required}: {:?}", "spawning a child must not open or delegate a provider Workdir session"
record.installed_tools
);
}
let remote_requests = remote_client.requests();
let operate_requests = remote_requests
.iter()
.filter(|request| request.body.is_some())
.collect::<Vec<_>>();
assert_eq!(
operate_requests.len(),
1,
"remote requests: {remote_requests:?}"
);
let operation_body: serde_json::Value = serde_json::from_str(
operate_requests[0]
.body
.as_deref()
.expect("remote operation body"),
)
.unwrap();
let rules = operation_body["delegations"][0]["rules"]
.as_array()
.expect("delegation rules");
assert_eq!(rules.len(), 1, "remote operation body: {operation_body}");
assert_eq!(rules[0]["target"], "");
assert!(
!operation_body.to_string().contains(
child_bash_output_dir
.to_str()
.expect("UTF-8 test output directory")
)
); );
} }
@@ -1534,6 +1485,7 @@ enabled = false
.and_then(serde_json::Value::as_object) .and_then(serde_json::Value::as_object)
.expect("schema properties"); .expect("schema properties");
assert!(properties.contains_key("cwd"), "schema: {schema}"); assert!(properties.contains_key("cwd"), "schema: {schema}");
assert!(properties.contains_key("command"), "schema: {schema}");
let required = schema let required = schema
.get("required") .get("required")
.and_then(serde_json::Value::as_array) .and_then(serde_json::Value::as_array)
@@ -1663,7 +1615,6 @@ enabled = false
#[derive(Debug, Default)] #[derive(Debug, Default)]
struct StrictRemoteWorkdirWorkspaceClient { struct StrictRemoteWorkdirWorkspaceClient {
requests: Mutex<Vec<WorkspaceRequest>>, requests: Mutex<Vec<WorkspaceRequest>>,
foreign_scope_rejections: AtomicUsize,
} }
impl StrictRemoteWorkdirWorkspaceClient { impl StrictRemoteWorkdirWorkspaceClient {
@@ -1695,59 +1646,10 @@ enabled = false
self.requests self.requests
.lock() .lock()
.expect("remote Workdir request lock") .expect("remote Workdir request lock")
.push(request.clone()); .push(request);
if request.path.ends_with("/fence") { Err(WorkspaceClientError::Request(
return Ok(WorkspaceResponse { "SubWorker spawn must not call the remote Workdir provider".into(),
status: 200, ))
body: serde_json::json!({ "value": "remote-fence-1" }).to_string(),
});
}
let body: serde_json::Value = serde_json::from_str(
request
.body
.as_deref()
.ok_or_else(|| WorkspaceClientError::Request("missing request body".into()))?,
)
.map_err(|error| WorkspaceClientError::Request(error.to_string()))?;
let has_foreign_scope = body
.get("delegations")
.and_then(serde_json::Value::as_array)
.into_iter()
.flatten()
.flat_map(|delegation| {
delegation
.get("rules")
.and_then(serde_json::Value::as_array)
.into_iter()
.flatten()
})
.filter_map(|rule| rule.get("target").and_then(serde_json::Value::as_str))
.any(|target| Path::new(target).is_absolute());
if has_foreign_scope {
self.foreign_scope_rejections.fetch_add(1, Ordering::SeqCst);
return Ok(WorkspaceResponse {
status: 403,
body: serde_json::json!({
"code": "out_of_scope",
"message": "Worker-host path is outside the remote Workdir namespace"
})
.to_string(),
});
}
Ok(WorkspaceResponse {
status: 200,
body: serde_json::json!({
"operation": "stat",
"result": {
"path": "",
"kind": "directory",
"size": 0
}
})
.to_string(),
})
} }
} }
+4
View File
@@ -332,6 +332,7 @@ async fn shutdown_closes_bound_workdir_session() {
command: "sleep 30".to_owned(), command: "sleep 30".to_owned(),
timeout_secs: 60, timeout_secs: 60,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: None, tool_call_id: None,
}) })
@@ -376,6 +377,7 @@ async fn controller_projects_workdir_command_events_and_snapshot_state() {
command: "printf ready; sleep 0.3; printf done".to_owned(), command: "printf ready; sleep 0.3; printf done".to_owned(),
timeout_secs: 5, timeout_secs: 5,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: Some("tool-command-1".into()), tool_call_id: Some("tool-command-1".into()),
}) })
@@ -484,6 +486,7 @@ async fn controller_refreshes_command_snapshot_after_high_output_provider_lag()
.to_owned(), .to_owned(),
timeout_secs: 10, timeout_secs: 10,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: Some("tool-high-output".into()), tool_call_id: Some("tool-high-output".into()),
}) })
@@ -560,6 +563,7 @@ async fn controller_startup_failure_closes_bound_workdir_session() {
command: "printf unreachable".to_owned(), command: "printf unreachable".to_owned(),
timeout_secs: 5, timeout_secs: 5,
output_limit: 1024, output_limit: 1024,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: None, tool_call_id: None,
}) })
+20 -156
View File
@@ -48,8 +48,7 @@ use workdir::http::{
}; };
use workdir::workspace::{ use workdir::workspace::{
MaterializerKind, WorkingDirectoryCleanupTarget, WorkingDirectoryOccupancy, MaterializerKind, WorkingDirectoryCleanupTarget, WorkingDirectoryOccupancy,
WorkingDirectoryStatusKind, WorkingDirectorySummary, WorkspaceWorkdirSessionFence, WorkingDirectoryStatusKind, WorkingDirectorySummary, WorkspaceWorkdirSessionOperationRequest,
WorkspaceWorkdirSessionOperationRequest,
}; };
use workdir::{CommandHandle, WorkdirSessionHandle}; use workdir::{CommandHandle, WorkdirSessionHandle};
use worker::feature::builtin::{WorkerObservationSubject, WorkerObservationSubjectRef}; use worker::feature::builtin::{WorkerObservationSubject, WorkerObservationSubjectRef};
@@ -355,7 +354,6 @@ static EMBEDDED_RUNTIME_REQUEST_IDENTITY: std::sync::LazyLock<
struct WorkdirCommandSession { struct WorkdirCommandSession {
source: WorkdirSessionHandle, source: WorkdirSessionHandle,
provider_handle: CommandHandle, provider_handle: CommandHandle,
delegations: Vec<workdir::WorkdirDelegationRequest>,
} }
enum RegisteredWorkdirSession { enum RegisteredWorkdirSession {
@@ -398,7 +396,6 @@ impl WorkdirSessionRegistry {
worker: RuntimeWorkerRef, worker: RuntimeWorkerRef,
source: WorkdirSessionHandle, source: WorkdirSessionHandle,
provider_handle: CommandHandle, provider_handle: CommandHandle,
delegations: Vec<workdir::WorkdirDelegationRequest>,
) -> CommandHandle { ) -> CommandHandle {
let external_handle = loop { let external_handle = loop {
let candidate = CommandHandle(Uuid::now_v7().to_string()); let candidate = CommandHandle(Uuid::now_v7().to_string());
@@ -414,7 +411,6 @@ impl WorkdirSessionRegistry {
WorkdirCommandSession { WorkdirCommandSession {
source, source,
provider_handle, provider_handle,
delegations,
}, },
); );
external_handle external_handle
@@ -2586,10 +2582,6 @@ fn build_inner_router(api: WorkspaceApi) -> Router {
post(scoped_attach_current_worker_workdir) post(scoped_attach_current_worker_workdir)
.delete(scoped_detach_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( .route(
"/api/w/{workspace_id}/workers/self/workdir-session/operations", "/api/w/{workspace_id}/workers/self/workdir-session/operations",
post(scoped_execute_current_worker_workdir_operation), post(scoped_execute_current_worker_workdir_operation),
@@ -7341,46 +7333,11 @@ async fn scoped_detach_current_worker_workdir(
})) }))
} }
async fn scoped_current_worker_workdir_session_fence(
State(api): State<WorkspaceApi>,
AxumPath(path): AxumPath<ScopedWorkspacePath>,
headers: HeaderMap,
) -> ApiResult<Json<WorkspaceWorkdirSessionFence>> {
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(())
}
}
fn validated_current_worker_attachment( fn validated_current_worker_attachment(
api: &WorkspaceApi, api: &WorkspaceApi,
worker: &RuntimeWorkerRef, worker: &RuntimeWorkerRef,
expected_session_fence: Option<&str>,
) -> ApiResult<WorkerWorkdirLinkRecord> { ) -> ApiResult<WorkerWorkdirLinkRecord> {
let link = current_worker_active_attachment(api, worker)?; current_worker_active_attachment(api, worker)
validate_current_worker_workdir_session_fence(&link, expected_session_fence)?;
Ok(link)
} }
#[derive(Debug)] #[derive(Debug)]
@@ -7435,23 +7392,13 @@ async fn scoped_execute_current_worker_workdir_operation(
) -> std::result::Result<Json<WorkdirSessionOperationResult>, WorkdirOperationApiError> { ) -> std::result::Result<Json<WorkdirSessionOperationResult>, WorkdirOperationApiError> {
validate_workspace_scope(&api, &path.workspace_id)?; validate_workspace_scope(&api, &path.workspace_id)?;
let worker = current_worker_identity(&api, &path.workspace_id, &headers)?; let worker = current_worker_identity(&api, &path.workspace_id, &headers)?;
let expected_session_fence = request.expected_session_fence;
let delegations = request.delegations;
let result = match request.operation { let result = match request.operation {
WorkdirSessionOperation::CommandStart(command) => { WorkdirSessionOperation::CommandStart(command) => {
let session_lock = current_worker_session_lock(&api, &worker); let session_lock = current_worker_session_lock(&api, &worker);
let _session_guard = session_lock.lock().await; let _session_guard = session_lock.lock().await;
let link = validated_current_worker_attachment( let link = validated_current_worker_attachment(&api, &worker)?;
&api,
&worker,
expected_session_fence.as_deref(),
)?;
let source = open_current_worker_workdir_session_locked(&api, &worker, &link).await?; let source = open_current_worker_workdir_session_locked(&api, &worker, &link).await?;
let applied = let provider_handle = source
apply_current_worker_delegations(&worker, source.clone(), delegations.clone())
.await?;
let provider_handle = applied
.scoped_session
.start_command(command) .start_command(command)
.await .await
.map_err(|error| current_worker_workdir_operation_error(&worker, error))?; .map_err(|error| current_worker_workdir_operation_error(&worker, error))?;
@@ -7470,58 +7417,32 @@ async fn scoped_execute_current_worker_workdir_operation(
.workdir_sessions .workdir_sessions
.lock() .lock()
.expect("Workdir session registry lock poisoned") .expect("Workdir session registry lock poisoned")
.register_command( .register_command(worker.clone(), registered_source, provider_handle);
worker.clone(),
registered_source,
provider_handle,
delegations,
);
WorkdirSessionOperationResult::CommandStart(external_handle) WorkdirSessionOperationResult::CommandStart(external_handle)
} }
WorkdirSessionOperation::CommandStatus(external_handle) => { WorkdirSessionOperation::CommandStatus(external_handle) => {
let (session, provider_handle) = current_worker_command_session( let (session, provider_handle) =
&api, current_worker_command_session(&api, &worker, &external_handle)?;
&worker,
&external_handle,
&delegations,
expected_session_fence.as_deref(),
)
.await?;
session session
.scoped_session
.command_status(provider_handle) .command_status(provider_handle)
.await .await
.map(WorkdirSessionOperationResult::CommandStatus) .map(WorkdirSessionOperationResult::CommandStatus)
.map_err(|error| current_worker_workdir_operation_error(&worker, error))? .map_err(|error| current_worker_workdir_operation_error(&worker, error))?
} }
WorkdirSessionOperation::CommandOutput(mut output) => { WorkdirSessionOperation::CommandOutput(mut output) => {
let (session, provider_handle) = current_worker_command_session( let (session, provider_handle) =
&api, current_worker_command_session(&api, &worker, &output.handle)?;
&worker,
&output.handle,
&delegations,
expected_session_fence.as_deref(),
)
.await?;
output.handle = provider_handle; output.handle = provider_handle;
session session
.scoped_session
.command_output(output) .command_output(output)
.await .await
.map(WorkdirSessionOperationResult::CommandOutput) .map(WorkdirSessionOperationResult::CommandOutput)
.map_err(|error| current_worker_workdir_operation_error(&worker, error))? .map_err(|error| current_worker_workdir_operation_error(&worker, error))?
} }
WorkdirSessionOperation::CommandCancel(external_handle) => { WorkdirSessionOperation::CommandCancel(external_handle) => {
let (session, provider_handle) = current_worker_command_session( let (session, provider_handle) =
&api, current_worker_command_session(&api, &worker, &external_handle)?;
&worker,
&external_handle,
&delegations,
expected_session_fence.as_deref(),
)
.await?;
session session
.scoped_session
.cancel_command(provider_handle) .cancel_command(provider_handle)
.await .await
.map(|()| WorkdirSessionOperationResult::CommandCancel) .map(|()| WorkdirSessionOperationResult::CommandCancel)
@@ -7536,14 +7457,9 @@ async fn scoped_execute_current_worker_workdir_operation(
| WorkdirSessionOperation::Grep(_)) => { | WorkdirSessionOperation::Grep(_)) => {
let session_lock = current_worker_session_lock(&api, &worker); let session_lock = current_worker_session_lock(&api, &worker);
let _session_guard = session_lock.lock().await; let _session_guard = session_lock.lock().await;
let link = validated_current_worker_attachment( let link = validated_current_worker_attachment(&api, &worker)?;
&api,
&worker,
expected_session_fence.as_deref(),
)?;
let source = open_current_worker_workdir_session_locked(&api, &worker, &link).await?; let source = open_current_worker_workdir_session_locked(&api, &worker, &link).await?;
let applied = apply_current_worker_delegations(&worker, source, delegations).await?; execute_workdir_session_operation(&source, operation)
execute_workdir_session_operation(&applied.scoped_session, operation)
.await .await
.map_err(|error| current_worker_workdir_operation_error(&worker, error))? .map_err(|error| current_worker_workdir_operation_error(&worker, error))?
} }
@@ -7551,29 +7467,12 @@ async fn scoped_execute_current_worker_workdir_operation(
Ok(Json(result)) Ok(Json(result))
} }
async fn apply_current_worker_delegations( fn current_worker_command_session(
worker: &RuntimeWorkerRef,
source: WorkdirSessionHandle,
delegations: Vec<workdir::WorkdirDelegationRequest>,
) -> Result<workdir::AppliedWorkdirDelegation> {
workdir::apply_delegation_chain(source, delegations)
.await
.map_err(|error| Error::RuntimeOperationFailed {
runtime_id: worker.runtime_id.clone(),
code: "workdir_session_delegation_failed".to_string(),
message: error.to_string(),
})
}
async fn current_worker_command_session(
api: &WorkspaceApi, api: &WorkspaceApi,
worker: &RuntimeWorkerRef, worker: &RuntimeWorkerRef,
external_handle: &CommandHandle, external_handle: &CommandHandle,
delegations: &[workdir::WorkdirDelegationRequest], ) -> std::result::Result<(WorkdirSessionHandle, CommandHandle), WorkdirOperationApiError> {
expected_session_fence: Option<&str>, let _link = validated_current_worker_attachment(api, worker)?;
) -> std::result::Result<(workdir::AppliedWorkdirDelegation, CommandHandle), WorkdirOperationApiError>
{
let _link = validated_current_worker_attachment(api, worker, expected_session_fence)?;
let command = api let command = api
.workdir_sessions .workdir_sessions
.lock() .lock()
@@ -7585,15 +7484,7 @@ async fn current_worker_command_session(
workdir::WorkdirError::UnknownCommand(external_handle.0.clone()), workdir::WorkdirError::UnknownCommand(external_handle.0.clone()),
)) ))
})?; })?;
if command.delegations != delegations { Ok((command.source, command.provider_handle))
return Err(Error::WorkdirAttachmentConflict(
"command lifecycle delegation differs from CommandStart".to_string(),
)
.into());
}
let session =
apply_current_worker_delegations(worker, command.source, command.delegations).await?;
Ok((session, command.provider_handle))
} }
fn current_worker_workdir_operation_error( fn current_worker_workdir_operation_error(
@@ -16775,6 +16666,7 @@ mod tests {
command: "printf ready; sleep 30".to_string(), command: "printf ready; sleep 30".to_string(),
timeout_secs: 60, timeout_secs: 60,
output_limit: 4096, output_limit: 4096,
cwd: None,
spill_dir: None, spill_dir: None,
tool_call_id: Some("tool-call-command-session".to_string()), tool_call_id: Some("tool-call-command-session".to_string()),
}) })
@@ -16784,12 +16676,8 @@ mod tests {
let mut registry = WorkdirSessionRegistry::default(); let mut registry = WorkdirSessionRegistry::default();
registry.insert_attachment(worker.clone(), source.clone()); registry.insert_attachment(worker.clone(), source.clone());
let registered_source = registry.remove_attachment(&worker).unwrap(); let registered_source = registry.remove_attachment(&worker).unwrap();
let external_handle = registry.register_command( let external_handle =
worker.clone(), registry.register_command(worker.clone(), registered_source, provider_handle.clone());
registered_source,
provider_handle.clone(),
Vec::new(),
);
assert_ne!(external_handle, provider_handle); assert_ne!(external_handle, provider_handle);
let refreshed: WorkdirSessionHandle = Arc::new(workdir::LocalWorkdirSession::new( let refreshed: WorkdirSessionHandle = Arc::new(workdir::LocalWorkdirSession::new(
@@ -23503,30 +23391,6 @@ mod tests {
assert_eq!(response.status(), StatusCode::BAD_REQUEST); 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] #[tokio::test]
async fn backend_workdir_session_proxy_executes_typed_operations() { async fn backend_workdir_session_proxy_executes_typed_operations() {
use manifest::Scope; use manifest::Scope;
+1 -1
View File
@@ -15,7 +15,7 @@
}; };
review = { review = {
instructions = "Use the current Ticket Merge Request as review authority. Call `ShowMergeRequest` and confirm its source selector resolves to exact committed implementation HEAD, then spawn one actual direct-child SubWorker with profile builtin:reviewer, write scope for Workdir inspection and command validation, and only the Ticket id in the structured review handoff. The trusted spawn layer records `ReviewRequested` with the exact source ref and injects review capability; do not place commit/ref identity, capability material, or a prewritten verdict in model input. The child must commit `ReviewMergeRequest`; prose output and Worker observation are not approval authority. After the structured result for the exact current source ref exists, request a Flow transition."; instructions = "Use the current Ticket Merge Request as review authority. Call `ShowMergeRequest` and confirm its source selector resolves to exact committed implementation HEAD, then spawn one actual direct-child SubWorker with profile builtin:reviewer, write scope plus an explicit command grant for Workdir inspection and command validation, and only the Ticket id in the structured review handoff. The trusted spawn layer records `ReviewRequested` with the exact source ref and injects review capability; do not place commit/ref identity, capability material, or a prewritten verdict in model input. The child must commit `ReviewMergeRequest`; prose output and Worker observation are not approval authority. After the structured result for the exact current source ref exists, request a Flow transition.";
transitions = { transitions = {
approved = { approved = {
target = "complete"; target = "complete";
@@ -1,8 +1,8 @@
Spawn a parent-owned Internal SubWorker session to split context for a delegated task. The parent Worker's write scope is reduced by the scope passed here; the Internal SubWorker starts running `task` immediately without creating a Runtime Worker record, OS process, PID, or Unix socket. It remains available for follow-up turns until explicitly stopped or its parent exits. Spawn a parent-owned Internal SubWorker session to split context for a delegated task. The parent Worker's write scope is reduced by the scope passed here; the Internal SubWorker starts running `task` immediately without creating a Runtime Worker record, OS process, PID, or Unix socket. It remains available for follow-up turns until explicitly stopped or its parent exits.
Optional `cwd`: when provided, the spawned SubWorker's tool default working directory only. It must be an absolute existing directory covered by the child's delegated readable scope, and it does not change workspace/Profile/memory/Ticket roots or grant authority. `name` must be unique among this Worker's direct children. Optional `cwd`: when provided, the spawned SubWorker's tool default working directory only. It must be a Workdir-relative existing directory covered by the child's readable scope, and it does not change workspace/Profile/memory/Ticket roots or grant authority. `name` must be unique among this Worker's direct children.
Profile selection: `profile` may be omitted or set to `default` to use the effective child default profile, set to `inherit` to derive reusable child configuration from this Worker, or set to one of the registry selectors below. Raw/path profile selectors are not accepted by SubWorkerSpawn. `scope` is always the only delegated filesystem capability; profile scope is replaced by the explicit SubWorkerSpawn scope. Profile selection: `profile` may be omitted or set to `default` to use the effective child default profile, set to `inherit` to derive reusable child configuration from this Worker, or set to one of the registry selectors below. Raw/path profile selectors are not accepted by SubWorkerSpawn. `scope` is the child's only filesystem capability and replaces profile scope. `command` is a separate explicit grant, defaults to false, and is accepted only with a writable scope; writable scope alone does not grant command execution.
Default profile: {{ default_profile }} Default profile: {{ default_profile }}
Special selector: inherit — derive reusable model/worker/tool policy from the spawner while replacing worker.name and scope. Special selector: inherit — derive reusable model/worker/tool policy from the spawner while replacing worker.name and scope.