feat: delegate subworker access through workdir sessions

Cargo.lock

@@ -6067,6 +6067,7 @@ dependencies = [
  "config-source",
  "dotenv",
  "flow",
+ "fs-operation",
  "fs4",
  "futures",
  "futures-util",

crates/workdir/src/delegation.rs

@@ -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)
+        ));
+    }
+}

crates/workdir/src/http.rs

@@ -120,7 +120,10 @@ impl WorkdirTransportError {
             WorkdirError::UnknownCommand(_) => {
                 (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::InvalidPath(_)
             | WorkdirError::RelativePath(_)
@@ -169,7 +172,7 @@ mod client {
     use reqwest::{Client, StatusCode, Url};
 
     use super::*;
-    use crate::{Workdir, WorkdirSession};
+    use crate::{Workdir, WorkdirSession, WorkdirSessionHandle};
 
     /// Provides a fresh bearer token for each Runtime request. Backend
     /// implementations can mint short-lived capability tokens without making a
@@ -310,6 +313,21 @@ mod client {
             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> {
             match self.operate(WorkdirSessionOperation::Stat(request)).await? {
                 WorkdirSessionOperationResult::Stat(result) => Ok(result),

crates/workdir/src/lib.rs

@@ -5,6 +5,7 @@
 //! bound to one Worker. Tools consume sessions; they do not own Workdir
 //! materialization or cleanup.
 
+mod delegation;
 pub mod http;
 mod local;
 mod operation;
@@ -12,11 +13,14 @@ pub mod workspace;
 
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
-use std::sync::atomic::{AtomicBool, Ordering};
 
 use async_trait::async_trait;
 use serde::{Deserialize, Serialize};
 
+pub use delegation::{
+    ReadOnlyWorkdirSession, WorkdirDelegation, WorkdirDelegationPermission,
+    WorkdirDelegationRequest, WorkdirDelegationRule, delegation_capable_session,
+};
 pub use fs_operation::{
     ContentHash, EditRequest, EditResult, EntryKind, FsPath as WorkdirPath, GlobRequest,
     GlobResult, GrepOutputMode, GrepRequest, GrepResult, ListEntry, ListRequest, ListResult,
@@ -140,6 +144,30 @@ pub trait WorkdirSession: std::fmt::Debug + Send + Sync {
     fn workdir(&self) -> &Workdir;
     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 read(&self, request: ReadRequest) -> Result<ReadResult, 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>;
 
-/// 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)]
 pub enum WorkdirError {
+    #[error("Workdir operation denied: {0}")]
+    Denied(String),
+
+    #[error("Workdir session is closed")]
+    SessionClosed,
+
     #[error("Workdir session does not support {0:?}")]
     Unsupported(WorkdirSessionCapability),
 

crates/workdir/src/local.rs

@@ -29,8 +29,8 @@ use crate::{
     CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest,
     EditResult, GlobRequest, GlobResult, GrepRequest, GrepResult, ListRequest, ListResult,
     ReadRequest, ReadResult, StatRequest, StatResult, Workdir, WorkdirError, WorkdirPath,
-    WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability, WriteRequest,
-    WriteResult,
+    WorkdirSession, WorkdirSessionCapabilities, WorkdirSessionCapability, WorkdirSessionHandle,
+    WriteRequest, WriteResult,
 };
 #[cfg(test)]
 use crate::{EntryKind, WriteOutcome};
@@ -371,6 +371,10 @@ impl WorkdirSession for LocalWorkdirSession {
         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> {
         self.ensure_capability(WorkdirSessionCapability::Read)?;
         let logical = request.path.clone();

crates/workdir/src/workspace.rs

@@ -314,3 +314,17 @@ mod tests {
         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,
+}

crates/worker/Cargo.toml

@@ -33,6 +33,7 @@ config-source = { path = "../config-source" }
 include_dir = "0.7.4"
 fs4 = { workspace = true, features = ["sync"] }
 flow = { path = "../flow" }
+fs-operation = { workspace = true }
 libc = { workspace = true }
 schemars = { workspace = true }
 ticket = { workspace = true }

crates/worker/src/controller.rs

@@ -681,18 +681,20 @@ where
 {
     // Worker-immutable snapshots taken before the mutable worker borrow
     // below so the worker borrow doesn't conflict with reads on `worker`.
-    let scope_handle = worker.scope().clone();
     let feature_config = worker.manifest().feature.clone();
-    if feature_config.manage_workdir.enabled {
-        if let Some(existing) = worker.workdir_session().cloned() {
-            existing.close().await.map_err(std::io::Error::other)?;
-        }
+    if feature_config.manage_workdir.enabled && worker.workdir_session().is_none() {
         let workspace_client = worker.workspace_client_handle();
-        worker.bind_workdir_session(Some(
+        worker.bind_workdir_session(Some(workdir::delegation_capable_session(
             crate::feature::builtin::manage_workdir::WorkspaceAttachedWorkdirSession::handle(
                 workspace_client,
             ),
-        ));
+        )));
+    }
+    if feature_config.sub_worker.enabled
+        && let Some(existing) = worker.workdir_session().cloned()
+        && !existing.is_delegation_capable()
+    {
+        worker.bind_workdir_session(Some(workdir::delegation_capable_session(existing)));
     }
     let worker_workdir = worker.workdir_session().cloned();
     let local_filesystem = worker.local_working_directory().cloned();
@@ -844,6 +846,7 @@ where
     }
 
     let host_worker_observation_provider = worker.worker_observation_provider();
+    let source_workdir_session = worker.workdir_session().cloned();
     {
         let workspace_client = worker.workspace_client_handle();
         let engine = worker.engine_mut();
@@ -902,27 +905,18 @@ where
             Arc<dyn crate::feature::builtin::worker_observation::WorkerObservationProvider>,
         > = Vec::new();
 
-        // Worker-orchestration tools (SubWorkerSpawn + three control tools) share
-        // the Worker-scoped `SpawnedWorkerRegistry` (also consumed by the main
-        // loop's `WorkerEvent` handler). Expose them only behind the explicit
-        // profile feature and require delegation authority up front so enabling
-        // the surface cannot imply broad child scope by accident.
+        // Worker-orchestration tools derive child filesystem authority from the
+        // active provider-backed Workdir session. The tool remains registered
+        // without one so invocation fails deterministically until the parent
+        // attaches a Workdir.
         if feature_config.sub_worker.enabled {
             let spawner_cwd = local_filesystem
                 .as_ref()
                 .map(|local| local.cwd.clone())
-                .ok_or_else(|| {
-                    std::io::Error::new(
-                        std::io::ErrorKind::InvalidInput,
-                        "worker spawn tools require local Worker filesystem authority",
-                    )
-                })?;
-            let spawner_workspace_root = local_workspace_root.clone().ok_or_else(|| {
-                std::io::Error::new(
-                    std::io::ErrorKind::InvalidInput,
-                    "worker spawn tools require local Worker filesystem authority",
-                )
-            })?;
+                .unwrap_or_else(|| PathBuf::from("/"));
+            let spawner_workspace_root = local_workspace_root
+                .clone()
+                .unwrap_or_else(|| PathBuf::from("/"));
             engine.register_tool(sub_worker_spawn_tool(
                 spawner_name.clone(),
                 spawner_workspace_context,
@@ -930,9 +924,9 @@ where
                 runtime_base.clone(),
                 spawner_workspace_root,
                 spawner_cwd.clone(),
+                source_workdir_session,
                 spawned_registry.clone(),
                 spawner_manifest,
-                scope_handle,
                 prompts,
             ));
             observation_providers.push(Arc::new(

crates/worker/src/feature/builtin/manage_workdir.rs

@@ -16,7 +16,8 @@ use serde_json::json;
 use workdir::http::{WorkdirSessionOperation, WorkdirSessionOperationResult};
 use workdir::workspace::{
     WorkingDirectoryDetailResponse as WorkdirDetailResponse,
-    WorkingDirectoryListResponse as WorkdirListResponse,
+    WorkingDirectoryListResponse as WorkdirListResponse, WorkspaceWorkdirSessionFence,
+    WorkspaceWorkdirSessionOperationRequest,
 };
 use workdir::{
     CommandHandle, CommandOutput, CommandOutputRequest, CommandRequest, CommandStatus, EditRequest,
@@ -153,6 +154,7 @@ struct WorkspaceHttpWorkdirBackend {
 pub struct WorkspaceAttachedWorkdirSession {
     client: Arc<dyn WorkspaceClient>,
     workdir: Workdir,
+    expected_session_fence: Option<String>,
 }
 
 impl WorkspaceAttachedWorkdirSession {
@@ -160,6 +162,7 @@ impl WorkspaceAttachedWorkdirSession {
         Arc::new(Self {
             client,
             workdir: Workdir::new("workspace-attachment"),
+            expected_session_fence: None,
         })
     }
 
@@ -176,7 +179,11 @@ impl WorkspaceAttachedWorkdirSession {
                 "/api/w/{}/workers/self/workdir-session/operations",
                 encode_path_segment(workspace_id)
             ),
-            serde_json::to_string(&operation).map_err(|error| {
+            serde_json::to_string(&WorkspaceWorkdirSessionOperationRequest {
+                expected_session_fence: self.expected_session_fence.clone(),
+                operation,
+            })
+            .map_err(|error| {
                 WorkdirError::Transport(format!(
                     "failed to encode Workspace Workdir operation: {error}"
                 ))
@@ -224,6 +231,43 @@ impl WorkdirSession for WorkspaceAttachedWorkdirSession {
         WorkdirSessionCapabilities::ALL
     }
 
+    async fn capture_delegation_source(&self) -> Result<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> {
         match self.operate(WorkdirSessionOperation::Stat(request))? {
             WorkdirSessionOperationResult::Stat(result) => Ok(result),
@@ -1002,11 +1046,55 @@ mod tests {
         );
         let body: serde_json::Value =
             serde_json::from_str(requests[0].body.as_deref().unwrap()).unwrap();
-        assert_eq!(body["operation"], "stat");
+        assert_eq!(body["operation"]["operation"], "stat");
+        assert!(body.get("expected_session_fence").is_none());
         assert!(body.get("runtime_id").is_none());
         assert!(body.get("session_id").is_none());
     }
 
+    #[tokio::test]
+    async fn delegated_attached_session_carries_captured_fence_on_operations() {
+        let client = Arc::new(RecordingWorkspaceClient::new(vec![
+            response(json!({"value": "attachment-fence"})),
+            response(json!({
+                "operation": "stat",
+                "result": {"path": "visible.txt", "kind": "file", "size": 8}
+            })),
+        ]));
+        let parent = workdir::delegation_capable_session(WorkspaceAttachedWorkdirSession::handle(
+            client.clone(),
+        ));
+        let delegation = parent
+            .delegate(workdir::WorkdirDelegationRequest {
+                rules: vec![workdir::WorkdirDelegationRule {
+                    target: workdir::WorkdirPath::new("visible.txt").unwrap(),
+                    permission: workdir::WorkdirDelegationPermission::Read,
+                    recursive: false,
+                }],
+                cwd: workdir::WorkdirPath::new("visible.txt").unwrap(),
+            })
+            .await
+            .unwrap();
+        delegation
+            .scoped_session
+            .stat(StatRequest {
+                path: workdir::WorkdirPath::new("visible.txt").unwrap(),
+            })
+            .await
+            .unwrap();
+
+        let requests = client.requests();
+        assert_eq!(requests.len(), 2);
+        assert_eq!(
+            requests[0].path,
+            "/api/w/workspace%2Ftest/workers/self/workdir-session/fence"
+        );
+        let body: serde_json::Value =
+            serde_json::from_str(requests[1].body.as_deref().unwrap()).unwrap();
+        assert_eq!(body["expected_session_fence"], "attachment-fence");
+        assert_eq!(body["operation"]["operation"], "stat");
+    }
+
     #[test]
     fn invalid_or_extra_inputs_are_rejected_before_workspace_request() {
         let client = Arc::new(RecordingWorkspaceClient::new(Vec::new()));

crates/worker/src/spawn/registry.rs

@@ -19,6 +19,7 @@ use session_store::{
     WorkerMetadataStore, WorkerReclaimedChild, WorkerSpawnedChild, WorkerStoreError,
 };
 use tracing::warn;
+use workdir::WorkdirDelegation;
 
 use crate::internal_worker::InternalWorkerSessionHandle;
 use crate::runtime::dir::{RuntimeDir, SpawnedWorkerRecord};
@@ -28,6 +29,9 @@ use crate::runtime::worker_allocation;
 pub(crate) struct InternalSpawnedWorkerRecord {
     pub worker_name: String,
     pub scope_delegated: Vec<ScopeRule>,
+    pub workdir_delegation: Arc<WorkdirDelegation>,
+    #[cfg(test)]
+    pub installed_tools: Arc<[String]>,
     pub session: InternalWorkerSessionHandle,
     scope_reclaimed: Arc<AtomicBool>,
 }
@@ -36,11 +40,16 @@ impl InternalSpawnedWorkerRecord {
     pub(crate) fn new(
         worker_name: String,
         scope_delegated: Vec<ScopeRule>,
+        workdir_delegation: WorkdirDelegation,
+        #[cfg(test)] installed_tools: Vec<String>,
         session: InternalWorkerSessionHandle,
     ) -> Self {
         Self {
             worker_name,
             scope_delegated,
+            workdir_delegation: Arc::new(workdir_delegation),
+            #[cfg(test)]
+            installed_tools: installed_tools.into(),
             session,
             scope_reclaimed: Arc::new(AtomicBool::new(false)),
         }
@@ -247,6 +256,7 @@ impl SpawnedWorkerRegistry {
         if !record.claim_scope_reclaim() {
             return Ok(false);
         }
+        record.workdir_delegation.release();
         let result = if let Some(parent_scope) = &self.parent_scope {
             parent_scope
                 .update(|current| current.with_removed_deny_rules(delegated_write_rules(record)))

crates/worker/src/spawn/tool.rs

@@ -10,16 +10,21 @@ use std::sync::Arc;
 
 use arc_swap::ArcSwap;
 use async_trait::async_trait;
+use fs_operation::FsPath;
 use llm_engine::tool::{Tool, ToolDefinition, ToolError, ToolMeta, ToolOutput};
 use manifest::{
-    CompactionConfigPartial, DelegationScope, EngineManifestConfig, FileUploadLimitsPartial,
-    Permission, PermissionConfigPartial, ProfileDiscovery, ProfileError, ProfileRegistry,
-    ProfileRegistrySource, ProfileResolveOptions, ProfileResolver, ProfileSelector, Scope,
-    ScopeConfig, ScopeRule, SessionConfigPartial, SharedScope, ToolOutputLimitsPartial,
-    WorkerManifest, WorkerManifestConfig, WorkerMetaConfig,
+    CompactionConfigPartial, EngineManifestConfig, FileUploadLimitsPartial, Permission,
+    PermissionConfigPartial, ProfileDiscovery, ProfileError, ProfileRegistry,
+    ProfileRegistrySource, ProfileResolveOptions, ProfileResolver, ProfileSelector, ScopeConfig,
+    ScopeRule, SessionConfigPartial, ToolOutputLimitsPartial, WorkerManifest, WorkerManifestConfig,
+    WorkerMetaConfig,
 };
 use serde::Deserialize;
 use tokio::sync::mpsc;
+use workdir::{
+    WorkdirDelegationPermission, WorkdirDelegationRequest, WorkdirDelegationRule,
+    WorkdirSessionHandle,
+};
 
 use crate::PromptCatalogSource;
 use crate::controller::register_worker_tools;
@@ -270,6 +275,8 @@ pub struct SubWorkerSpawnTool {
     /// Directory the spawned SubWorker's tools should use when the LLM did not
     /// override it. Defaults to the spawner's cwd.
     spawner_cwd: PathBuf,
+    /// Active provider-backed Workdir session from which child leases are captured.
+    source_workdir_session: Option<WorkdirSessionHandle>,
     /// Parent-owned in-memory registry shared by the five SubWorker tools.
     registry: Arc<SpawnedWorkerRegistry>,
     /// Spawner's resolved Manifest. `profile = "inherit"` derives the
@@ -279,18 +286,6 @@ pub struct SubWorkerSpawnTool {
     prompt_loader: PromptCatalogSource,
     /// Compact selector list shared by tool description and diagnostics.
     available_profiles: AvailableProfiles,
-    /// Spawner's runtime scope. After a successful spawn, the
-    /// `Permission::Write` rules in the delegated scope are revoked
-    /// from the spawner's in-memory view (a `deny(Write, target)` is
-    /// pushed on top, downgrading the spawner's effective access on
-    /// those paths to `Read`). Mirrors the worker-allocation's
-    /// `effective_write` semantics: Write is the only permission
-    /// tracked across Workers, so revocation only touches Write.
-    spawner_scope: SharedScope,
-    /// Filesystem scope this Worker is allowed to subdelegate to children.
-    /// This is intentionally separate from `spawner_scope`, which authorizes
-    /// the current Worker's own direct tools.
-    delegation_scope: DelegationScope,
     internal_client_override: Option<Box<dyn llm_engine::llm_client::LlmClient>>,
 }
 
@@ -308,12 +303,11 @@ impl SubWorkerSpawnTool {
         runtime_base: PathBuf,
         workspace_root: PathBuf,
         spawner_cwd: PathBuf,
+        source_workdir_session: Option<WorkdirSessionHandle>,
         registry: Arc<SpawnedWorkerRegistry>,
         spawner_manifest: WorkerManifest,
         prompt_loader: PromptCatalogSource,
         available_profiles: AvailableProfiles,
-        spawner_scope: SharedScope,
-        delegation_scope: DelegationScope,
     ) -> Self {
         Self {
             spawner_name,
@@ -322,12 +316,11 @@ impl SubWorkerSpawnTool {
             runtime_base,
             workspace_root,
             spawner_cwd,
+            source_workdir_session,
             registry,
             spawner_manifest,
             prompt_loader,
             available_profiles,
-            spawner_scope,
-            delegation_scope,
             internal_client_override: None,
         }
     }
@@ -386,8 +379,16 @@ impl Tool for SubWorkerSpawnTool {
             .map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
 
         let scope_allow = parse_scope(&input.scope)?;
-        self.validate_delegation_scope(&scope_allow)?;
-        let child_cwd = validate_spawn_cwd(input.cwd.as_deref(), &scope_allow, &self.spawner_cwd)?;
+        let source_workdir_session =
+            require_active_workdir_session(self.source_workdir_session.as_ref())?;
+        let delegation_request =
+            self.workdir_delegation_request(input.cwd.as_deref(), &scope_allow)?;
+        let workdir_delegation = source_workdir_session
+            .delegate(delegation_request)
+            .await
+            .map_err(|error| {
+                ToolError::InvalidArgument(format!("delegate Workdir session: {error}"))
+            })?;
 
         let spawn_selector =
             parse_spawn_profile_selector(input.profile.as_deref()).map_err(|msg| {
@@ -413,11 +414,14 @@ impl Tool for SubWorkerSpawnTool {
             allow: scope_allow.clone(),
             deny: Vec::new(),
         };
-        let child_manifest =
+        let mut child_manifest =
             WorkerManifest::try_from(WorkerManifestConfig::builtin_defaults().merge(child_config))
                 .map_err(|error| {
                     ToolError::ExecutionFailed(format!("resolve child manifest: {error}"))
                 })?;
+        // Delegated children stay bound to their scoped session and cannot use
+        // Workspace attachment tools to replace it with parent-level authority.
+        child_manifest.feature.manage_workdir.enabled = false;
         let reviewer_capability = input.review.as_ref().map(|review| {
             (
                 review.ticket_id.clone(),
@@ -458,8 +462,7 @@ impl Tool for SubWorkerSpawnTool {
                 self.workspace_context.clone()
             };
         let store = EphemeralSessionStore::default();
-        let filesystem_authority =
-            WorkerFilesystemAuthority::local(self.workspace_root.clone(), child_cwd.clone());
+        let filesystem_authority = WorkerFilesystemAuthority::None;
         let mut child = Worker::<Box<dyn llm_engine::llm_client::LlmClient>, EphemeralSessionStore>::from_internal_manifest_with_context(
             child_manifest,
             store.clone(),
@@ -472,6 +475,7 @@ impl Tool for SubWorkerSpawnTool {
         )
         .await
         .map_err(|error| ToolError::ExecutionFailed(format!("build Internal Worker: {error}")))?;
+        child.bind_workdir_session(Some(workdir_delegation.scoped_session.clone()));
         let child_scope = child.scope().clone();
         let child_registry = SpawnedWorkerRegistry::new_internal(input.name.clone(), child_scope);
         register_worker_tools(
@@ -488,23 +492,14 @@ impl Tool for SubWorkerSpawnTool {
         .map_err(|error| {
             ToolError::ExecutionFailed(format!("install Internal Worker features: {error}"))
         })?;
-        // Transfer delegated Write authority before the child accepts its first turn. This closes
-        // the parallel-tool window where parent and child could otherwise both write the same path.
-        // The machine-wide allocation remains owned by the parent Worker; no fake child PID/socket
-        // identity is introduced.
-        let revoke_write: Vec<ScopeRule> = scope_allow
-            .iter()
-            .filter(|rule| rule.permission == Permission::Write)
-            .cloned()
+        #[cfg(test)]
+        let installed_tools = child
+            .engine()
+            .tool_server_handle()
+            .tool_definitions_sorted()
+            .into_iter()
+            .map(|definition| definition.name)
             .collect();
-        if !revoke_write.is_empty() {
-            self.spawner_scope
-                .update(|current| current.with_added_deny_rules(revoke_write.clone()))
-                .map_err(|error| {
-                    ToolError::ExecutionFailed(format!("revoke spawner scope: {error}"))
-                })?;
-        }
-
         let child_name = input.name.clone();
         let registry = Arc::downgrade(&self.registry);
         let parent_notifications = self.parent_notifications.clone();
@@ -530,19 +525,9 @@ impl Tool for SubWorkerSpawnTool {
             })),
         )
         .await;
-        let session = match session_result {
-            Ok(session) => session,
-            Err(error) => {
-                if !revoke_write.is_empty() {
-                    let _ = self
-                        .spawner_scope
-                        .update(|current| current.with_removed_deny_rules(revoke_write.clone()));
-                }
-                return Err(ToolError::ExecutionFailed(format!(
-                    "prepare Internal Worker session: {error}"
-                )));
-            }
-        };
+        let session = session_result.map_err(|error| {
+            ToolError::ExecutionFailed(format!("prepare Internal Worker session: {error}"))
+        })?;
 
         if let Some((ticket_id, capability_token)) = &reviewer_capability {
             let workspace_id = self.workspace_context.workspace_id().ok_or_else(|| {
@@ -605,15 +590,13 @@ impl Tool for SubWorkerSpawnTool {
         let record = crate::spawn::registry::InternalSpawnedWorkerRecord::new(
             input.name.clone(),
             scope_allow,
+            workdir_delegation,
+            #[cfg(test)]
+            installed_tools,
             session.clone(),
         );
         if let Err(error) = name_reservation.commit(record) {
             let _ = session.stop().await;
-            if !revoke_write.is_empty() {
-                let _ = self
-                    .spawner_scope
-                    .update(|current| current.with_removed_deny_rules(revoke_write));
-            }
             return Err(ToolError::ExecutionFailed(format!(
                 "register Internal Worker session: {error}"
             )));
@@ -636,27 +619,73 @@ impl Tool for SubWorkerSpawnTool {
 }
 
 impl SubWorkerSpawnTool {
-    fn validate_delegation_scope(&self, scope_allow: &[ScopeRule]) -> Result<(), ToolError> {
-        if self.delegation_scope.is_empty() && !scope_allow.is_empty() {
-            return Err(ToolError::InvalidArgument(
-                "SubWorkerSpawn requires delegation authority, but this Worker has no delegation scope grant; direct filesystem scope only authorizes this Worker's own tools".into(),
-            ));
+    fn workdir_delegation_request(
+        &self,
+        cwd: Option<&Path>,
+        scope_allow: &[ScopeRule],
+    ) -> Result<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 {
-            let allowed = self
-                .delegation_scope
-                .allows_rule(rule)
-                .map_err(|error| ToolError::InvalidArgument(error.to_string()))?;
-            if !allowed {
-                return Err(ToolError::InvalidArgument(format!(
-                    "requested child scope {} {:?} is outside this Worker's delegation scope grant",
-                    rule.target.display(),
-                    rule.permission
-                )));
-            }
-        }
-        Ok(())
+        Ok(WorkdirDelegationRequest {
+            rules,
+            cwd: self.logical_workdir_path(cwd)?,
+        })
     }
+
+    fn logical_workdir_path(&self, path: &Path) -> Result<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> {
@@ -681,63 +710,6 @@ fn parse_scope(rules: &[ScopeRuleInput]) -> Result<Vec<ScopeRule>, ToolError> {
         .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
 /// Worker runtime process via the hidden `--spawn-config-json` flag.
 /// `WorkerManifestConfig`'s `Serialize` impl is the single source of truth for the
@@ -944,9 +916,9 @@ pub(crate) fn sub_worker_spawn_tool(
     runtime_base: PathBuf,
     workspace_root: PathBuf,
     spawner_cwd: PathBuf,
+    source_workdir_session: Option<WorkdirSessionHandle>,
     registry: Arc<SpawnedWorkerRegistry>,
     spawner_manifest: WorkerManifest,
-    spawner_scope: SharedScope,
     prompts: Arc<ArcSwap<PromptCatalog>>,
 ) -> ToolDefinition {
     sub_worker_spawn_tool_impl(
@@ -956,9 +928,9 @@ pub(crate) fn sub_worker_spawn_tool(
         runtime_base,
         workspace_root,
         spawner_cwd,
+        source_workdir_session,
         registry,
         spawner_manifest,
-        spawner_scope,
         prompts,
     )
 }
@@ -970,9 +942,9 @@ fn sub_worker_spawn_tool_impl(
     runtime_base: PathBuf,
     workspace_root: PathBuf,
     spawner_cwd: PathBuf,
+    source_workdir_session: Option<WorkdirSessionHandle>,
     registry: Arc<SpawnedWorkerRegistry>,
     spawner_manifest: WorkerManifest,
-    spawner_scope: SharedScope,
     prompts: Arc<ArcSwap<PromptCatalog>>,
 ) -> ToolDefinition {
     Arc::new(move || {
@@ -1002,13 +974,11 @@ fn sub_worker_spawn_tool_impl(
             runtime_base.clone(),
             workspace_root.clone(),
             spawner_cwd.clone(),
+            source_workdir_session.clone(),
             registry.clone(),
             spawner_manifest.clone(),
             prompts.load_full().source(),
             available_profiles,
-            spawner_scope.clone(),
-            DelegationScope::from_config(&spawner_manifest.delegation_scope)
-                .expect("resolved Worker manifest has a valid delegation scope"),
         ));
         (meta, tool)
     })
@@ -1017,6 +987,7 @@ fn sub_worker_spawn_tool_impl(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use manifest::{DelegationScope, Scope, SharedScope};
     use std::pin::Pin;
     use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
     use std::time::Duration;
@@ -1035,6 +1006,16 @@ mod tests {
         WorkspaceClient, WorkspaceClientError, WorkspaceRequest, WorkspaceResponse,
     };
 
+    #[test]
+    fn missing_active_workdir_session_fails_deterministically() {
+        let error = require_active_workdir_session(None).unwrap_err();
+        assert!(matches!(
+            error,
+            ToolError::InvalidArgument(message)
+                if message.contains("requires an active Workdir session")
+        ));
+    }
+
     #[test]
     fn reviewer_handoff_requires_explicit_builtin_profile_and_read_only_scope() {
         let valid: SubWorkerSpawnInput = serde_json::from_value(serde_json::json!({
@@ -1132,6 +1113,15 @@ extract_threshold = 4000
         let fail_requests = Arc::new(AtomicBool::new(false));
         let prompt_loader = PromptCatalogSource::builtins_only();
         let (parent_method_tx, mut parent_method_rx) = mpsc::channel(8);
+        let source_workdir_session = workdir::delegation_capable_session(Arc::new(
+            workdir::LocalWorkdirSession::materialized_bound(
+                workdir::Workdir::new("test-workdir"),
+                workspace_root.clone(),
+                workspace_root.clone(),
+                spawner_scope.clone(),
+                workdir::WorkdirSessionCapabilities::ALL,
+            ),
+        ));
         let tool = SubWorkerSpawnTool::new(
             "parent".into(),
             workspace_context,
@@ -1139,12 +1129,11 @@ extract_threshold = 4000
             runtime.path().to_path_buf(),
             workspace_root.clone(),
             workspace_root.clone(),
+            Some(source_workdir_session),
             registry.clone(),
             manifest.clone(),
             prompt_loader,
             available_profiles,
-            spawner_scope.clone(),
-            DelegationScope::from_config(&manifest.delegation_scope).unwrap(),
         )
         .with_internal_client(Box::new(ScriptedInternalClient {
             calls: calls.clone(),
@@ -1161,7 +1150,7 @@ extract_threshold = 4000
             "task": "review immutable commit",
             "scope": [{
                 "target": workspace_root.clone(),
-                "permission": "write",
+                "permission": "read",
                 "recursive": true
             }]
         });
@@ -1188,16 +1177,30 @@ extract_threshold = 4000
             .await
             .expect("spawn project reviewer as Internal Worker");
         assert!(output.summary.contains("internal worker `reviewer-child`"));
-        assert!(!spawner_scope.snapshot().is_writable(&workspace_root));
+        assert!(spawner_scope.snapshot().is_writable(&workspace_root));
         let record = registry
             .get_internal("reviewer-child")
             .expect("Internal reviewer registry record");
+        assert!(record.installed_tools.iter().any(|name| name == "Read"));
+        for denied in ["Write", "Edit", "Bash"] {
+            assert!(
+                !record.installed_tools.iter().any(|name| name == denied),
+                "read-only child unexpectedly received {denied}: {:?}",
+                record.installed_tools
+            );
+        }
+        assert!(
+            !record
+                .installed_tools
+                .iter()
+                .any(|name| matches!(name.as_str(), "WorkdirAttachSelf" | "WorkdirDetachSelf"))
+        );
         assert_eq!(
             record.session.wait_until_idle().await,
             crate::internal_worker::InternalWorkerSessionStatus::Idle
         );
         assert_eq!(calls.load(Ordering::SeqCst), 1);
-        assert!(observed_parent_write_revoked.load(Ordering::SeqCst));
+        assert!(!observed_parent_write_revoked.load(Ordering::SeqCst));
         assert!(observed_instruction_override.load(Ordering::SeqCst));
         let completion = tokio::time::timeout(Duration::from_secs(1), parent_method_rx.recv())
             .await
@@ -1295,7 +1298,11 @@ extract_threshold = 4000
         assert_eq!(calls.load(Ordering::SeqCst), 3);
         assert!(
             spawner_scope.snapshot().is_writable(&workspace_root),
-            "Failed terminal child must automatically reclaim its delegated write scope"
+            "Failed terminal child must release its delegated Workdir session"
+        );
+        assert!(
+            !record.workdir_delegation.is_active(),
+            "failed child must revoke cloned scoped sessions"
         );
         assert!(registry.get_internal("reviewer-child").is_some());
 
@@ -1315,7 +1322,7 @@ extract_threshold = 4000
         )
         .await
         .unwrap();
-        assert!(!spawner_scope.snapshot().is_writable(&workspace_root));
+        assert!(spawner_scope.snapshot().is_writable(&workspace_root));
         drop(list);
         drop(send);
         drop(stop);
@@ -1343,45 +1350,6 @@ extract_threshold = 4000
         );
     }
 
-    #[test]
-    fn spawn_worker_validate_cwd_requires_absolute_existing_directory_in_child_scope() {
-        let root = TempDir::new().unwrap();
-        let child_cwd = root.path().join("child");
-        std::fs::create_dir(&child_cwd).unwrap();
-        let file_path = root.path().join("file.txt");
-        std::fs::write(&file_path, "not a dir").unwrap();
-        let outside = TempDir::new().unwrap();
-        let missing = root.path().join("missing");
-        let rules = vec![abs_rule(root.path(), Permission::Write)];
-
-        assert_eq!(
-            validate_spawn_cwd(None, &rules, root.path()).unwrap(),
-            root.path()
-        );
-        assert_eq!(
-            validate_spawn_cwd(Some(&child_cwd), &rules, root.path()).unwrap(),
-            std::fs::canonicalize(&child_cwd).unwrap()
-        );
-
-        for (cwd, expected) in [
-            (Path::new("relative"), "must be absolute"),
-            (missing.as_path(), "does not exist"),
-            (file_path.as_path(), "must be a directory"),
-            (
-                outside.path(),
-                "outside the child's delegated readable scope",
-            ),
-        ] {
-            let err = validate_spawn_cwd(Some(cwd), &rules, root.path()).unwrap_err();
-            match err {
-                ToolError::InvalidArgument(message) => {
-                    assert!(message.contains(expected), "{message}")
-                }
-                other => panic!("expected InvalidArgument, got {other:?}"),
-            }
-        }
-    }
-
     #[test]
     fn orchestration_delegation_allows_root_read_and_worktree_writes_not_root_writes() {
         let tmp = TempDir::new().unwrap();

crates/workspace-server/src/server.rs

@@ -49,7 +49,8 @@ use workdir::workspace::{
     WorkingDirectoryDetailResponse as BrowserWorkingDirectoryDetailResponse,
     WorkingDirectoryDiagnostic, WorkingDirectoryDiagnosticSeverity,
     WorkingDirectoryListResponse as BrowserWorkingDirectoryListResponse, WorkingDirectoryOccupancy,
-    WorkingDirectoryStatusKind, WorkingDirectorySummary,
+    WorkingDirectoryStatusKind, WorkingDirectorySummary, WorkspaceWorkdirSessionFence,
+    WorkspaceWorkdirSessionOperationRequest,
 };
 use worker::feature::builtin::{WorkerObservationSubject, WorkerObservationSubjectRef};
 use worker_runtime::resource::{BackendResourceError, BackendResourceFetchRequest};
@@ -1523,6 +1524,10 @@ pub fn build_router(api: WorkspaceApi) -> Router {
             post(scoped_attach_current_worker_workdir)
                 .delete(scoped_detach_current_worker_workdir),
         )
+        .route(
+            "/api/w/{workspace_id}/workers/self/workdir-session/fence",
+            get(scoped_current_worker_workdir_session_fence),
+        )
         .route(
             "/api/w/{workspace_id}/workers/self/workdir-session/operations",
             post(scoped_execute_current_worker_workdir_operation),
@@ -5043,7 +5048,7 @@ async fn scoped_attach_current_worker_workdir(
         worker: worker.clone(),
         workdir_id: workdir_id.to_string(),
         role: "attachment".to_string(),
-        linked_at: Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true),
+        linked_at: Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true),
         unlinked_at: None,
     })?;
     if let Err(error) = open_current_worker_workdir_session_locked(&api, &worker, &link).await {
@@ -5086,19 +5091,55 @@ async fn scoped_detach_current_worker_workdir(
     }))
 }
 
+async fn scoped_current_worker_workdir_session_fence(
+    State(api): State<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(
     State(api): State<WorkspaceApi>,
     AxumPath(path): AxumPath<ScopedWorkspacePath>,
     headers: HeaderMap,
-    Json(operation): Json<WorkdirSessionOperation>,
+    Json(request): Json<WorkspaceWorkdirSessionOperationRequest>,
 ) -> ApiResult<Json<WorkdirSessionOperationResult>> {
     validate_workspace_scope(&api, &path.workspace_id)?;
     let worker = current_worker_identity(&api, &path.workspace_id, &headers)?;
     let session_lock = current_worker_session_lock(&api, &worker);
     let _session_guard = session_lock.lock().await;
     let link = current_worker_active_attachment(&api, &worker)?;
+    validate_current_worker_workdir_session_fence(
+        &link,
+        request.expected_session_fence.as_deref(),
+    )?;
     let session = open_current_worker_workdir_session_locked(&api, &worker, &link).await?;
-    let result = execute_workdir_session_operation(&session, operation)
+    let result = execute_workdir_session_operation(&session, request.operation)
         .await
         .map_err(|error| Error::RuntimeOperationFailed {
             runtime_id: worker.runtime_id.clone(),
@@ -16167,6 +16208,30 @@ mod tests {
         assert_eq!(response.status(), StatusCode::BAD_REQUEST);
     }
 
+    #[test]
+    fn delegated_workdir_session_fence_rejects_reattached_link() {
+        let first = WorkerWorkdirLinkRecord {
+            workspace_id: "workspace-a".to_string(),
+            worker: workdir::workspace::RuntimeWorkerRef::new("runtime-a", "worker-a"),
+            workdir_id: "workdir-a".to_string(),
+            role: "primary".to_string(),
+            linked_at: "2026-01-01T00:00:00Z".to_string(),
+            unlinked_at: None,
+        };
+        let expected = current_worker_workdir_session_fence(&first);
+        assert!(validate_current_worker_workdir_session_fence(&first, None).is_ok());
+        assert!(validate_current_worker_workdir_session_fence(&first, Some(&expected)).is_ok());
+
+        let reattached = WorkerWorkdirLinkRecord {
+            linked_at: "2026-01-01T00:00:01Z".to_string(),
+            ..first
+        };
+        assert!(matches!(
+            validate_current_worker_workdir_session_fence(&reattached, Some(&expected)),
+            Err(Error::WorkdirAttachmentConflict(_))
+        ));
+    }
+
     #[tokio::test]
     async fn backend_workdir_session_proxy_executes_typed_operations() {
         use manifest::Scope;