Merge develop into T-731 source and align scoped authorization logging fixture

crates/workdir/src/checkout_search_tests.rs

@@ -212,17 +212,17 @@ async fn checkout_search_composes_actual_nested_wrappers_and_output_root() {
     };
     assert_eq!(result.entries[0].path, path("keep/yes.txt"));
     // A second actual scoped wrapper has its own cwd and nonrecursive layer.
-    // Logical rules avoid requesting resolved authorization from a wrapper
-    // that conservatively does not expose host-side resolution authority.
+    // Default resolved authorization must survive every wrapper.
     std::fs::create_dir(root.path().join("child/keep/deep")).unwrap();
     std::fs::write(
         root.path().join("child/keep/deep/secret"),
         "needle hidden nested",
     )
     .unwrap();
-    let mut own_rule = rule("keep", false);
-    own_rule.symlink_policy = manifest::SymlinkPolicy::Logical;
-    let second = stacked.scope(scope(vec![own_rule], "keep")).await.unwrap();
+    let second = stacked
+        .scope(scope(vec![rule("keep", false)], "keep"))
+        .await
+        .unwrap();
     let result = second
         .tool_session()
         .grep(grep(GrepOutputMode::Content))

crates/workdir/src/local.rs

@@ -1830,6 +1830,9 @@ impl WorkdirSession for LocalWorkdirSession {
 
     async fn start_command(&self, request: CommandRequest) -> Result<CommandHandle, WorkdirError> {
         self.ensure_capability(WorkdirSessionCapability::Command)?;
+        // Command admission, resource capture and registration must be atomic
+        // with terminal close, which drains commands and releases resources.
+        let _admission = self.inner.close_lock.lock().await;
         self.ensure_open()?;
         if let Some(spill_dir) = request.spill_dir.as_deref()
             && !self
@@ -1854,6 +1857,15 @@ impl WorkdirSession for LocalWorkdirSession {
         let telemetry = self.inner.command_telemetry.clone();
         let command_environment = self.inner.command_environment.clone();
         let (cancel, cancel_rx) = watch::channel(false);
+        // A session drop only schedules command cancellation. Keep its Runtime
+        // resources (including cleanup exclusion) alive until the task actually
+        // finishes or its future is dropped, not just until the handle disappears.
+        let resources = self
+            .inner
+            .resources
+            .lock()
+            .unwrap_or_else(|error| error.into_inner())
+            .clone();
         let task = tokio::spawn(async move {
             let output = run_command(
                 cwd,
@@ -1862,6 +1874,7 @@ impl WorkdirSession for LocalWorkdirSession {
                 telemetry,
                 command_environment,
                 cancel_rx,
+                resources,
             )
             .await;
             let _ = completion_tx.send(true);
@@ -2194,6 +2207,7 @@ async fn run_command(
     telemetry: CommandTelemetry,
     command_environment: BTreeMap<String, String>,
     mut cancel: watch::Receiver<bool>,
+    resources: Vec<Arc<dyn WorkdirSessionResource>>,
 ) -> Result<CommandOutput, WorkdirError> {
     let stdout = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?;
     let stderr = tempfile::NamedTempFile::new().map_err(|error| WorkdirError::io(&cwd, error))?;
@@ -2313,7 +2327,7 @@ async fn run_command(
             let stdout_path = stdout_path.to_path_buf();
             let stderr_path = stderr_path.to_path_buf();
             Some(
-                tokio::task::spawn_blocking(move || {
+                spawn_resource_owned_blocking(resources.clone(), move || {
                     persist_command_output(&stdout_path, &stderr_path, &spill_dir)
                 })
                 .await
@@ -2335,6 +2349,18 @@ async fn run_command(
     })
 }
 
+// Blocking spills cannot be aborted with their async waiter. Transfer resource
+// ownership into the actual filesystem task so cleanup waits for its completion.
+fn spawn_resource_owned_blocking<T: Send + 'static>(
+    resources: Vec<Arc<dyn WorkdirSessionResource>>,
+    effect: impl FnOnce() -> T + Send + 'static,
+) -> JoinHandle<T> {
+    tokio::task::spawn_blocking(move || {
+        let _resources = resources;
+        effect()
+    })
+}
+
 #[derive(Debug, Default)]
 struct CommandOutputDecoder {
     read_offset: u64,
@@ -4180,6 +4206,152 @@ mod tests {
         assert_eq!(terminal, Some((handle.0, CommandStatus::TimedOut, None)));
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn command_admission_serializes_with_close_before_capturing_resources() {
+        let dir = TempDir::new().unwrap();
+        let session = Arc::new(make_fs(&dir));
+        let held = session.inner.close_lock.lock().await;
+        let (closing, close_entered) = tokio::sync::oneshot::channel();
+        let close_session = session.clone();
+        let close = tokio::spawn(async move {
+            closing.send(()).unwrap();
+            WorkdirSession::close(close_session.as_ref()).await
+        });
+        close_entered.await.unwrap();
+        let (starting, start_entered) = tokio::sync::oneshot::channel();
+        let start_session = session.clone();
+        let start = tokio::spawn(async move {
+            starting.send(()).unwrap();
+            WorkdirSession::start_command(
+                start_session.as_ref(),
+                CommandRequest {
+                    command: "true".into(),
+                    timeout_secs: 5,
+                    output_limit: 1024,
+                    cwd: WorkdirPath::root(),
+                    spill_dir: None,
+                    tool_call_id: None,
+                },
+            )
+            .await
+        });
+        start_entered.await.unwrap();
+        assert!(
+            !start.is_finished(),
+            "command started outside terminal-close admission"
+        );
+        drop(held);
+        tokio::time::timeout(Duration::from_secs(5), close)
+            .await
+            .unwrap()
+            .unwrap()
+            .unwrap();
+        assert!(matches!(
+            tokio::time::timeout(Duration::from_secs(5), start)
+                .await
+                .unwrap()
+                .unwrap(),
+            Err(WorkdirError::Unavailable(message)) if message.contains("is closed")
+        ));
+        assert!(
+            session.command_snapshot().is_empty(),
+            "closed admission launched a command"
+        );
+    }
+
+    #[tokio::test]
+    async fn blocking_output_spill_retains_resources_after_waiter_is_cancelled() {
+        #[derive(Debug)]
+        struct Resource(Option<tokio::sync::oneshot::Sender<()>>);
+        impl Drop for Resource {
+            fn drop(&mut self) {
+                let _ = self.0.take().unwrap().send(());
+            }
+        }
+        let (released, mut release) = tokio::sync::oneshot::channel();
+        let resources: Vec<Arc<dyn WorkdirSessionResource>> =
+            vec![Arc::new(Resource(Some(released)))];
+        let (entered, started) = tokio::sync::oneshot::channel();
+        let (finish, wait) = std::sync::mpsc::channel();
+        // Exercise the resource-transfer policy at the same blocking boundary
+        // used by output persistence, without filesystem timing assumptions.
+        let waiter = tokio::spawn(async move {
+            spawn_resource_owned_blocking(resources, move || {
+                entered.send(()).unwrap();
+                wait.recv_timeout(Duration::from_secs(5)).unwrap();
+            })
+            .await
+            .unwrap();
+        });
+        tokio::time::timeout(Duration::from_secs(5), started)
+            .await
+            .unwrap()
+            .unwrap();
+        waiter.abort();
+        assert!(waiter.await.unwrap_err().is_cancelled());
+        assert!(
+            matches!(
+                release.try_recv(),
+                Err(tokio::sync::oneshot::error::TryRecvError::Empty)
+            ),
+            "blocking spill lost its resources when the waiter was cancelled"
+        );
+        finish.send(()).unwrap();
+        tokio::time::timeout(Duration::from_secs(5), release)
+            .await
+            .unwrap()
+            .unwrap();
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn session_drop_retains_resources_until_command_task_cancellation_completes() {
+        #[derive(Debug)]
+        struct Resource(Option<tokio::sync::oneshot::Sender<()>>);
+        impl Drop for Resource {
+            fn drop(&mut self) {
+                let _ = self.0.take().unwrap().send(());
+            }
+        }
+        let dir = TempDir::new().unwrap();
+        let (released, mut release) = tokio::sync::oneshot::channel();
+        let session = LocalWorkdirSession::materialized_bound_with_environment(
+            Workdir::new("command-resource-session"),
+            dir.path().to_path_buf(),
+            dir.path().to_path_buf(),
+            SharedScope::new(Scope::writable(dir.path()).unwrap()),
+            WorkdirSessionCapabilities::ALL,
+            BTreeMap::new(),
+            vec![Arc::new(Resource(Some(released)))],
+        );
+        WorkdirSession::start_command(
+            &session,
+            CommandRequest {
+                command: "true".into(),
+                timeout_secs: 5,
+                output_limit: 1024,
+                cwd: WorkdirPath::root(),
+                spill_dir: None,
+                tool_call_id: None,
+            },
+        )
+        .await
+        .unwrap();
+        // The sole executor has not polled the command task, so dropping the
+        // session schedules its abort but cannot yet complete cancellation.
+        drop(session);
+        assert!(
+            matches!(
+                release.try_recv(),
+                Err(tokio::sync::oneshot::error::TryRecvError::Empty)
+            ),
+            "resources released before the aborted command future was dropped"
+        );
+        tokio::time::timeout(Duration::from_secs(5), release)
+            .await
+            .unwrap()
+            .unwrap();
+    }
+
     #[tokio::test]
     async fn closing_session_releases_runtime_resources() {
         #[derive(Debug)]

crates/workdir/src/scope.rs

@@ -1030,6 +1030,51 @@ impl WorkdirSession for ScopedWorkdirSession {
         self.capabilities
     }
 
+    async fn authorize_scope_path(
+        &self,
+        mut request: WorkdirScopeAuthorizationRequest,
+    ) -> Result<(), WorkdirError> {
+        let _scope_guard = self.scope_lock.lock().await;
+        // Like filesystem operations, requests use this session's coordinates.
+        // Keep every wrapper's attenuation before asking the provider to resolve
+        // symlinks; forwarding the caller's rules alone would discard our scope.
+        for rule in &mut request.rules {
+            rule.target = self.resolve_path(&rule.target)?;
+        }
+        self.validate_scope(&request.rules, false)?;
+        request.path = self
+            .resolve_operation_path(&request.path, request.permission)
+            .await?;
+        match request.permission {
+            WorkdirToolScopePermission::Read => {
+                self.ensure_read(&request.path, WorkdirSessionCapability::Read)?;
+            }
+            WorkdirToolScopePermission::Write => {
+                self.ensure_write(&request.path, WorkdirSessionCapability::Write)?;
+            }
+        }
+        self.source.authorize_scope_path(request).await?;
+        self.ensure_active()
+    }
+
+    async fn scope_rules_overlap(
+        &self,
+        mut request: WorkdirScopeOverlapRequest,
+    ) -> Result<bool, WorkdirError> {
+        let _scope_guard = self.scope_lock.lock().await;
+        // Comparing write rules is a read-only query, not a write authorization.
+        // In particular it must work while those paths are leased to children.
+        for rule in [&mut request.left, &mut request.right] {
+            rule.target = self
+                .resolve_operation_path(&rule.target, WorkdirToolScopePermission::Read)
+                .await?;
+            self.ensure_read(&rule.target, WorkdirSessionCapability::Read)?;
+        }
+        let overlaps = self.source.scope_rules_overlap(request).await?;
+        self.ensure_active()?;
+        Ok(overlaps)
+    }
+
     async fn checkout_search(
         &self,
         mut request: crate::CheckoutSearchRequest,
@@ -1769,6 +1814,8 @@ mod tests {
     use super::*;
     use crate::LocalWorkdirSession;
 
+    include!("scope_delegation_tests.rs");
+
     fn fs_path(path: &str) -> FsPath {
         FsPath::new(path).unwrap()
     }

crates/workdir/src/scope_delegation_tests.rs

@@ -0,0 +1,334 @@
+// Included in scope::tests; exercise real provider resolution through stacked brokers.
+
+fn filesystem_scope(path: &str, permission: WorkdirToolScopePermission) -> WorkdirToolScope {
+    let mut scope = request(path, permission);
+    scope.command = false;
+    scope
+}
+
+fn authorization(
+    path: &str,
+    permission: WorkdirToolScopePermission,
+) -> WorkdirScopeAuthorizationRequest {
+    WorkdirScopeAuthorizationRequest {
+        rules: filesystem_scope("", permission).rules,
+        path: fs_path(path),
+        permission,
+    }
+}
+
+#[tokio::test]
+async fn stacked_brokers_delegate_all_filesystem_provider_kinds() {
+    use crate::{BoundedReadLimits, ExternalWorkdirRoot};
+    for kind in [
+        "repository",
+        "read-only-wrapper",
+        "external-read",
+        "external-write",
+        "external-read-command",
+        "external-all",
+    ] {
+        let root = TempDir::new().unwrap();
+        fs::create_dir(root.path().join("child")).unwrap();
+        fs::create_dir(root.path().join("other")).unwrap();
+        fs::write(root.path().join("child/input"), "visible").unwrap();
+        fs::write(root.path().join("other/secret"), "hidden").unwrap();
+        #[cfg(unix)]
+        std::os::unix::fs::symlink("../other", root.path().join("child/escape")).unwrap();
+        let capabilities = match kind {
+            "read-only-wrapper" | "external-read" => WorkdirSessionCapabilities::READ_ONLY,
+            "external-write" => WorkdirSessionCapabilities::READ_WRITE,
+            "external-read-command" => WorkdirSessionCapabilities::READ_ONLY
+                .union(WorkdirSessionCapabilities::COMMAND_ONLY),
+            _ => WorkdirSessionCapabilities::ALL,
+        };
+        let source: WorkdirSessionHandle = if kind.starts_with("external-") {
+            Arc::new(
+                LocalWorkdirSession::external_with_capabilities_pinned(
+                    Workdir::new(kind),
+                    ExternalWorkdirRoot::pin(root.path()).unwrap(),
+                    BoundedReadLimits::new(4096, 1024).unwrap(),
+                    capabilities,
+                )
+                .unwrap(),
+            )
+        } else {
+            let source: WorkdirSessionHandle = Arc::new(LocalWorkdirSession::new(
+                Scope::writable(root.path()).unwrap(),
+                root.path().to_path_buf(),
+            ));
+            if kind == "read-only-wrapper" {
+                Arc::new(ReadOnlyWorkdirSession::new(source))
+            } else {
+                source
+            }
+        };
+        // Backend capability wrapper, attachment routing, then the Worker broker.
+        let backend = WorkdirToolBroker::with_capabilities(source, capabilities);
+        let router = crate::WorkdirSessionRouter::new();
+        let routed = router
+            .attach(
+                crate::WorkdirAttachmentAlias::new("selected").unwrap(),
+                backend.tool_session(),
+            )
+            .unwrap();
+        let worker = WorkdirToolBroker::new(routed);
+        let child = worker
+            .scope(filesystem_scope("child", WorkdirToolScopePermission::Read))
+            .await
+            .unwrap();
+        assert_eq!(
+            child.read(read("input")).await.unwrap().bytes,
+            b"visible",
+            "{kind}"
+        );
+        assert!(child.write(write("denied", "no")).await.is_err(), "{kind}");
+        #[cfg(unix)]
+        assert!(child.read(read("escape/secret")).await.is_err(), "{kind}");
+        child.close().await.unwrap();
+        let writable = worker
+            .scope(filesystem_scope("child", WorkdirToolScopePermission::Write))
+            .await;
+        if !capabilities.supports(WorkdirSessionCapability::Write) {
+            assert!(writable.is_err(), "{kind}");
+            continue;
+        }
+        let writable = writable.unwrap();
+        writable.write(write("created", "child")).await.unwrap();
+        assert!(
+            worker.write(write("child/blocked", "no")).await.is_err(),
+            "{kind}"
+        );
+        assert!(
+            worker
+                .scope(filesystem_scope("child", WorkdirToolScopePermission::Write))
+                .await
+                .is_err(),
+            "{kind}"
+        );
+        // A non-overlapping write lease requires the second missing provider method.
+        let sibling = worker
+            .scope(filesystem_scope("other", WorkdirToolScopePermission::Write))
+            .await
+            .unwrap();
+        sibling.write(write("created", "sibling")).await.unwrap();
+        assert!(
+            !writable
+                .capabilities
+                .supports(WorkdirSessionCapability::Command)
+        );
+        writable.close().await.unwrap();
+        worker
+            .write(write("child/released", "parent"))
+            .await
+            .unwrap();
+        sibling.close().await.unwrap();
+        let command_child = worker
+            .scope(request("child", WorkdirToolScopePermission::Write))
+            .await;
+        if capabilities.supports(WorkdirSessionCapability::Command) {
+            let command_child = command_child.unwrap();
+            assert_eq!(
+                run_command(&command_child.tool_session(), "printf delegated", kind)
+                    .await
+                    .content,
+                "delegated"
+            );
+            command_child.close().await.unwrap();
+        } else {
+            assert!(command_child.is_err());
+        }
+    }
+}
+
+#[tokio::test]
+async fn stacked_authorization_preserves_ceiling_scope_and_revocation() {
+    let root = TempDir::new().unwrap();
+    fs::create_dir_all(root.path().join("allowed/deep")).unwrap();
+    let backend = session(root.path());
+    let read_only = WorkdirToolBroker::with_capabilities(
+        backend.tool_session(),
+        WorkdirSessionCapabilities::READ_ONLY,
+    );
+    read_only
+        .authorize_scope_path(authorization("allowed", WorkdirToolScopePermission::Read))
+        .await
+        .unwrap();
+    assert!(
+        read_only
+            .authorize_scope_path(authorization("allowed", WorkdirToolScopePermission::Write))
+            .await
+            .is_err()
+    );
+    let command_only = WorkdirToolBroker::with_capabilities(
+        backend.tool_session(),
+        WorkdirSessionCapabilities::COMMAND_ONLY,
+    );
+    assert!(
+        command_only
+            .authorize_scope_path(authorization("allowed", WorkdirToolScopePermission::Read))
+            .await
+            .is_err()
+    );
+    assert!(
+        command_only
+            .scope(filesystem_scope(
+                "allowed",
+                WorkdirToolScopePermission::Read
+            ))
+            .await
+            .is_err()
+    );
+
+    let mut limited = filesystem_scope("allowed", WorkdirToolScopePermission::Read);
+    limited.rules[0].recursive = false;
+    let parent = backend.scope(limited).await.unwrap();
+    let stacked = WorkdirToolBroker::new(parent.tool_session());
+    let mut valid = filesystem_scope("", WorkdirToolScopePermission::Read);
+    valid.rules[0].recursive = false;
+    let child = stacked.scope(valid.clone()).await.unwrap();
+    assert!(
+        stacked
+            .scope(filesystem_scope("", WorkdirToolScopePermission::Read))
+            .await
+            .is_err()
+    );
+    let mut logical = valid.clone();
+    logical.rules[0].symlink_policy = SymlinkPolicy::Logical;
+    assert!(stacked.scope(logical).await.is_err());
+    assert!(
+        stacked
+            .authorize_scope_path(authorization("deep/file", WorkdirToolScopePermission::Read))
+            .await
+            .is_err()
+    );
+    parent.close().await.unwrap();
+    assert!(stacked.scope(valid).await.is_err());
+    assert!(child.read(read("deep/file")).await.is_err());
+    assert!(
+        stacked
+            .scope_rules_overlap(WorkdirScopeOverlapRequest {
+                left: filesystem_scope("", WorkdirToolScopePermission::Read)
+                    .rules
+                    .remove(0),
+                right: filesystem_scope("", WorkdirToolScopePermission::Read)
+                    .rules
+                    .remove(0),
+            })
+            .await
+            .is_err()
+    );
+}
+
+#[tokio::test]
+async fn stacked_scope_rebases_rules_paths_and_overlap_once() {
+    let root = TempDir::new().unwrap();
+    fs::create_dir_all(root.path().join("base/inner/one")).unwrap();
+    fs::create_dir_all(root.path().join("base/inner/two")).unwrap();
+    fs::write(root.path().join("base/inner/one/input"), "nested").unwrap();
+    let backend = session(root.path());
+    let first = backend
+        .scope(filesystem_scope("base", WorkdirToolScopePermission::Write))
+        .await
+        .unwrap();
+    let worker = WorkdirToolBroker::new(first.tool_session());
+    let second = worker
+        .scope(filesystem_scope("inner", WorkdirToolScopePermission::Write))
+        .await
+        .unwrap();
+    let nested = WorkdirToolBroker::new(second.tool_session());
+    let leaf = nested
+        .scope(filesystem_scope("one", WorkdirToolScopePermission::Write))
+        .await
+        .unwrap();
+    assert_eq!(leaf.read(read("input")).await.unwrap().bytes, b"nested");
+    leaf.write(write("output", "yes")).await.unwrap();
+    assert_eq!(
+        fs::read(root.path().join("base/inner/one/output")).unwrap(),
+        b"yes"
+    );
+    assert!(nested.write(write("one/blocked", "no")).await.is_err());
+    let sibling = nested
+        .scope(filesystem_scope("two", WorkdirToolScopePermission::Write))
+        .await
+        .unwrap();
+    sibling.write(write("output", "sibling")).await.unwrap();
+    assert!(
+        nested
+            .scope(filesystem_scope("one", WorkdirToolScopePermission::Write))
+            .await
+            .is_err()
+    );
+    leaf.close().await.unwrap();
+    nested.write(write("one/released", "yes")).await.unwrap();
+    sibling.close().await.unwrap();
+    first.close().await.unwrap();
+    assert!(
+        nested
+            .authorize_scope_path(authorization("one", WorkdirToolScopePermission::Read))
+            .await
+            .is_err()
+    );
+}
+
+#[cfg(unix)]
+#[tokio::test]
+async fn stacked_scope_keeps_resolved_symlink_and_lease_checks() {
+    use std::os::unix::fs::symlink;
+    let root = TempDir::new().unwrap();
+    fs::create_dir_all(root.path().join("base/target")).unwrap();
+    fs::create_dir(root.path().join("secret")).unwrap();
+    fs::write(root.path().join("secret/key"), "hidden").unwrap();
+    symlink("target", root.path().join("base/alias-a")).unwrap();
+    symlink("target", root.path().join("base/alias-b")).unwrap();
+    symlink("../../secret", root.path().join("base/target/escape")).unwrap();
+    let backend = session(root.path());
+    let parent = backend
+        .scope(filesystem_scope("base", WorkdirToolScopePermission::Write))
+        .await
+        .unwrap();
+    let worker = WorkdirToolBroker::new(parent.tool_session());
+    let first = worker
+        .scope(filesystem_scope(
+            "alias-a",
+            WorkdirToolScopePermission::Write,
+        ))
+        .await
+        .unwrap();
+    first.write(write("ok", "allowed")).await.unwrap();
+    assert!(first.read(read("escape/key")).await.is_err());
+    assert!(
+        worker
+            .scope(filesystem_scope(
+                "alias-a/escape",
+                WorkdirToolScopePermission::Read
+            ))
+            .await
+            .is_err()
+    );
+    assert!(
+        worker
+            .scope(filesystem_scope(
+                "alias-b",
+                WorkdirToolScopePermission::Write
+            ))
+            .await
+            .is_err()
+    );
+    assert!(worker.write(write("target/blocked", "no")).await.is_err());
+    // Direct authorization through the wrapper must not bypass its existing write lease.
+    assert!(
+        worker
+            .authorize_scope_path(authorization(
+                "target/blocked",
+                WorkdirToolScopePermission::Write
+            ))
+            .await
+            .is_err()
+    );
+    first.close().await.unwrap();
+    worker
+        .write(write("alias-b/released", "yes"))
+        .await
+        .unwrap();
+}