Compare commits
2
Commits
89ee5e48a5
...
21bd089a23
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
21bd089a23 | ||
|
|
88f463e633 |
Generated
+1
@@ -6067,6 +6067,7 @@ dependencies = [
|
|||||||
"config-source",
|
"config-source",
|
||||||
"dotenv",
|
"dotenv",
|
||||||
"flow",
|
"flow",
|
||||||
|
"fs-operation",
|
||||||
"fs4",
|
"fs4",
|
||||||
"futures",
|
"futures",
|
||||||
"futures-util",
|
"futures-util",
|
||||||
|
|||||||
@@ -0,0 +1,781 @@
|
|||||||
|
use std::collections::HashMap;
|
||||||
|
use std::path::Path;
|
||||||
|
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
|
||||||
|
use std::sync::{Arc, Mutex, Weak};
|
||||||
|
|
||||||
|
use async_trait::async_trait;
|
||||||
|
use fs_operation::{
|
||||||
|
EditRequest, EditResult, FsPath, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest,
|
||||||
|
ListResult, ReadRequest, ReadResult, StatRequest, StatResult, WriteRequest, WriteResult,
|
||||||
|
};
|
||||||
|
|
||||||
|
use crate::{
|
||||||
|
CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, Workdir,
|
||||||
|
WorkdirError, WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability,
|
||||||
|
WorkdirSessionHandle,
|
||||||
|
};
|
||||||
|
|
||||||
|
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||||
|
pub enum WorkdirDelegationPermission {
|
||||||
|
Read,
|
||||||
|
Write,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||||
|
pub struct WorkdirDelegationRule {
|
||||||
|
pub target: FsPath,
|
||||||
|
pub permission: WorkdirDelegationPermission,
|
||||||
|
pub recursive: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, Eq, PartialEq)]
|
||||||
|
pub struct WorkdirDelegationRequest {
|
||||||
|
pub rules: Vec<WorkdirDelegationRule>,
|
||||||
|
pub cwd: FsPath,
|
||||||
|
}
|
||||||
|
|
||||||
|
pub struct WorkdirDelegation {
|
||||||
|
pub scoped_session: WorkdirSessionHandle,
|
||||||
|
pub capabilities: WorkdirSessionCapabilities,
|
||||||
|
validity: Arc<SessionValidity>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::fmt::Debug for WorkdirDelegation {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
f.debug_struct("WorkdirDelegation")
|
||||||
|
.field("workdir", &self.scoped_session.workdir())
|
||||||
|
.field("capabilities", &self.capabilities)
|
||||||
|
.field("active", &self.is_active())
|
||||||
|
.finish()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl WorkdirDelegation {
|
||||||
|
pub fn is_active(&self) -> bool {
|
||||||
|
self.validity.is_active()
|
||||||
|
}
|
||||||
|
|
||||||
|
pub fn release(&self) {
|
||||||
|
self.validity.active.store(false, Ordering::Release);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
impl Drop for WorkdirDelegation {
|
||||||
|
fn drop(&mut self) {
|
||||||
|
self.release();
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug)]
|
||||||
|
struct SessionValidity {
|
||||||
|
active: AtomicBool,
|
||||||
|
parent: Option<Arc<SessionValidity>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl SessionValidity {
|
||||||
|
fn root() -> Arc<Self> {
|
||||||
|
Arc::new(Self {
|
||||||
|
active: AtomicBool::new(true),
|
||||||
|
parent: None,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn child(parent: Arc<Self>) -> Arc<Self> {
|
||||||
|
Arc::new(Self {
|
||||||
|
active: AtomicBool::new(true),
|
||||||
|
parent: Some(parent),
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_active(&self) -> bool {
|
||||||
|
self.active.load(Ordering::Acquire)
|
||||||
|
&& self.parent.as_ref().is_none_or(|parent| parent.is_active())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug)]
|
||||||
|
struct ActiveWriteLease {
|
||||||
|
validity: Weak<SessionValidity>,
|
||||||
|
rules: Vec<WorkdirDelegationRule>,
|
||||||
|
}
|
||||||
|
|
||||||
|
struct DelegatingWorkdirSession {
|
||||||
|
source: WorkdirSessionHandle,
|
||||||
|
scope: Option<Vec<WorkdirDelegationRule>>,
|
||||||
|
capabilities: WorkdirSessionCapabilities,
|
||||||
|
validity: Arc<SessionValidity>,
|
||||||
|
child_write_leases: Mutex<HashMap<u64, ActiveWriteLease>>,
|
||||||
|
next_lease_id: AtomicU64,
|
||||||
|
closes_source: bool,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl std::fmt::Debug for DelegatingWorkdirSession {
|
||||||
|
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||||
|
f.debug_struct("DelegatingWorkdirSession")
|
||||||
|
.field("workdir", &self.source.workdir())
|
||||||
|
.field("scope", &self.scope)
|
||||||
|
.field("capabilities", &self.capabilities)
|
||||||
|
.field("active", &self.validity.is_active())
|
||||||
|
.finish_non_exhaustive()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Wrap a provider session with logical-path delegation and parent write gates.
|
||||||
|
pub fn delegation_capable_session(source: WorkdirSessionHandle) -> WorkdirSessionHandle {
|
||||||
|
let capabilities = source.capabilities();
|
||||||
|
Arc::new(DelegatingWorkdirSession {
|
||||||
|
source,
|
||||||
|
scope: None,
|
||||||
|
capabilities,
|
||||||
|
validity: SessionValidity::root(),
|
||||||
|
child_write_leases: Mutex::new(HashMap::new()),
|
||||||
|
next_lease_id: AtomicU64::new(1),
|
||||||
|
closes_source: true,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
impl DelegatingWorkdirSession {
|
||||||
|
fn ensure_active(&self) -> Result<(), WorkdirError> {
|
||||||
|
if self.validity.is_active() {
|
||||||
|
Ok(())
|
||||||
|
} else {
|
||||||
|
Err(WorkdirError::SessionClosed)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ensure_capability(
|
||||||
|
&self,
|
||||||
|
required: WorkdirSessionCapability,
|
||||||
|
operation: &'static str,
|
||||||
|
) -> Result<(), WorkdirError> {
|
||||||
|
self.ensure_active()?;
|
||||||
|
if self.capabilities.supports(required) {
|
||||||
|
Ok(())
|
||||||
|
} else {
|
||||||
|
Err(WorkdirError::Denied(format!(
|
||||||
|
"delegated workdir session does not permit {operation}"
|
||||||
|
)))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ensure_path(
|
||||||
|
&self,
|
||||||
|
path: &FsPath,
|
||||||
|
permission: WorkdirDelegationPermission,
|
||||||
|
) -> Result<(), WorkdirError> {
|
||||||
|
self.ensure_active()?;
|
||||||
|
if let Some(scope) = &self.scope {
|
||||||
|
if !scope
|
||||||
|
.iter()
|
||||||
|
.any(|rule| rule_allows_path(rule, path, permission))
|
||||||
|
{
|
||||||
|
return Err(WorkdirError::Denied(format!(
|
||||||
|
"logical workdir path `{path}` is outside the delegated {permission:?} scope"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if permission == WorkdirDelegationPermission::Write {
|
||||||
|
self.ensure_parent_write_available(path)?;
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ensure_read(
|
||||||
|
&self,
|
||||||
|
path: &FsPath,
|
||||||
|
capability: WorkdirSessionCapability,
|
||||||
|
) -> Result<(), WorkdirError> {
|
||||||
|
self.ensure_capability(capability, "read operations")?;
|
||||||
|
self.ensure_path(path, WorkdirDelegationPermission::Read)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ensure_write(
|
||||||
|
&self,
|
||||||
|
path: &FsPath,
|
||||||
|
capability: WorkdirSessionCapability,
|
||||||
|
) -> Result<(), WorkdirError> {
|
||||||
|
self.ensure_capability(capability, "write operations")?;
|
||||||
|
self.ensure_path(path, WorkdirDelegationPermission::Write)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ensure_command(&self, starting: bool) -> Result<(), WorkdirError> {
|
||||||
|
self.ensure_capability(WorkdirSessionCapability::Command, "command execution")?;
|
||||||
|
if starting && self.has_active_write_lease() {
|
||||||
|
return Err(WorkdirError::Denied(
|
||||||
|
"command execution is denied while a child holds a write delegation".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
|
||||||
|
fn ensure_parent_write_available(&self, path: &FsPath) -> Result<(), WorkdirError> {
|
||||||
|
let mut leases = self
|
||||||
|
.child_write_leases
|
||||||
|
.lock()
|
||||||
|
.expect("workdir delegation lease mutex poisoned");
|
||||||
|
leases.retain(|_, lease| lease.validity.upgrade().is_some_and(|v| v.is_active()));
|
||||||
|
if leases.values().any(|lease| {
|
||||||
|
lease.rules.iter().any(|rule| {
|
||||||
|
rule.permission == WorkdirDelegationPermission::Write
|
||||||
|
&& rule_allows_path(rule, path, WorkdirDelegationPermission::Write)
|
||||||
|
})
|
||||||
|
}) {
|
||||||
|
Err(WorkdirError::Denied(format!(
|
||||||
|
"logical workdir path `{path}` is leased to a child session"
|
||||||
|
)))
|
||||||
|
} else {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn has_active_write_lease(&self) -> bool {
|
||||||
|
let mut leases = self
|
||||||
|
.child_write_leases
|
||||||
|
.lock()
|
||||||
|
.expect("workdir delegation lease mutex poisoned");
|
||||||
|
leases.retain(|_, lease| lease.validity.upgrade().is_some_and(|v| v.is_active()));
|
||||||
|
leases.values().any(|lease| {
|
||||||
|
lease
|
||||||
|
.rules
|
||||||
|
.iter()
|
||||||
|
.any(|rule| rule.permission == WorkdirDelegationPermission::Write)
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
fn validate_delegation_rules(
|
||||||
|
&self,
|
||||||
|
rules: &[WorkdirDelegationRule],
|
||||||
|
) -> Result<WorkdirSessionCapabilities, WorkdirError> {
|
||||||
|
self.ensure_active()?;
|
||||||
|
if rules.is_empty() {
|
||||||
|
return Err(WorkdirError::Denied(
|
||||||
|
"workdir delegation requires at least one logical scope rule".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
let writable = rules
|
||||||
|
.iter()
|
||||||
|
.any(|rule| rule.permission == WorkdirDelegationPermission::Write);
|
||||||
|
if !self.capabilities.supports(WorkdirSessionCapability::Read)
|
||||||
|
|| (writable
|
||||||
|
&& (!self.capabilities.supports(WorkdirSessionCapability::Write)
|
||||||
|
|| !self.capabilities.supports(WorkdirSessionCapability::Edit)))
|
||||||
|
{
|
||||||
|
return Err(WorkdirError::Denied(
|
||||||
|
"parent workdir session cannot delegate the requested capabilities".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
for requested in rules {
|
||||||
|
if let Some(scope) = &self.scope {
|
||||||
|
if !scope
|
||||||
|
.iter()
|
||||||
|
.any(|parent| rule_contains_rule(parent, requested))
|
||||||
|
{
|
||||||
|
return Err(WorkdirError::Denied(format!(
|
||||||
|
"logical workdir scope `{}` exceeds the parent delegation",
|
||||||
|
requested.target
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
let mut delegated = vec![WorkdirSessionCapability::Read];
|
||||||
|
for capability in [
|
||||||
|
WorkdirSessionCapability::Glob,
|
||||||
|
WorkdirSessionCapability::Grep,
|
||||||
|
] {
|
||||||
|
if self.capabilities.supports(capability) {
|
||||||
|
delegated.push(capability);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if writable {
|
||||||
|
delegated.push(WorkdirSessionCapability::Write);
|
||||||
|
delegated.push(WorkdirSessionCapability::Edit);
|
||||||
|
}
|
||||||
|
Ok(WorkdirSessionCapabilities::from_capabilities(delegated))
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl WorkdirSession for DelegatingWorkdirSession {
|
||||||
|
fn workdir(&self) -> &Workdir {
|
||||||
|
self.source.workdir()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn capabilities(&self) -> WorkdirSessionCapabilities {
|
||||||
|
self.capabilities
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_delegation_capable(&self) -> bool {
|
||||||
|
true
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn capture_delegation_source(&self) -> Result<WorkdirSessionHandle, WorkdirError> {
|
||||||
|
self.ensure_active()?;
|
||||||
|
if self.scope.is_some() {
|
||||||
|
return Err(WorkdirError::Denied(
|
||||||
|
"scoped Workdir sessions cannot expose their provider source".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
self.source.capture_delegation_source().await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn delegate(
|
||||||
|
&self,
|
||||||
|
request: WorkdirDelegationRequest,
|
||||||
|
) -> Result<WorkdirDelegation, WorkdirError> {
|
||||||
|
let capabilities = self.validate_delegation_rules(&request.rules)?;
|
||||||
|
if !request
|
||||||
|
.rules
|
||||||
|
.iter()
|
||||||
|
.any(|rule| rule_allows_path(rule, &request.cwd, WorkdirDelegationPermission::Read))
|
||||||
|
{
|
||||||
|
return Err(WorkdirError::Denied(format!(
|
||||||
|
"delegated cwd `{}` is outside the delegated readable scope",
|
||||||
|
request.cwd
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
let source = self.source.capture_delegation_source().await?;
|
||||||
|
let validity = SessionValidity::child(self.validity.clone());
|
||||||
|
let id = self.next_lease_id.fetch_add(1, Ordering::Relaxed);
|
||||||
|
if request
|
||||||
|
.rules
|
||||||
|
.iter()
|
||||||
|
.any(|rule| rule.permission == WorkdirDelegationPermission::Write)
|
||||||
|
{
|
||||||
|
self.child_write_leases
|
||||||
|
.lock()
|
||||||
|
.expect("workdir delegation lease mutex poisoned")
|
||||||
|
.insert(
|
||||||
|
id,
|
||||||
|
ActiveWriteLease {
|
||||||
|
validity: Arc::downgrade(&validity),
|
||||||
|
rules: request.rules.clone(),
|
||||||
|
},
|
||||||
|
);
|
||||||
|
}
|
||||||
|
let child: WorkdirSessionHandle = Arc::new(DelegatingWorkdirSession {
|
||||||
|
source,
|
||||||
|
scope: Some(request.rules),
|
||||||
|
capabilities,
|
||||||
|
validity: validity.clone(),
|
||||||
|
child_write_leases: Mutex::new(HashMap::new()),
|
||||||
|
next_lease_id: AtomicU64::new(1),
|
||||||
|
closes_source: false,
|
||||||
|
});
|
||||||
|
let scoped_session: WorkdirSessionHandle =
|
||||||
|
if capabilities == WorkdirSessionCapabilities::READ_ONLY {
|
||||||
|
Arc::new(ReadOnlyWorkdirSession::new(child))
|
||||||
|
} else {
|
||||||
|
child
|
||||||
|
};
|
||||||
|
Ok(WorkdirDelegation {
|
||||||
|
scoped_session,
|
||||||
|
capabilities,
|
||||||
|
validity,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
|
||||||
|
self.ensure_read(&request.path, WorkdirSessionCapability::Read)?;
|
||||||
|
self.source.stat(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
|
||||||
|
self.ensure_read(&request.path, WorkdirSessionCapability::Read)?;
|
||||||
|
self.source.read(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn write(&self, request: WriteRequest) -> Result<WriteResult, WorkdirError> {
|
||||||
|
self.ensure_write(&request.path, WorkdirSessionCapability::Write)?;
|
||||||
|
self.source.write(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn edit(&self, request: EditRequest) -> Result<EditResult, WorkdirError> {
|
||||||
|
self.ensure_write(&request.path, WorkdirSessionCapability::Edit)?;
|
||||||
|
self.source.edit(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
|
||||||
|
self.ensure_read(&request.path, WorkdirSessionCapability::Read)?;
|
||||||
|
self.source.list(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
|
||||||
|
self.ensure_read(&request.path, WorkdirSessionCapability::Glob)?;
|
||||||
|
self.source.glob(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
|
||||||
|
self.ensure_read(&request.path, WorkdirSessionCapability::Grep)?;
|
||||||
|
self.source.grep(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn start_command(&self, request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
|
||||||
|
self.ensure_command(true)?;
|
||||||
|
self.source.start_command(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn command_status(&self, handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
|
||||||
|
self.ensure_command(false)?;
|
||||||
|
self.source.command_status(handle).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn command_output(
|
||||||
|
&self,
|
||||||
|
request: CommandOutputRequest,
|
||||||
|
) -> Result<CommandOutput, WorkdirError> {
|
||||||
|
self.ensure_command(false)?;
|
||||||
|
self.source.command_output(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn cancel_command(&self, handle: CommandHandle) -> Result<(), WorkdirError> {
|
||||||
|
self.ensure_command(false)?;
|
||||||
|
self.source.cancel_command(handle).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn close(&self) -> Result<(), WorkdirError> {
|
||||||
|
self.validity.active.store(false, Ordering::Release);
|
||||||
|
if self.closes_source {
|
||||||
|
self.source.close().await
|
||||||
|
} else {
|
||||||
|
Ok(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// A fail-closed read-only view over an already scoped delegated session.
|
||||||
|
#[derive(Debug)]
|
||||||
|
pub struct ReadOnlyWorkdirSession {
|
||||||
|
inner: WorkdirSessionHandle,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ReadOnlyWorkdirSession {
|
||||||
|
pub fn new(inner: WorkdirSessionHandle) -> Self {
|
||||||
|
Self { inner }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[async_trait]
|
||||||
|
impl WorkdirSession for ReadOnlyWorkdirSession {
|
||||||
|
fn workdir(&self) -> &Workdir {
|
||||||
|
self.inner.workdir()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn capabilities(&self) -> WorkdirSessionCapabilities {
|
||||||
|
WorkdirSessionCapabilities::READ_ONLY
|
||||||
|
}
|
||||||
|
|
||||||
|
fn is_delegation_capable(&self) -> bool {
|
||||||
|
true
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn delegate(
|
||||||
|
&self,
|
||||||
|
request: WorkdirDelegationRequest,
|
||||||
|
) -> Result<WorkdirDelegation, WorkdirError> {
|
||||||
|
if request
|
||||||
|
.rules
|
||||||
|
.iter()
|
||||||
|
.any(|rule| rule.permission == WorkdirDelegationPermission::Write)
|
||||||
|
{
|
||||||
|
return Err(WorkdirError::Denied(
|
||||||
|
"read-only workdir session cannot delegate write access".into(),
|
||||||
|
));
|
||||||
|
}
|
||||||
|
self.inner.delegate(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
|
||||||
|
self.inner.stat(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
|
||||||
|
self.inner.read(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn write(&self, _request: WriteRequest) -> Result<WriteResult, WorkdirError> {
|
||||||
|
Err(WorkdirError::Denied("read-only workdir session".into()))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn edit(&self, _request: EditRequest) -> Result<EditResult, WorkdirError> {
|
||||||
|
Err(WorkdirError::Denied("read-only workdir session".into()))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
|
||||||
|
self.inner.list(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
|
||||||
|
self.inner.glob(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
|
||||||
|
self.inner.grep(request).await
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn start_command(&self, _request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
|
||||||
|
Err(WorkdirError::Denied("read-only workdir session".into()))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn command_status(&self, _handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
|
||||||
|
Err(WorkdirError::Denied("read-only workdir session".into()))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn command_output(
|
||||||
|
&self,
|
||||||
|
_request: CommandOutputRequest,
|
||||||
|
) -> Result<CommandOutput, WorkdirError> {
|
||||||
|
Err(WorkdirError::Denied("read-only workdir session".into()))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn cancel_command(&self, _handle: CommandHandle) -> Result<(), WorkdirError> {
|
||||||
|
Err(WorkdirError::Denied("read-only workdir session".into()))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn close(&self) -> Result<(), WorkdirError> {
|
||||||
|
self.inner.close().await
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn rule_allows_path(
|
||||||
|
rule: &WorkdirDelegationRule,
|
||||||
|
path: &FsPath,
|
||||||
|
required: WorkdirDelegationPermission,
|
||||||
|
) -> bool {
|
||||||
|
if required == WorkdirDelegationPermission::Write
|
||||||
|
&& rule.permission != WorkdirDelegationPermission::Write
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
path_in_rule(rule, path)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn path_in_rule(rule: &WorkdirDelegationRule, path: &FsPath) -> bool {
|
||||||
|
let target = Path::new(rule.target.as_str());
|
||||||
|
let path = Path::new(path.as_str());
|
||||||
|
if path == target {
|
||||||
|
return true;
|
||||||
|
}
|
||||||
|
let Ok(suffix) = path.strip_prefix(target) else {
|
||||||
|
return false;
|
||||||
|
};
|
||||||
|
let depth = suffix.components().count();
|
||||||
|
rule.recursive || depth <= 1
|
||||||
|
}
|
||||||
|
|
||||||
|
fn rule_contains_rule(parent: &WorkdirDelegationRule, child: &WorkdirDelegationRule) -> bool {
|
||||||
|
if child.permission == WorkdirDelegationPermission::Write
|
||||||
|
&& parent.permission != WorkdirDelegationPermission::Write
|
||||||
|
{
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
if !path_in_rule(parent, &child.target) {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
parent.recursive || !child.recursive
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use std::fs;
|
||||||
|
|
||||||
|
use manifest::{Permission, Scope, ScopeConfig, ScopeRule, SharedScope};
|
||||||
|
use tempfile::TempDir;
|
||||||
|
|
||||||
|
use super::*;
|
||||||
|
use crate::LocalWorkdirSession;
|
||||||
|
|
||||||
|
fn fs_path(path: &str) -> FsPath {
|
||||||
|
FsPath::new(path).unwrap()
|
||||||
|
}
|
||||||
|
|
||||||
|
fn session(root: &Path) -> WorkdirSessionHandle {
|
||||||
|
let scope = SharedScope::new(
|
||||||
|
Scope::from_config(&ScopeConfig {
|
||||||
|
allow: vec![ScopeRule {
|
||||||
|
target: root.to_path_buf(),
|
||||||
|
permission: Permission::Write,
|
||||||
|
recursive: true,
|
||||||
|
}],
|
||||||
|
deny: Vec::new(),
|
||||||
|
})
|
||||||
|
.unwrap(),
|
||||||
|
);
|
||||||
|
delegation_capable_session(Arc::new(LocalWorkdirSession::materialized_bound(
|
||||||
|
Workdir::new("delegation-test"),
|
||||||
|
root.to_path_buf(),
|
||||||
|
root.to_path_buf(),
|
||||||
|
scope,
|
||||||
|
WorkdirSessionCapabilities::ALL,
|
||||||
|
)))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn request(path: &str, permission: WorkdirDelegationPermission) -> WorkdirDelegationRequest {
|
||||||
|
WorkdirDelegationRequest {
|
||||||
|
rules: vec![WorkdirDelegationRule {
|
||||||
|
target: fs_path(path),
|
||||||
|
permission,
|
||||||
|
recursive: true,
|
||||||
|
}],
|
||||||
|
cwd: fs_path(path),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn read(path: &str) -> ReadRequest {
|
||||||
|
ReadRequest {
|
||||||
|
path: fs_path(path),
|
||||||
|
offset: 0,
|
||||||
|
limit: 20,
|
||||||
|
max_bytes: 1024,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fn write(path: &str, content: &str) -> WriteRequest {
|
||||||
|
WriteRequest {
|
||||||
|
path: fs_path(path),
|
||||||
|
content: content.as_bytes().to_vec(),
|
||||||
|
expected_hash: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn non_recursive_rule_covers_target_and_direct_children_only() {
|
||||||
|
let rule = WorkdirDelegationRule {
|
||||||
|
target: fs_path("docs"),
|
||||||
|
permission: WorkdirDelegationPermission::Read,
|
||||||
|
recursive: false,
|
||||||
|
};
|
||||||
|
assert!(path_in_rule(&rule, &fs_path("docs")));
|
||||||
|
assert!(path_in_rule(&rule, &fs_path("docs/readme.md")));
|
||||||
|
assert!(!path_in_rule(&rule, &fs_path("docs/guides/start.md")));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn read_only_delegation_allows_prefix_and_denies_siblings_and_mutation() {
|
||||||
|
let root = TempDir::new().unwrap();
|
||||||
|
fs::create_dir_all(root.path().join("docs")).unwrap();
|
||||||
|
fs::create_dir_all(root.path().join("secret")).unwrap();
|
||||||
|
fs::write(root.path().join("docs/readme.md"), "visible").unwrap();
|
||||||
|
fs::write(root.path().join("secret/key"), "hidden").unwrap();
|
||||||
|
let parent = session(root.path());
|
||||||
|
|
||||||
|
let child = parent
|
||||||
|
.delegate(request("docs", WorkdirDelegationPermission::Read))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert_eq!(child.capabilities, WorkdirSessionCapabilities::READ_ONLY);
|
||||||
|
assert_eq!(
|
||||||
|
child
|
||||||
|
.scoped_session
|
||||||
|
.read(read("docs/readme.md"))
|
||||||
|
.await
|
||||||
|
.unwrap()
|
||||||
|
.bytes,
|
||||||
|
b"visible"
|
||||||
|
);
|
||||||
|
assert!(matches!(
|
||||||
|
child.scoped_session.read(read("secret/key")).await,
|
||||||
|
Err(WorkdirError::Denied(_))
|
||||||
|
));
|
||||||
|
assert!(matches!(
|
||||||
|
child.scoped_session.write(write("docs/new.md", "no")).await,
|
||||||
|
Err(WorkdirError::Denied(_))
|
||||||
|
));
|
||||||
|
assert!(
|
||||||
|
!child
|
||||||
|
.capabilities
|
||||||
|
.supports(WorkdirSessionCapability::Command)
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn write_lease_blocks_parent_region_until_release() {
|
||||||
|
let root = TempDir::new().unwrap();
|
||||||
|
fs::create_dir_all(root.path().join("leased")).unwrap();
|
||||||
|
fs::create_dir_all(root.path().join("other")).unwrap();
|
||||||
|
let parent = session(root.path());
|
||||||
|
let child = parent
|
||||||
|
.delegate(request("leased", WorkdirDelegationPermission::Write))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
assert!(matches!(
|
||||||
|
parent.write(write("leased/file", "parent")).await,
|
||||||
|
Err(WorkdirError::Denied(_))
|
||||||
|
));
|
||||||
|
parent.write(write("other/file", "parent")).await.unwrap();
|
||||||
|
child
|
||||||
|
.scoped_session
|
||||||
|
.write(write("leased/file", "child"))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
child.release();
|
||||||
|
parent
|
||||||
|
.write(write("leased/parent", "parent"))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(matches!(
|
||||||
|
child.scoped_session.read(read("leased/file")).await,
|
||||||
|
Err(WorkdirError::SessionClosed)
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn nested_delegation_is_attenuated_and_parent_revocation_cascades() {
|
||||||
|
let root = TempDir::new().unwrap();
|
||||||
|
fs::create_dir_all(root.path().join("docs/sub")).unwrap();
|
||||||
|
fs::create_dir_all(root.path().join("docs/peer")).unwrap();
|
||||||
|
fs::write(root.path().join("docs/sub/a"), "a").unwrap();
|
||||||
|
fs::write(root.path().join("docs/peer/b"), "b").unwrap();
|
||||||
|
let root_session = session(root.path());
|
||||||
|
let child = root_session
|
||||||
|
.delegate(request("docs", WorkdirDelegationPermission::Read))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
let nested = child
|
||||||
|
.scoped_session
|
||||||
|
.delegate(request("docs/sub", WorkdirDelegationPermission::Read))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
nested
|
||||||
|
.scoped_session
|
||||||
|
.read(read("docs/sub/a"))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
assert!(matches!(
|
||||||
|
nested.scoped_session.read(read("docs/peer/b")).await,
|
||||||
|
Err(WorkdirError::Denied(_))
|
||||||
|
));
|
||||||
|
assert!(
|
||||||
|
child
|
||||||
|
.scoped_session
|
||||||
|
.delegate(request("docs/sub", WorkdirDelegationPermission::Write))
|
||||||
|
.await
|
||||||
|
.is_err()
|
||||||
|
);
|
||||||
|
|
||||||
|
child.release();
|
||||||
|
assert!(matches!(
|
||||||
|
nested.scoped_session.read(read("docs/sub/a")).await,
|
||||||
|
Err(WorkdirError::SessionClosed)
|
||||||
|
));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::test]
|
||||||
|
async fn closing_parent_invalidates_delegated_sessions() {
|
||||||
|
let root = TempDir::new().unwrap();
|
||||||
|
fs::create_dir_all(root.path().join("docs")).unwrap();
|
||||||
|
fs::write(root.path().join("docs/a"), "a").unwrap();
|
||||||
|
let parent = session(root.path());
|
||||||
|
let child = parent
|
||||||
|
.delegate(request("docs", WorkdirDelegationPermission::Read))
|
||||||
|
.await
|
||||||
|
.unwrap();
|
||||||
|
|
||||||
|
parent.close().await.unwrap();
|
||||||
|
assert!(matches!(
|
||||||
|
child.scoped_session.read(read("docs/a")).await,
|
||||||
|
Err(WorkdirError::SessionClosed)
|
||||||
|
));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -120,7 +120,10 @@ impl WorkdirTransportError {
|
|||||||
WorkdirError::UnknownCommand(_) => {
|
WorkdirError::UnknownCommand(_) => {
|
||||||
(Code::UnknownCommand, "Workdir command was not found")
|
(Code::UnknownCommand, "Workdir command was not found")
|
||||||
}
|
}
|
||||||
WorkdirError::Unavailable(_) => (Code::Unavailable, "Workdir session is unavailable"),
|
WorkdirError::Unavailable(_) | WorkdirError::SessionClosed => {
|
||||||
|
(Code::Unavailable, "Workdir session is unavailable")
|
||||||
|
}
|
||||||
|
WorkdirError::Denied(_) => (Code::InvalidRequest, "Workdir operation was denied"),
|
||||||
WorkdirError::Transport(_) => (Code::Internal, "Workdir transport failed"),
|
WorkdirError::Transport(_) => (Code::Internal, "Workdir transport failed"),
|
||||||
WorkdirError::InvalidPath(_)
|
WorkdirError::InvalidPath(_)
|
||||||
| WorkdirError::RelativePath(_)
|
| WorkdirError::RelativePath(_)
|
||||||
@@ -169,7 +172,7 @@ mod client {
|
|||||||
use reqwest::{Client, StatusCode, Url};
|
use reqwest::{Client, StatusCode, Url};
|
||||||
|
|
||||||
use super::*;
|
use super::*;
|
||||||
use crate::{Workdir, WorkdirSession};
|
use crate::{Workdir, WorkdirSession, WorkdirSessionHandle};
|
||||||
|
|
||||||
/// 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
|
||||||
@@ -310,6 +313,21 @@ mod client {
|
|||||||
self.capabilities
|
self.capabilities
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn capture_delegation_source(&self) -> Result<WorkdirSessionHandle, WorkdirError> {
|
||||||
|
if self.closed.load(Ordering::Acquire) {
|
||||||
|
return Err(WorkdirError::SessionClosed);
|
||||||
|
}
|
||||||
|
Ok(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,
|
||||||
|
closed: AtomicBool::new(false),
|
||||||
|
}))
|
||||||
|
}
|
||||||
|
|
||||||
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),
|
||||||
|
|||||||
+35
-100
@@ -5,6 +5,7 @@
|
|||||||
//! 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;
|
||||||
@@ -12,11 +13,14 @@ pub mod workspace;
|
|||||||
|
|
||||||
use std::path::{Path, PathBuf};
|
use std::path::{Path, PathBuf};
|
||||||
use std::sync::Arc;
|
use std::sync::Arc;
|
||||||
use std::sync::atomic::{AtomicBool, Ordering};
|
|
||||||
|
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
use serde::{Deserialize, Serialize};
|
use serde::{Deserialize, Serialize};
|
||||||
|
|
||||||
|
pub use delegation::{
|
||||||
|
ReadOnlyWorkdirSession, WorkdirDelegation, WorkdirDelegationPermission,
|
||||||
|
WorkdirDelegationRequest, WorkdirDelegationRule, 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,
|
||||||
@@ -140,6 +144,30 @@ 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
|
||||||
|
}
|
||||||
|
|
||||||
|
/// 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) -> 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>;
|
||||||
@@ -160,107 +188,14 @@ pub trait WorkdirSession: std::fmt::Debug + Send + Sync {
|
|||||||
|
|
||||||
pub type WorkdirSessionHandle = Arc<dyn WorkdirSession>;
|
pub type WorkdirSessionHandle = Arc<dyn WorkdirSession>;
|
||||||
|
|
||||||
/// Ephemeral least-authority view over an existing Workdir session.
|
|
||||||
///
|
|
||||||
/// The wrapper exposes only stat/read/list/glob/grep and never forwards write,
|
|
||||||
/// edit, command, or close authority to the underlying Worker session. Closing
|
|
||||||
/// the wrapper is terminal for the view but deliberately leaves the owner's
|
|
||||||
/// source session open.
|
|
||||||
#[derive(Debug)]
|
|
||||||
pub struct ReadOnlyWorkdirSession {
|
|
||||||
source: WorkdirSessionHandle,
|
|
||||||
closed: AtomicBool,
|
|
||||||
}
|
|
||||||
|
|
||||||
impl ReadOnlyWorkdirSession {
|
|
||||||
pub fn new(source: WorkdirSessionHandle) -> Self {
|
|
||||||
Self {
|
|
||||||
source,
|
|
||||||
closed: AtomicBool::new(false),
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
fn ensure_open(&self) -> Result<(), WorkdirError> {
|
|
||||||
if self.closed.load(Ordering::Acquire) {
|
|
||||||
Err(WorkdirError::Unavailable(
|
|
||||||
"read-only Workdir session is closed".to_string(),
|
|
||||||
))
|
|
||||||
} else {
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[async_trait]
|
|
||||||
impl WorkdirSession for ReadOnlyWorkdirSession {
|
|
||||||
fn workdir(&self) -> &Workdir {
|
|
||||||
self.source.workdir()
|
|
||||||
}
|
|
||||||
|
|
||||||
fn capabilities(&self) -> WorkdirSessionCapabilities {
|
|
||||||
WorkdirSessionCapabilities::READ_ONLY
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn stat(&self, request: StatRequest) -> Result<StatResult, WorkdirError> {
|
|
||||||
self.ensure_open()?;
|
|
||||||
self.source.stat(request).await
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn read(&self, request: ReadRequest) -> Result<ReadResult, WorkdirError> {
|
|
||||||
self.ensure_open()?;
|
|
||||||
self.source.read(request).await
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn write(&self, _request: WriteRequest) -> Result<WriteResult, WorkdirError> {
|
|
||||||
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Write))
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn edit(&self, _request: EditRequest) -> Result<EditResult, WorkdirError> {
|
|
||||||
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Edit))
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn list(&self, request: ListRequest) -> Result<ListResult, WorkdirError> {
|
|
||||||
self.ensure_open()?;
|
|
||||||
self.source.list(request).await
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn glob(&self, request: GlobRequest) -> Result<GlobResult, WorkdirError> {
|
|
||||||
self.ensure_open()?;
|
|
||||||
self.source.glob(request).await
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn grep(&self, request: GrepRequest) -> Result<GrepResult, WorkdirError> {
|
|
||||||
self.ensure_open()?;
|
|
||||||
self.source.grep(request).await
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn start_command(&self, _request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
|
|
||||||
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn command_status(&self, _handle: CommandHandle) -> Result<CommandStatus, WorkdirError> {
|
|
||||||
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn command_output(
|
|
||||||
&self,
|
|
||||||
_request: CommandOutputRequest,
|
|
||||||
) -> Result<CommandOutput, WorkdirError> {
|
|
||||||
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn cancel_command(&self, _handle: CommandHandle) -> Result<(), WorkdirError> {
|
|
||||||
Err(WorkdirError::Unsupported(WorkdirSessionCapability::Command))
|
|
||||||
}
|
|
||||||
|
|
||||||
async fn close(&self) -> Result<(), WorkdirError> {
|
|
||||||
self.closed.store(true, Ordering::Release);
|
|
||||||
Ok(())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, thiserror::Error)]
|
#[derive(Debug, thiserror::Error)]
|
||||||
pub enum WorkdirError {
|
pub enum WorkdirError {
|
||||||
|
#[error("Workdir operation denied: {0}")]
|
||||||
|
Denied(String),
|
||||||
|
|
||||||
|
#[error("Workdir session is closed")]
|
||||||
|
SessionClosed,
|
||||||
|
|
||||||
#[error("Workdir session does not support {0:?}")]
|
#[error("Workdir session does not support {0:?}")]
|
||||||
Unsupported(WorkdirSessionCapability),
|
Unsupported(WorkdirSessionCapability),
|
||||||
|
|
||||||
|
|||||||
@@ -29,8 +29,8 @@ use crate::{
|
|||||||
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,
|
||||||
ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirError, WorkdirPath,
|
ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirError, WorkdirPath,
|
||||||
WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability, WriteRequest,
|
WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability, WorkdirSessionHandle,
|
||||||
WriteResult,
|
WriteRequest, WriteResult,
|
||||||
};
|
};
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
use crate::{EntryKind, WriteOutcome};
|
use crate::{EntryKind, WriteOutcome};
|
||||||
@@ -371,6 +371,10 @@ impl WorkdirSession for LocalWorkdirSession {
|
|||||||
self.inner.capabilities
|
self.inner.capabilities
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn capture_delegation_source(&self) -> Result<WorkdirSessionHandle, WorkdirError> {
|
||||||
|
Ok(Arc::new(self.clone()))
|
||||||
|
}
|
||||||
|
|
||||||
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();
|
||||||
|
|||||||
@@ -314,3 +314,17 @@ mod tests {
|
|||||||
assert_eq!(decoded, detail);
|
assert_eq!(decoded, detail);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
#[serde(deny_unknown_fields)]
|
||||||
|
pub struct WorkspaceWorkdirSessionOperationRequest {
|
||||||
|
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||||
|
pub expected_session_fence: Option<String>,
|
||||||
|
pub operation: crate::http::WorkdirSessionOperation,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)]
|
||||||
|
#[serde(deny_unknown_fields)]
|
||||||
|
pub struct WorkspaceWorkdirSessionFence {
|
||||||
|
pub value: String,
|
||||||
|
}
|
||||||
|
|||||||
@@ -33,6 +33,7 @@ config-source = { path = "../config-source" }
|
|||||||
include_dir = "0.7.4"
|
include_dir = "0.7.4"
|
||||||
fs4 = { workspace = true, features = ["sync"] }
|
fs4 = { workspace = true, features = ["sync"] }
|
||||||
flow = { path = "../flow" }
|
flow = { path = "../flow" }
|
||||||
|
fs-operation = { workspace = true }
|
||||||
libc = { workspace = true }
|
libc = { workspace = true }
|
||||||
schemars = { workspace = true }
|
schemars = { workspace = true }
|
||||||
ticket = { workspace = true }
|
ticket = { workspace = true }
|
||||||
|
|||||||
@@ -681,18 +681,20 @@ where
|
|||||||
{
|
{
|
||||||
// Worker-immutable snapshots taken before the mutable worker borrow
|
// Worker-immutable snapshots taken before the mutable worker borrow
|
||||||
// below so the worker borrow doesn't conflict with reads on `worker`.
|
// below so the worker borrow doesn't conflict with reads on `worker`.
|
||||||
let scope_handle = worker.scope().clone();
|
|
||||||
let feature_config = worker.manifest().feature.clone();
|
let feature_config = worker.manifest().feature.clone();
|
||||||
if feature_config.manage_workdir.enabled {
|
if feature_config.manage_workdir.enabled && worker.workdir_session().is_none() {
|
||||||
if let Some(existing) = worker.workdir_session().cloned() {
|
|
||||||
existing.close().await.map_err(std::io::Error::other)?;
|
|
||||||
}
|
|
||||||
let workspace_client = worker.workspace_client_handle();
|
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(
|
crate::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession::handle(
|
||||||
workspace_client,
|
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 worker_workdir = worker.workdir_session().cloned();
|
||||||
let local_filesystem = worker.local_working_directory().cloned();
|
let local_filesystem = worker.local_working_directory().cloned();
|
||||||
@@ -844,6 +846,7 @@ 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();
|
||||||
@@ -902,27 +905,18 @@ where
|
|||||||
Arc<dyn crate::feature::builtin::worker_observation::WorkerObservationProvider>,
|
Arc<dyn crate::feature::builtin::worker_observation::WorkerObservationProvider>,
|
||||||
> = Vec::new();
|
> = Vec::new();
|
||||||
|
|
||||||
// Worker-orchestration tools (SubWorkerSpawn + three control tools) share
|
// Worker-orchestration tools derive child filesystem authority from the
|
||||||
// the Worker-scoped `SpawnedWorkerRegistry` (also consumed by the main
|
// active provider-backed Workdir session. The tool remains registered
|
||||||
// loop's `WorkerEvent` handler). Expose them only behind the explicit
|
// without one so invocation fails deterministically until the parent
|
||||||
// profile feature and require delegation authority up front so enabling
|
// attaches a Workdir.
|
||||||
// the surface cannot imply broad child scope by accident.
|
|
||||||
if feature_config.sub_worker.enabled {
|
if feature_config.sub_worker.enabled {
|
||||||
let spawner_cwd = local_filesystem
|
let spawner_cwd = local_filesystem
|
||||||
.as_ref()
|
.as_ref()
|
||||||
.map(|local| local.cwd.clone())
|
.map(|local| local.cwd.clone())
|
||||||
.ok_or_else(|| {
|
.unwrap_or_else(|| PathBuf::from("/"));
|
||||||
std::io::Error::new(
|
let spawner_workspace_root = local_workspace_root
|
||||||
std::io::ErrorKind::InvalidInput,
|
.clone()
|
||||||
"worker spawn tools require local Worker filesystem authority",
|
.unwrap_or_else(|| PathBuf::from("/"));
|
||||||
)
|
|
||||||
})?;
|
|
||||||
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",
|
|
||||||
)
|
|
||||||
})?;
|
|
||||||
engine.register_tool(sub_worker_spawn_tool(
|
engine.register_tool(sub_worker_spawn_tool(
|
||||||
spawner_name.clone(),
|
spawner_name.clone(),
|
||||||
spawner_workspace_context,
|
spawner_workspace_context,
|
||||||
@@ -930,9 +924,9 @@ where
|
|||||||
runtime_base.clone(),
|
runtime_base.clone(),
|
||||||
spawner_workspace_root,
|
spawner_workspace_root,
|
||||||
spawner_cwd.clone(),
|
spawner_cwd.clone(),
|
||||||
|
source_workdir_session,
|
||||||
spawned_registry.clone(),
|
spawned_registry.clone(),
|
||||||
spawner_manifest,
|
spawner_manifest,
|
||||||
scope_handle,
|
|
||||||
prompts,
|
prompts,
|
||||||
));
|
));
|
||||||
observation_providers.push(Arc::new(
|
observation_providers.push(Arc::new(
|
||||||
|
|||||||
@@ -16,7 +16,8 @@ use serde_json::json;
|
|||||||
use workdir::http::{WorkdirSessionOperation, WorkdirSessionOperationResult};
|
use workdir::http::{WorkdirSessionOperation, WorkdirSessionOperationResult};
|
||||||
use workdir::workspace::{
|
use workdir::workspace::{
|
||||||
WorkingDirectoryDetailResponse as WorkdirDetailResponse,
|
WorkingDirectoryDetailResponse as WorkdirDetailResponse,
|
||||||
WorkingDirectoryListResponse as WorkdirListResponse,
|
WorkingDirectoryListResponse as WorkdirListResponse, WorkspaceWorkdirSessionFence,
|
||||||
|
WorkspaceWorkdirSessionOperationRequest,
|
||||||
};
|
};
|
||||||
use workdir::{
|
use workdir::{
|
||||||
CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest,
|
CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest,
|
||||||
@@ -153,6 +154,7 @@ 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>,
|
||||||
}
|
}
|
||||||
|
|
||||||
impl WorkspaceAttachedWorkdirSession {
|
impl WorkspaceAttachedWorkdirSession {
|
||||||
@@ -160,6 +162,7 @@ impl WorkspaceAttachedWorkdirSession {
|
|||||||
Arc::new(Self {
|
Arc::new(Self {
|
||||||
client,
|
client,
|
||||||
workdir: Workdir::new("workspace-attachment"),
|
workdir: Workdir::new("workspace-attachment"),
|
||||||
|
expected_session_fence: None,
|
||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -176,7 +179,11 @@ 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(&operation).map_err(|error| {
|
serde_json::to_string(&WorkspaceWorkdirSessionOperationRequest {
|
||||||
|
expected_session_fence: self.expected_session_fence.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}"
|
||||||
))
|
))
|
||||||
@@ -224,6 +231,43 @@ impl WorkdirSession for WorkspaceAttachedWorkdirSession {
|
|||||||
WorkdirSessionCapabilities::ALL
|
WorkdirSessionCapabilities::ALL
|
||||||
}
|
}
|
||||||
|
|
||||||
|
async fn capture_delegation_source(&self) -> 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
|
||||||
|
};
|
||||||
|
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<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),
|
||||||
@@ -1002,11 +1046,55 @@ mod tests {
|
|||||||
);
|
);
|
||||||
let body: serde_json::Value =
|
let body: serde_json::Value =
|
||||||
serde_json::from_str(requests[0].body.as_deref().unwrap()).unwrap();
|
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("runtime_id").is_none());
|
||||||
assert!(body.get("session_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]
|
#[test]
|
||||||
fn invalid_or_extra_inputs_are_rejected_before_workspace_request() {
|
fn invalid_or_extra_inputs_are_rejected_before_workspace_request() {
|
||||||
let client = Arc::new(RecordingWorkspaceClient::new(Vec::new()));
|
let client = Arc::new(RecordingWorkspaceClient::new(Vec::new()));
|
||||||
|
|||||||
@@ -19,6 +19,7 @@ use session_store::{
|
|||||||
WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerStoreError,
|
WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerStoreError,
|
||||||
};
|
};
|
||||||
use tracing::warn;
|
use tracing::warn;
|
||||||
|
use workdir::WorkdirDelegation;
|
||||||
|
|
||||||
use crate::internal_worker::InternalWorkerSessionHandle;
|
use crate::internal_worker::InternalWorkerSessionHandle;
|
||||||
use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord};
|
use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord};
|
||||||
@@ -28,6 +29,9 @@ use crate::runtime::worker_allocation;
|
|||||||
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>,
|
||||||
|
#[cfg(test)]
|
||||||
|
pub installed_tools: Arc<[String]>,
|
||||||
pub session: InternalWorkerSessionHandle,
|
pub session: InternalWorkerSessionHandle,
|
||||||
scope_reclaimed: Arc<AtomicBool>,
|
scope_reclaimed: Arc<AtomicBool>,
|
||||||
}
|
}
|
||||||
@@ -36,11 +40,16 @@ 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,
|
||||||
|
#[cfg(test)] installed_tools: Vec<String>,
|
||||||
session: InternalWorkerSessionHandle,
|
session: InternalWorkerSessionHandle,
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
worker_name,
|
worker_name,
|
||||||
scope_delegated,
|
scope_delegated,
|
||||||
|
workdir_delegation: Arc::new(workdir_delegation),
|
||||||
|
#[cfg(test)]
|
||||||
|
installed_tools: installed_tools.into(),
|
||||||
session,
|
session,
|
||||||
scope_reclaimed: Arc::new(AtomicBool::new(false)),
|
scope_reclaimed: Arc::new(AtomicBool::new(false)),
|
||||||
}
|
}
|
||||||
@@ -247,6 +256,7 @@ impl SpawnedWorkerRegistry {
|
|||||||
if !record.claim_scope_reclaim() {
|
if !record.claim_scope_reclaim() {
|
||||||
return Ok(false);
|
return Ok(false);
|
||||||
}
|
}
|
||||||
|
record.workdir_delegation.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)))
|
||||||
|
|||||||
+156
-188
@@ -10,16 +10,21 @@ use std::sync::Arc;
|
|||||||
|
|
||||||
use arc_swap::ArcSwap;
|
use arc_swap::ArcSwap;
|
||||||
use async_trait::async_trait;
|
use async_trait::async_trait;
|
||||||
|
use fs_operation::FsPath;
|
||||||
use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
|
use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
|
||||||
use manifest::{
|
use manifest::{
|
||||||
CompactionConfigPartial, DelegationScope, EngineManifestConfig, FileUploadLimitsPartial,
|
CompactionConfigPartial, EngineManifestConfig, FileUploadLimitsPartial, Permission,
|
||||||
Permission, PermissionConfigPartial, ProfileDiscovery, ProfileError, ProfileRegistry,
|
PermissionConfigPartial, ProfileDiscovery, ProfileError, ProfileRegistry,
|
||||||
ProfileRegistrySource, ProfileResolveOptions, ProfileResolver, ProfileSelector, Scope,
|
ProfileRegistrySource, ProfileResolveOptions, ProfileResolver, ProfileSelector, ScopeConfig,
|
||||||
ScopeConfig, ScopeRule, SessionConfigPartial, SharedScope, ToolOutputLimitsPartial,
|
ScopeRule, SessionConfigPartial, ToolOutputLimitsPartial, WorkerManifest, WorkerManifestConfig,
|
||||||
WorkerManifest, WorkerManifestConfig, WorkerMetaConfig,
|
WorkerMetaConfig,
|
||||||
};
|
};
|
||||||
use serde::Deserialize;
|
use serde::Deserialize;
|
||||||
use tokio::sync::mpsc;
|
use tokio::sync::mpsc;
|
||||||
|
use workdir::{
|
||||||
|
WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule,
|
||||||
|
WorkdirSessionHandle,
|
||||||
|
};
|
||||||
|
|
||||||
use crate::PromptCatalogSource;
|
use crate::PromptCatalogSource;
|
||||||
use crate::controller::register_worker_tools;
|
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
|
/// 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.
|
||||||
spawner_cwd: PathBuf,
|
spawner_cwd: PathBuf,
|
||||||
|
/// Active provider-backed Workdir session from which child leases are captured.
|
||||||
|
source_workdir_session: Option<WorkdirSessionHandle>,
|
||||||
/// 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
|
||||||
@@ -279,18 +286,6 @@ pub struct SubWorkerSpawnTool {
|
|||||||
prompt_loader: PromptCatalogSource,
|
prompt_loader: PromptCatalogSource,
|
||||||
/// Compact selector list shared by tool description and diagnostics.
|
/// Compact selector list shared by tool description and diagnostics.
|
||||||
available_profiles: AvailableProfiles,
|
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<Box<dyn llm_engine::llm_client::LlmClient>>,
|
internal_client_override: Option<Box<dyn llm_engine::llm_client::LlmClient>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -308,12 +303,11 @@ impl SubWorkerSpawnTool {
|
|||||||
runtime_base: PathBuf,
|
runtime_base: PathBuf,
|
||||||
workspace_root: PathBuf,
|
workspace_root: PathBuf,
|
||||||
spawner_cwd: PathBuf,
|
spawner_cwd: PathBuf,
|
||||||
|
source_workdir_session: Option<WorkdirSessionHandle>,
|
||||||
registry: Arc<SpawnedWorkerRegistry>,
|
registry: Arc<SpawnedWorkerRegistry>,
|
||||||
spawner_manifest: WorkerManifest,
|
spawner_manifest: WorkerManifest,
|
||||||
prompt_loader: PromptCatalogSource,
|
prompt_loader: PromptCatalogSource,
|
||||||
available_profiles: AvailableProfiles,
|
available_profiles: AvailableProfiles,
|
||||||
spawner_scope: SharedScope,
|
|
||||||
delegation_scope: DelegationScope,
|
|
||||||
) -> Self {
|
) -> Self {
|
||||||
Self {
|
Self {
|
||||||
spawner_name,
|
spawner_name,
|
||||||
@@ -322,12 +316,11 @@ impl SubWorkerSpawnTool {
|
|||||||
runtime_base,
|
runtime_base,
|
||||||
workspace_root,
|
workspace_root,
|
||||||
spawner_cwd,
|
spawner_cwd,
|
||||||
|
source_workdir_session,
|
||||||
registry,
|
registry,
|
||||||
spawner_manifest,
|
spawner_manifest,
|
||||||
prompt_loader,
|
prompt_loader,
|
||||||
available_profiles,
|
available_profiles,
|
||||||
spawner_scope,
|
|
||||||
delegation_scope,
|
|
||||||
internal_client_override: None,
|
internal_client_override: None,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -386,8 +379,16 @@ impl Tool for SubWorkerSpawnTool {
|
|||||||
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
|
.map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
|
||||||
|
|
||||||
let scope_allow = parse_scope(&input.scope)?;
|
let scope_allow = parse_scope(&input.scope)?;
|
||||||
self.validate_delegation_scope(&scope_allow)?;
|
let source_workdir_session =
|
||||||
let child_cwd = validate_spawn_cwd(input.cwd.as_deref(), &scope_allow, &self.spawner_cwd)?;
|
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 =
|
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| {
|
||||||
@@ -413,11 +414,14 @@ impl Tool for SubWorkerSpawnTool {
|
|||||||
allow: scope_allow.clone(),
|
allow: scope_allow.clone(),
|
||||||
deny: Vec::new(),
|
deny: Vec::new(),
|
||||||
};
|
};
|
||||||
let child_manifest =
|
let mut child_manifest =
|
||||||
WorkerManifest::try_from(WorkerManifestConfig::builtin_defaults().merge(child_config))
|
WorkerManifest::try_from(WorkerManifestConfig::builtin_defaults().merge(child_config))
|
||||||
.map_err(|error| {
|
.map_err(|error| {
|
||||||
ToolError::ExecutionFailed(format!("resolve child manifest: {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| {
|
let reviewer_capability = input.review.as_ref().map(|review| {
|
||||||
(
|
(
|
||||||
review.ticket_id.clone(),
|
review.ticket_id.clone(),
|
||||||
@@ -458,8 +462,7 @@ impl Tool for SubWorkerSpawnTool {
|
|||||||
self.workspace_context.clone()
|
self.workspace_context.clone()
|
||||||
};
|
};
|
||||||
let store = EphemeralSessionStore::default();
|
let store = EphemeralSessionStore::default();
|
||||||
let filesystem_authority =
|
let filesystem_authority = WorkerFilesystemAuthority::None;
|
||||||
WorkerFilesystemAuthority::local(self.workspace_root.clone(), child_cwd.clone());
|
|
||||||
let mut child = Worker::<Box<dyn llm_engine::llm_client::LlmClient>, EphemeralSessionStore>::from_internal_manifest_with_context(
|
let mut child = Worker::<Box<dyn llm_engine::llm_client::LlmClient>, EphemeralSessionStore>::from_internal_manifest_with_context(
|
||||||
child_manifest,
|
child_manifest,
|
||||||
store.clone(),
|
store.clone(),
|
||||||
@@ -472,6 +475,7 @@ 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()));
|
||||||
let child_scope = child.scope().clone();
|
let child_scope = child.scope().clone();
|
||||||
let child_registry = SpawnedWorkerRegistry::new_internal(input.name.clone(), child_scope);
|
let child_registry = SpawnedWorkerRegistry::new_internal(input.name.clone(), child_scope);
|
||||||
register_worker_tools(
|
register_worker_tools(
|
||||||
@@ -488,23 +492,14 @@ impl Tool for SubWorkerSpawnTool {
|
|||||||
.map_err(|error| {
|
.map_err(|error| {
|
||||||
ToolError::ExecutionFailed(format!("install Internal Worker features: {error}"))
|
ToolError::ExecutionFailed(format!("install Internal Worker features: {error}"))
|
||||||
})?;
|
})?;
|
||||||
// Transfer delegated Write authority before the child accepts its first turn. This closes
|
#[cfg(test)]
|
||||||
// the parallel-tool window where parent and child could otherwise both write the same path.
|
let installed_tools = child
|
||||||
// The machine-wide allocation remains owned by the parent Worker; no fake child PID/socket
|
.engine()
|
||||||
// identity is introduced.
|
.tool_server_handle()
|
||||||
let revoke_write: Vec<ScopeRule> = scope_allow
|
.tool_definitions_sorted()
|
||||||
.iter()
|
.into_iter()
|
||||||
.filter(|rule| rule.permission == Permission::Write)
|
.map(|definition| definition.name)
|
||||||
.cloned()
|
|
||||||
.collect();
|
.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 child_name = input.name.clone();
|
||||||
let registry = Arc::downgrade(&self.registry);
|
let registry = Arc::downgrade(&self.registry);
|
||||||
let parent_notifications = self.parent_notifications.clone();
|
let parent_notifications = self.parent_notifications.clone();
|
||||||
@@ -530,19 +525,9 @@ impl Tool for SubWorkerSpawnTool {
|
|||||||
})),
|
})),
|
||||||
)
|
)
|
||||||
.await;
|
.await;
|
||||||
let session = match session_result {
|
let session = session_result.map_err(|error| {
|
||||||
Ok(session) => session,
|
ToolError::ExecutionFailed(format!("prepare Internal Worker session: {error}"))
|
||||||
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}"
|
|
||||||
)));
|
|
||||||
}
|
|
||||||
};
|
|
||||||
|
|
||||||
if let Some((ticket_id, capability_token)) = &reviewer_capability {
|
if let Some((ticket_id, capability_token)) = &reviewer_capability {
|
||||||
let workspace_id = self.workspace_context.workspace_id().ok_or_else(|| {
|
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(
|
let record = crate::spawn::registry::InternalSpawnedWorkerRecord::new(
|
||||||
input.name.clone(),
|
input.name.clone(),
|
||||||
scope_allow,
|
scope_allow,
|
||||||
|
workdir_delegation,
|
||||||
|
#[cfg(test)]
|
||||||
|
installed_tools,
|
||||||
session.clone(),
|
session.clone(),
|
||||||
);
|
);
|
||||||
if let Err(error) = name_reservation.commit(record) {
|
if let Err(error) = name_reservation.commit(record) {
|
||||||
let _ = session.stop().await;
|
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!(
|
return Err(ToolError::ExecutionFailed(format!(
|
||||||
"register Internal Worker session: {error}"
|
"register Internal Worker session: {error}"
|
||||||
)));
|
)));
|
||||||
@@ -636,27 +619,73 @@ impl Tool for SubWorkerSpawnTool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
impl SubWorkerSpawnTool {
|
impl SubWorkerSpawnTool {
|
||||||
fn validate_delegation_scope(&self, scope_allow: &[ScopeRule]) -> Result<(), ToolError> {
|
fn workdir_delegation_request(
|
||||||
if self.delegation_scope.is_empty() && !scope_allow.is_empty() {
|
&self,
|
||||||
return Err(ToolError::InvalidArgument(
|
cwd: Option<&Path>,
|
||||||
"SubWorkerSpawn requires delegation authority, but this Worker has no delegation scope grant; direct filesystem scope only authorizes this Worker's own tools".into(),
|
scope_allow: &[ScopeRule],
|
||||||
));
|
) -> Result<WorkdirDelegationRequest, ToolError> {
|
||||||
|
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::<Result<Vec<_>, 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 {
|
Ok(WorkdirDelegationRequest {
|
||||||
let allowed = self
|
rules,
|
||||||
.delegation_scope
|
cwd: self.logical_workdir_path(cwd)?,
|
||||||
.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(())
|
|
||||||
}
|
}
|
||||||
|
|
||||||
|
fn logical_workdir_path(&self, path: &Path) -> Result<FsPath, ToolError> {
|
||||||
|
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<Vec<ScopeRule>, ToolError> {
|
fn parse_scope(rules: &[ScopeRuleInput]) -> Result<Vec<ScopeRule>, ToolError> {
|
||||||
@@ -681,63 +710,6 @@ fn parse_scope(rules: &[ScopeRuleInput]) -> Result<Vec<ScopeRule>, ToolError> {
|
|||||||
.collect()
|
.collect()
|
||||||
}
|
}
|
||||||
|
|
||||||
fn validate_spawn_cwd(
|
|
||||||
cwd: Option<&Path>,
|
|
||||||
scope_allow: &[ScopeRule],
|
|
||||||
default_cwd: &Path,
|
|
||||||
) -> Result<PathBuf, ToolError> {
|
|
||||||
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
|
/// Serialise the internal manifest config that gets handed to the child
|
||||||
/// Worker runtime process via the hidden `--spawn-config-json` flag.
|
/// Worker runtime process via the hidden `--spawn-config-json` flag.
|
||||||
/// `WorkerManifestConfig`'s `Serialize` impl is the single source of truth for the
|
/// `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,
|
runtime_base: PathBuf,
|
||||||
workspace_root: PathBuf,
|
workspace_root: PathBuf,
|
||||||
spawner_cwd: PathBuf,
|
spawner_cwd: PathBuf,
|
||||||
|
source_workdir_session: Option<WorkdirSessionHandle>,
|
||||||
registry: Arc<SpawnedWorkerRegistry>,
|
registry: Arc<SpawnedWorkerRegistry>,
|
||||||
spawner_manifest: WorkerManifest,
|
spawner_manifest: WorkerManifest,
|
||||||
spawner_scope: SharedScope,
|
|
||||||
prompts: Arc<ArcSwap<PromptCatalog>>,
|
prompts: Arc<ArcSwap<PromptCatalog>>,
|
||||||
) -> ToolDefinition {
|
) -> ToolDefinition {
|
||||||
sub_worker_spawn_tool_impl(
|
sub_worker_spawn_tool_impl(
|
||||||
@@ -956,9 +928,9 @@ pub(crate) fn sub_worker_spawn_tool(
|
|||||||
runtime_base,
|
runtime_base,
|
||||||
workspace_root,
|
workspace_root,
|
||||||
spawner_cwd,
|
spawner_cwd,
|
||||||
|
source_workdir_session,
|
||||||
registry,
|
registry,
|
||||||
spawner_manifest,
|
spawner_manifest,
|
||||||
spawner_scope,
|
|
||||||
prompts,
|
prompts,
|
||||||
)
|
)
|
||||||
}
|
}
|
||||||
@@ -970,9 +942,9 @@ fn sub_worker_spawn_tool_impl(
|
|||||||
runtime_base: PathBuf,
|
runtime_base: PathBuf,
|
||||||
workspace_root: PathBuf,
|
workspace_root: PathBuf,
|
||||||
spawner_cwd: PathBuf,
|
spawner_cwd: PathBuf,
|
||||||
|
source_workdir_session: Option<WorkdirSessionHandle>,
|
||||||
registry: Arc<SpawnedWorkerRegistry>,
|
registry: Arc<SpawnedWorkerRegistry>,
|
||||||
spawner_manifest: WorkerManifest,
|
spawner_manifest: WorkerManifest,
|
||||||
spawner_scope: SharedScope,
|
|
||||||
prompts: Arc<ArcSwap<PromptCatalog>>,
|
prompts: Arc<ArcSwap<PromptCatalog>>,
|
||||||
) -> ToolDefinition {
|
) -> ToolDefinition {
|
||||||
Arc::new(move || {
|
Arc::new(move || {
|
||||||
@@ -1002,13 +974,11 @@ fn sub_worker_spawn_tool_impl(
|
|||||||
runtime_base.clone(),
|
runtime_base.clone(),
|
||||||
workspace_root.clone(),
|
workspace_root.clone(),
|
||||||
spawner_cwd.clone(),
|
spawner_cwd.clone(),
|
||||||
|
source_workdir_session.clone(),
|
||||||
registry.clone(),
|
registry.clone(),
|
||||||
spawner_manifest.clone(),
|
spawner_manifest.clone(),
|
||||||
prompts.load_full().source(),
|
prompts.load_full().source(),
|
||||||
available_profiles,
|
available_profiles,
|
||||||
spawner_scope.clone(),
|
|
||||||
DelegationScope::from_config(&spawner_manifest.delegation_scope)
|
|
||||||
.expect("resolved Worker manifest has a valid delegation scope"),
|
|
||||||
));
|
));
|
||||||
(meta, tool)
|
(meta, tool)
|
||||||
})
|
})
|
||||||
@@ -1017,6 +987,7 @@ fn sub_worker_spawn_tool_impl(
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use manifest::{DelegationScope, Scope, SharedScope};
|
||||||
use std::pin::Pin;
|
use std::pin::Pin;
|
||||||
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
|
||||||
use std::time::Duration;
|
use std::time::Duration;
|
||||||
@@ -1035,6 +1006,16 @@ mod tests {
|
|||||||
WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceResponse,
|
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]
|
#[test]
|
||||||
fn reviewer_handoff_requires_explicit_builtin_profile_and_read_only_scope() {
|
fn reviewer_handoff_requires_explicit_builtin_profile_and_read_only_scope() {
|
||||||
let valid: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
|
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 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(
|
||||||
|
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(
|
let tool = SubWorkerSpawnTool::new(
|
||||||
"parent".into(),
|
"parent".into(),
|
||||||
workspace_context,
|
workspace_context,
|
||||||
@@ -1139,12 +1129,11 @@ extract_threshold = 4000
|
|||||||
runtime.path().to_path_buf(),
|
runtime.path().to_path_buf(),
|
||||||
workspace_root.clone(),
|
workspace_root.clone(),
|
||||||
workspace_root.clone(),
|
workspace_root.clone(),
|
||||||
|
Some(source_workdir_session),
|
||||||
registry.clone(),
|
registry.clone(),
|
||||||
manifest.clone(),
|
manifest.clone(),
|
||||||
prompt_loader,
|
prompt_loader,
|
||||||
available_profiles,
|
available_profiles,
|
||||||
spawner_scope.clone(),
|
|
||||||
DelegationScope::from_config(&manifest.delegation_scope).unwrap(),
|
|
||||||
)
|
)
|
||||||
.with_internal_client(Box::new(ScriptedInternalClient {
|
.with_internal_client(Box::new(ScriptedInternalClient {
|
||||||
calls: calls.clone(),
|
calls: calls.clone(),
|
||||||
@@ -1161,7 +1150,7 @@ extract_threshold = 4000
|
|||||||
"task": "review immutable commit",
|
"task": "review immutable commit",
|
||||||
"scope": [{
|
"scope": [{
|
||||||
"target": workspace_root.clone(),
|
"target": workspace_root.clone(),
|
||||||
"permission": "write",
|
"permission": "read",
|
||||||
"recursive": true
|
"recursive": true
|
||||||
}]
|
}]
|
||||||
});
|
});
|
||||||
@@ -1188,16 +1177,30 @@ extract_threshold = 4000
|
|||||||
.await
|
.await
|
||||||
.expect("spawn project reviewer as Internal Worker");
|
.expect("spawn project reviewer as Internal Worker");
|
||||||
assert!(output.summary.contains("internal worker `reviewer-child`"));
|
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
|
let record = registry
|
||||||
.get_internal("reviewer-child")
|
.get_internal("reviewer-child")
|
||||||
.expect("Internal reviewer registry record");
|
.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!(
|
assert_eq!(
|
||||||
record.session.wait_until_idle().await,
|
record.session.wait_until_idle().await,
|
||||||
crate::internal_worker::InternalWorkerSessionStatus::Idle
|
crate::internal_worker::InternalWorkerSessionStatus::Idle
|
||||||
);
|
);
|
||||||
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
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));
|
assert!(observed_instruction_override.load(Ordering::SeqCst));
|
||||||
let completion = tokio::time::timeout(Duration::from_secs(1), parent_method_rx.recv())
|
let completion = tokio::time::timeout(Duration::from_secs(1), parent_method_rx.recv())
|
||||||
.await
|
.await
|
||||||
@@ -1295,7 +1298,11 @@ extract_threshold = 4000
|
|||||||
assert_eq!(calls.load(Ordering::SeqCst), 3);
|
assert_eq!(calls.load(Ordering::SeqCst), 3);
|
||||||
assert!(
|
assert!(
|
||||||
spawner_scope.snapshot().is_writable(&workspace_root),
|
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());
|
assert!(registry.get_internal("reviewer-child").is_some());
|
||||||
|
|
||||||
@@ -1315,7 +1322,7 @@ extract_threshold = 4000
|
|||||||
)
|
)
|
||||||
.await
|
.await
|
||||||
.unwrap();
|
.unwrap();
|
||||||
assert!(!spawner_scope.snapshot().is_writable(&workspace_root));
|
assert!(spawner_scope.snapshot().is_writable(&workspace_root));
|
||||||
drop(list);
|
drop(list);
|
||||||
drop(send);
|
drop(send);
|
||||||
drop(stop);
|
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]
|
#[test]
|
||||||
fn orchestration_delegation_allows_root_read_and_worktree_writes_not_root_writes() {
|
fn orchestration_delegation_allows_root_read_and_worktree_writes_not_root_writes() {
|
||||||
let tmp = TempDir::new().unwrap();
|
let tmp = TempDir::new().unwrap();
|
||||||
|
|||||||
@@ -49,7 +49,8 @@ use workdir::workspace::{
|
|||||||
WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse,
|
WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse,
|
||||||
WorkingDirectoryDiagnostic, WorkingDirectoryDiagnosticSeverity,
|
WorkingDirectoryDiagnostic, WorkingDirectoryDiagnosticSeverity,
|
||||||
WorkingDirectoryListResponse as BrowserWorkingDirectoryListResponse, WorkingDirectoryOccupancy,
|
WorkingDirectoryListResponse as BrowserWorkingDirectoryListResponse, WorkingDirectoryOccupancy,
|
||||||
WorkingDirectoryStatusKind, WorkingDirectorySummary,
|
WorkingDirectoryStatusKind, WorkingDirectorySummary, WorkspaceWorkdirSessionFence,
|
||||||
|
WorkspaceWorkdirSessionOperationRequest,
|
||||||
};
|
};
|
||||||
use worker::feature::builtin::{WorkerObservationSubject, WorkerObservationSubjectRef};
|
use worker::feature::builtin::{WorkerObservationSubject, WorkerObservationSubjectRef};
|
||||||
use worker_runtime::resource::{BackendResourceError, BackendResourceFetchRequest};
|
use worker_runtime::resource::{BackendResourceError, BackendResourceFetchRequest};
|
||||||
@@ -1523,6 +1524,10 @@ pub fn build_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),
|
||||||
@@ -5043,7 +5048,7 @@ async fn scoped_attach_current_worker_workdir(
|
|||||||
worker: worker.clone(),
|
worker: worker.clone(),
|
||||||
workdir_id: workdir_id.to_string(),
|
workdir_id: workdir_id.to_string(),
|
||||||
role: "attachment".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,
|
unlinked_at: None,
|
||||||
})?;
|
})?;
|
||||||
if let Err(error) = open_current_worker_workdir_session_locked(&api, &worker, &link).await {
|
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<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(())
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async fn scoped_execute_current_worker_workdir_operation(
|
async fn scoped_execute_current_worker_workdir_operation(
|
||||||
State(api): State<WorkspaceApi>,
|
State(api): State<WorkspaceApi>,
|
||||||
AxumPath(path): AxumPath<ScopedWorkspacePath>,
|
AxumPath(path): AxumPath<ScopedWorkspacePath>,
|
||||||
headers: HeaderMap,
|
headers: HeaderMap,
|
||||||
Json(operation): Json<WorkdirSessionOperation>,
|
Json(request): Json<WorkspaceWorkdirSessionOperationRequest>,
|
||||||
) -> ApiResult<Json<WorkdirSessionOperationResult>> {
|
) -> ApiResult<Json<WorkdirSessionOperationResult>> {
|
||||||
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 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 = current_worker_active_attachment(&api, &worker)?;
|
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 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
|
.await
|
||||||
.map_err(|error| Error::RuntimeOperationFailed {
|
.map_err(|error| Error::RuntimeOperationFailed {
|
||||||
runtime_id: worker.runtime_id.clone(),
|
runtime_id: worker.runtime_id.clone(),
|
||||||
@@ -16167,6 +16208,30 @@ 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;
|
||||||
|
|||||||
Reference in New Issue
Block a user